Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/sources/git_policy_source.py: 29%

1from typing import Optional 

2 

3from git import Repo 

4from opal_common.git_utils.branch_tracker import BranchTracker 

5from opal_common.git_utils.exceptions import GitFailed 

6from opal_common.git_utils.repo_cloner import RepoCloner 

7from opal_common.http_utils import redact_url 

8from opal_common.logger import logger 

9from opal_common.sources.base_policy_source import BasePolicySource 

10 

11 

12class GitPolicySource(BasePolicySource): 

13 """Watches a git repository for changes and can trigger callbacks when 

14 detecting new commits on the tracked branch. 

15 

16 Checking for changes is done following a git pull from a tracked 

17 remote. The pull can be either triggered by a method (i.e: you can 

18 call it from a webhook) or can be triggered periodically by a polling 

19 task. 

20 

21 Args: 

22 remote_source_url(str): the base address to request the policy from 

23 local_clone_path(str): path for the local git to manage policies 

24 branch_name(str): name of remote branch in git to pull, default to master 

25 ssh_key (str, optional): private ssh key used to gain access to the cloned repo 

26 polling_interval(int): how many seconds need to wait between polling 

27 request_timeout(int): how many seconds need to wait until timeout 

28 """ 

29 

30 def __init__( 

31 self, 

32 remote_source_url: str, 

33 local_clone_path: str, 

34 branch_name: str = "master", 

35 ssh_key: Optional[str] = None, 

36 polling_interval: int = 0, 

37 request_timeout: int = 0, 

38 ): 

39 super().__init__( 

40 remote_source_url=remote_source_url, 

41 local_clone_path=local_clone_path, 

42 polling_interval=polling_interval, 

43 ) 

44 self._ssh_key = ssh_key 

45 

46 self._cloner = RepoCloner( 

47 remote_source_url, 

48 local_clone_path, 

49 branch_name=branch_name, 

50 ssh_key=self._ssh_key, 

51 clone_timeout=request_timeout, 

52 ) 

53 self._branch_name = branch_name 

54 self._tracker = None 

55 

56 async def get_initial_policy_state_from_remote(self): 

57 """Init remote data to local repo.""" 

58 try: 

59 try: 

60 # Check if path already contains valid repo 

61 repo = Repo(self._cloner.path) 

62 except: 

63 # If it doesn't - clone it 

64 result = await self._cloner.clone() 

65 repo = result.repo 

66 else: 

67 # If it does - validate remote url is correct and checkout required branch 

68 remote_urls = list(repo.remote().urls) 

69 if not self._cloner.url in remote_urls: 

70 # Don't bother with remove and reclone because this case shouldn't happen on reasobable usage 

71 raise GitFailed( 

72 RuntimeError( 

73 f"Existing repo has wrong remote url: {[redact_url(u) for u in remote_urls]}" 

74 ) 

75 ) 

76 else: 

77 logger.info( 

78 "SKIPPED cloning policy repo, found existing repo at '{path}' with remotes: {remote_urls}", 

79 path=self._cloner.path, 

80 remote_urls=[redact_url(u) for u in remote_urls], 

81 ) 

82 except GitFailed as e: 

83 await self._on_git_failed(e) 

84 return 

85 

86 self._tracker = BranchTracker( 

87 repo=repo, branch_name=self._branch_name, ssh_key=self._ssh_key 

88 ) 

89 

90 async def check_for_changes(self): 

91 """Calling this method will trigger a git pull from the tracked remote. 

92 

93 If after the pull the watcher detects new commits, it will call 

94 the callbacks registered with _on_new_policy(). 

95 """ 

96 logger.info( 

97 "Pulling changes from remote: '{remote}'", 

98 remote=self._tracker.tracked_remote.name, 

99 ) 

100 has_changes, prev, latest = self._tracker.pull() 

101 if not has_changes: 

102 logger.info("No new commits: HEAD is at '{head}'", head=latest.hexsha) 

103 else: 

104 logger.info( 

105 "Found new commits: old HEAD was '{prev_head}', new HEAD is '{new_head}'", 

106 prev_head=prev.hexsha, 

107 new_head=latest.hexsha, 

108 ) 

109 await self._on_new_policy(old=prev, new=latest)