Coverage for backend/django/migration_helper/check_existing_migrations_unchanged.py: 83%

1"""Fail CI if a migration that predates the latest reachable tag has changed.""" 

2 

3from __future__ import annotations 

4 

5import shlex 

6import subprocess 

7import sys 

8from pathlib import Path, PurePosixPath 

9 

10 

11def git(repo_root: Path, *args: str) -> str: 

12 result = subprocess.run( 

13 ["git", *args], 

14 cwd=repo_root, 

15 capture_output=True, 

16 text=True, 

17 ) 

18 

19 if result.returncode != 0: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true

20 error_output = result.stderr.strip() or result.stdout.strip() or "git command failed" 

21 raise RuntimeError(f"{shlex.join(['git', *args])} failed: {error_output}") 

22 

23 return result.stdout.strip() 

24 

25 

26def is_migration_file(path: str) -> bool: 

27 posix_path = PurePosixPath(path) 

28 return ( 

29 posix_path.suffix == ".py" 

30 and posix_path.name != "__init__.py" 

31 and len(posix_path.parts) >= 2 

32 and posix_path.parts[-2] == "migrations" 

33 ) 

34 

35 

36def get_latest_reachable_tag(repo_root: Path) -> str | None: 

37 tags = git( 

38 repo_root, 

39 "for-each-ref", 

40 "--merged=HEAD", 

41 "--sort=-creatordate", 

42 "--format=%(refname:strip=2)", 

43 "refs/tags", 

44 ).splitlines() 

45 django_tags = [tag for tag in tags if "django" in tag.lower()] 

46 return django_tags[0] if django_tags else None 

47 

48 

49def get_migration_files_at_ref(repo_root: Path, ref: str) -> set[str]: 

50 return { 

51 path 

52 for path in git(repo_root, "ls-tree", "-r", "--name-only", ref).splitlines() 

53 if is_migration_file(path) 

54 } 

55 

56 

57def get_changed_paths(repo_root: Path, ref: str) -> list[str]: 

58 changed_migration_paths: set[str] = set() 

59 tagged_migration_paths = get_migration_files_at_ref(repo_root, ref) 

60 

61 for line in git(repo_root, "diff", "--name-status", "--find-renames", f"{ref}..HEAD").splitlines(): 

62 if not line: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 continue 

64 

65 status, *paths = line.split("\t") 

66 relevant_paths = [path for path in paths if path in tagged_migration_paths] 

67 

68 if status.startswith("A") or not relevant_paths: 

69 continue 

70 

71 if status.startswith(("R", "C")) and len(paths) == 2: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 changed_migration_paths.add(f"{paths[0]} -> {paths[1]}") 

73 continue 

74 

75 changed_migration_paths.update(relevant_paths) 

76 

77 return sorted(changed_migration_paths) 

78 

79 

80def main() -> int: 

81 repo_root = Path(git(Path.cwd(), "rev-parse", "--show-toplevel")) 

82 latest_tag = get_latest_reachable_tag(repo_root) 

83 

84 if latest_tag is None: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true

85 print("No reachable Django git tags were found, skipping deployed migration immutability check.") 

86 return 0 

87 

88 changed_migration_paths = get_changed_paths(repo_root, latest_tag) 

89 

90 if not changed_migration_paths: 

91 print(f"No migration files from {latest_tag} were modified.") 

92 return 0 

93 

94 print(f"Migration files that already existed at {latest_tag} were modified:") 

95 for path in changed_migration_paths: 

96 print(f" - {path}") 

97 

98 return 1 

99 

100 

101if __name__ == "__main__": 101 ↛ exitline 101 didn't exit the module because the condition on line 101 was always true

102 sys.exit(main())