Coverage for .nox/test-3-12/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py: 12%

1"""Azure Devops provider using azure-cli to manage it.""" 

2from io import StringIO 

3import json 

4from typing import List 

5 

6try: 

7 from azure.cli.core import get_default_cli 

8except ImportError: 

9 raise ImportError('Azure Devops Provider requires installing extra dependencies, use pip install nskit[azure_devops]') 

10 

11from pydantic import HttpUrl 

12from pydantic_settings import SettingsConfigDict 

13 

14from nskit.vcs.providers.abstract import RepoClient, VCSProviderSettings 

15 

16# Want to use MS interactive Auth as default, but can't get it working, instead, using cli invoke 

17 

18 

19class AzureDevOpsSettings(VCSProviderSettings): 

20 """Azure DevOps settings.""" 

21 model_config = SettingsConfigDict(env_prefix='AZURE_DEVOPS_', env_file='.env') 

22 url: HttpUrl = "https://dev.azure.com" 

23 organisation: str 

24 project: str 

25 

26 @property 

27 def organisation_url(self): 

28 """Get the organistion Url.""" 

29 return f'{self.url}/{self.organisation}' 

30 

31 @property 

32 def project_url(self): 

33 """Get the project url.""" 

34 return f'{self.organisation_url}/{self.project}' 

35 

36 @property 

37 def repo_client(self) -> 'AzureDevOpsRepoClient': 

38 """Get the instantiated repo client.""" 

39 return AzureDevOpsRepoClient(self) 

40 

41 

42class AzureDevOpsRepoClient(RepoClient): 

43 """Client for managing Azure DevOps repos using azure-cli.""" 

44 

45 def __init__(self, config: AzureDevOpsSettings): 

46 """Initialise the client.""" 

47 self._cli = get_default_cli() 

48 self._config = config 

49 

50 def _invoke(self, command, out_file=None): 

51 return self._cli.invoke(command, out_file=out_file) 

52 

53 def check_exists(self, repo_name: str) -> bool: 

54 """Check if the repo exists in the project.""" 

55 output = StringIO() 

56 return not self._invoke(['repos', 

57 'show', 

58 '--organization', 

59 self._config.organisation_url, 

60 '--project', 

61 self._config.project, 

62 '-r', 

63 repo_name], 

64 out_file=output) 

65 

66 def create(self, repo_name: str): 

67 """Create the repo in the project.""" 

68 output = StringIO() 

69 return self._invoke(['repos', 

70 'create', 

71 '--organization', 

72 self._config.organisation_url, 

73 '--project', 

74 self._config.project, 

75 '--name', 

76 repo_name], 

77 out_file=output) 

78 

79 def delete(self, repo_name: str): 

80 """Delete the repo if it exists in the project.""" 

81 # We need to get the ID 

82 show_output = StringIO() 

83 result = self._invoke(['repos', 

84 'show', 

85 '--organization', 

86 self._config.organisation_url, 

87 '--project', 

88 self._config.project, 

89 '-r', 

90 repo_name], 

91 out_file=show_output) 

92 if not result: 

93 # Exists 

94 repo_info = json.loads(show_output.getvalue()) 

95 repo_id = repo_info['id'] 

96 output = StringIO() 

97 return self._invoke(['repos', 

98 'delete', 

99 '--organization', 

100 self._config.organisation_url, 

101 '--project', 

102 self._config.project, 

103 '--id', 

104 repo_id], 

105 out_file=output) 

106 

107 def get_remote_url(self, repo_name: str) -> HttpUrl: 

108 """Get the remote url for the repo.""" 

109 output = StringIO() 

110 result = self._invoke(['repos', 

111 'show', 

112 '--organization', 

113 self._config.organisation_url, 

114 '--project', 

115 self._config.project, 

116 '-r', 

117 repo_name], 

118 out_file=output) 

119 if not result: 

120 # Exists 

121 repo_info = json.loads(output.getvalue()) 

122 return repo_info['remoteUrl'] 

123 

124 def list(self) -> List[str]: 

125 """List the repos in the project.""" 

126 output = StringIO() 

127 result = self._invoke(['repos', 

128 'list', 

129 '--organization', 

130 self._config.organisation_url, 

131 '--project', 

132 self._config.project], 

133 out_file=output) 

134 if not result: 

135 # Exists 

136 repo_list = [u['name'] for u in json.loads(output.getvalue())] 

137 return repo_list