refactor(azure): get locations with self session (#5751)

This commit is contained in:
Pedro Martín
2024-11-13 19:00:18 +01:00
committed by GitHub
parent a419b4b898
commit 06b41cf8e6
3 changed files with 26 additions and 24 deletions
+23 -21
View File
@@ -198,7 +198,7 @@ class AzureProvider(Provider):
)
# TODO: should we keep this here or within the identity?
self._locations = self.get_locations(self.session)
self._locations = self.get_locations()
# Audit Config
if config_content:
@@ -942,33 +942,35 @@ class AzureProvider(Provider):
return identity
def get_locations(self, credentials) -> dict[str, list[str]]:
def get_locations(self) -> dict[str, list[str]]:
"""
Retrieves the locations available for each subscription using the provided credentials.
Args:
credentials: The credentials object used to authenticate the request.
Returns:
A dictionary containing the locations available for each subscription. The dictionary
has subscription display names as keys and lists of location names as values.
Examples:
>>> provider = AzureProvider(...)
>>> provider.get_locations()
{
'Subscription 1': ['eastus', 'eastus2', 'westus', 'westus2'],
'Subscription 2': ['eastus', 'eastus2', 'westus', 'westus2']
}
"""
locations = None
if credentials:
locations = {}
token = credentials.get_token("https://management.azure.com/.default").token
for display_name, subscription_id in self._identity.subscriptions.items():
locations.update({display_name: []})
url = f"https://management.azure.com/subscriptions/{subscription_id}/locations?api-version=2022-12-01"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
for location in data["value"]:
locations[display_name].append(location["name"])
credentials = self.session
subscription_client = SubscriptionClient(credentials)
locations = {}
for display_name, subscription_id in self._identity.subscriptions.items():
locations[display_name] = []
# List locations for each subscription
for location in subscription_client.subscriptions.list_locations(
subscription_id
):
locations[display_name].append(location.name)
return locations
@staticmethod
@@ -9,7 +9,7 @@ class network_watcher_enabled(Check):
report = Check_Report_Azure(self.metadata())
report.subscription = subscription
report.resource_name = "Network Watcher"
report.location = "Global"
report.location = "global"
report.resource_id = f"/subscriptions/{network_client.subscriptions[subscription]}/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/NetworkWatcher_*"
missing_locations = set(network_client.locations[subscription]) - set(
@@ -78,7 +78,7 @@ class Test_network_watcher_enabled:
assert result[0].subscription == AZURE_SUBSCRIPTION_NAME
assert result[0].resource_name == network_watcher_name
assert result[0].resource_id == network_watcher_id
assert result[0].location == "Global"
assert result[0].location == "global"
def test_network_valid_network_watchers(self):
network_client = mock.MagicMock
@@ -124,4 +124,4 @@ class Test_network_watcher_enabled:
assert result[0].subscription == AZURE_SUBSCRIPTION_NAME
assert result[0].resource_name == network_watcher_name
assert result[0].resource_id == network_watcher_id
assert result[0].location == "Global"
assert result[0].location == "global"