Coverage for .nox/test-3-9/lib/python3.9/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 

12 

13from nskit.common.configuration import SettingsConfigDict 

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', dotenv_extra='ignore') 

22 

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

24 organisation: str 

25 project: str 

26 

27 @property 

28 def organisation_url(self): 

29 """Get the organistion Url.""" 

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

31 

32 @property 

33 def project_url(self): 

34 """Get the project url.""" 

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

36 

37 @property 

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

39 """Get the instantiated repo client.""" 

40 return AzureDevOpsRepoClient(self) 

41 

42 

43class AzureDevOpsRepoClient(RepoClient): 

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

45 

46 def __init__(self, config: AzureDevOpsSettings): 

47 """Initialise the client.""" 

48 self._cli = get_default_cli() 

49 self._config = config 

50 

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

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

53 

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

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

56 output = StringIO() 

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

58 'show', 

59 '--organization', 

60 self._config.organisation_url, 

61 '--project', 

62 self._config.project, 

63 '-r', 

64 repo_name], 

65 out_file=output) 

66 

67 def create(self, repo_name: str): 

68 """Create the repo in the project.""" 

69 output = StringIO() 

70 return self._invoke(['repos', 

71 'create', 

72 '--organization', 

73 self._config.organisation_url, 

74 '--project', 

75 self._config.project, 

76 '--name', 

77 repo_name], 

78 out_file=output) 

79 

80 def delete(self, repo_name: str): 

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

82 # We need to get the ID 

83 show_output = StringIO() 

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

85 'show', 

86 '--organization', 

87 self._config.organisation_url, 

88 '--project', 

89 self._config.project, 

90 '-r', 

91 repo_name], 

92 out_file=show_output) 

93 if not result: 

94 # Exists 

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

96 repo_id = repo_info['id'] 

97 output = StringIO() 

98 return self._invoke(['repos', 

99 'delete', 

100 '--organization', 

101 self._config.organisation_url, 

102 '--project', 

103 self._config.project, 

104 '--id', 

105 repo_id], 

106 out_file=output) 

107 

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

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

110 output = StringIO() 

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

112 'show', 

113 '--organization', 

114 self._config.organisation_url, 

115 '--project', 

116 self._config.project, 

117 '-r', 

118 repo_name], 

119 out_file=output) 

120 if not result: 

121 # Exists 

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

123 return repo_info['remoteUrl'] 

124 

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

126 """List the repos in the project.""" 

127 output = StringIO() 

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

129 'list', 

130 '--organization', 

131 self._config.organisation_url, 

132 '--project', 

133 self._config.project], 

134 out_file=output) 

135 if not result: 

136 # Exists 

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

138 return repo_list