Coverage for gpc/executors/deploy_keys_executor.py: 94%

1# Third Party Libraries 

2import attr 

3 

4# Gitlab-Project-Configurator Modules 

5from gpc.change_setting import ChangePropertySetting 

6from gpc.executors.properties_updator import ChangePropertyExecutor 

7from gpc.parameters import RunMode 

8from gpc.property_manager import PropertyBean 

9 

10 

11@attr.s(eq=False) 

12class DeployKey(PropertyBean): 

13 can_push = attr.ib() # type: bool 

14 id = attr.ib() # type: int 

15 

16 def get_query(self): 

17 return {"id": self.id, "can_push": self.can_push} 

18 

19 def to_dict(self): 

20 return {"id": self.id, "can_push": self.can_push} 

21 

22 def __eq__(self, other): 

23 if not isinstance(other, DeployKey): 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 return False 

25 return self.id == other.id and self.can_push == other.can_push 

26 

27 

28class ChangeDeployKey(ChangePropertySetting): 

29 sub_properties = ["can_push"] 

30 status_to_process = ["removed", "updated", "kept", "added"] 

31 

32 

33class DeployKeysExecutor(ChangePropertyExecutor): 

34 order = 110 

35 name = "deploy_keys" 

36 

37 def _apply(self): 

38 if self.changes: 

39 for difference in self.changes[0].differences.values(): 

40 if difference["status"] == "updated": 

41 key = self.project.keys.get( 

42 difference["after"]["id"], retry_transient_errors=True 

43 ) 

44 key.can_push = difference["after"]["can_push"] 

45 key.save(retry_transient_errors=True) 

46 elif difference["status"] == "removed": 

47 self.project.keys.delete( 

48 difference["before"]["id"], retry_transient_errors=True 

49 ) 

50 elif difference["status"] == "added": 50 ↛ 39line 50 didn't jump to line 39 because the condition on line 50 was always true

51 self.project.keys.enable(difference["after"]["id"], retry_transient_errors=True) 

52 key = self.project.keys.get( 

53 difference["after"]["id"], retry_transient_errors=True 

54 ) 

55 key.can_push = difference["after"]["can_push"] 

56 key.save(retry_transient_errors=True) 

57 

58 def _update(self, mode: RunMode, members_user, members_group): 

59 if "deploy_keys" in self.rule and self.rule.deploy_keys is not None: 

60 after_keys = [] 

61 for key in self.rule.deploy_keys: 

62 after_keys.append(DeployKey(name=str(key.id), id=key.id, can_push=key.can_push)) 

63 before_keys = [] 

64 for key in self.project.keys.list(retry_transient_errors=True): # type: ignore 

65 before_keys.append(DeployKey(name=str(key.id), id=key.id, can_push=key.can_push)) 

66 self.changes.append( 

67 ChangeDeployKey( 

68 property_name="deploy_keys", 

69 before=before_keys, 

70 after=after_keys, 

71 show_diff_only=self.show_diff_only, 

72 ) 

73 )