Merge branch 'staging' into to-index-page-redesign_part-3

tests/domain/cloud/test_azure_csp.py

@@ -1,27 +1,81 @@
 import pytest
+from unittest.mock import Mock
 
 from uuid import uuid4
 
 from atst.domain.csp.cloud import AzureCloudProvider
 
 from tests.mock_azure import mock_azure, AUTH_CREDENTIALS
-from tests.factories import EnvironmentFactory
+from tests.factories import EnvironmentFactory, ApplicationFactory
 
 
-def test_create_environment_succeeds(mock_azure: AzureCloudProvider):
+# TODO: Directly test create subscription, provide all args √
+# TODO: Test create environment (create management group with parent)
+# TODO: Test create application (create manageemnt group with parent)
+# Create reusable mock for mocking the management group calls for multiple services
+#
+
+
+def test_create_subscription_succeeds(mock_azure: AzureCloudProvider):
     environment = EnvironmentFactory.create()
 
     subscription_id = str(uuid4())
 
+    credentials = mock_azure._get_credential_obj(AUTH_CREDENTIALS)
+    display_name = "Test Subscription"
+    billing_profile_id = str(uuid4())
+    sku_id = str(uuid4())
+    management_group_id = (
+        environment.cloud_id  # environment.csp_details.management_group_id?
+    )
+    billing_account_name = (
+        "?"  # environment.application.portfilio.csp_details.billing_account.name?
+    )
+    invoice_section_name = "?"  # environment.name? or something specific to billing?
+
     mock_azure.sdk.subscription.SubscriptionClient.return_value.subscription_factory.create_subscription.return_value.result.return_value.subscription_link = (
         f"subscriptions/{subscription_id}"
     )
 
+    result = mock_azure._create_subscription(
+        credentials,
+        display_name,
+        billing_profile_id,
+        sku_id,
+        management_group_id,
+        billing_account_name,
+        invoice_section_name,
+    )
+
+    assert result == subscription_id
+
+
+def mock_management_group_create(mock_azure, spec_dict):
+    mock_azure.sdk.managementgroups.ManagementGroupsAPI.return_value.management_groups.create_or_update.return_value.result.return_value = Mock(
+        **spec_dict
+    )
+
+
+def test_create_environment_succeeds(mock_azure: AzureCloudProvider):
+    environment = EnvironmentFactory.create()
+
+    mock_management_group_create(mock_azure, {"id": "Test Id"})
+
     result = mock_azure.create_environment(
         AUTH_CREDENTIALS, environment.creator, environment
     )
 
-    assert result == subscription_id
+    assert result.id == "Test Id"
+
+
+def test_create_application_succeeds(mock_azure: AzureCloudProvider):
+    application = ApplicationFactory.create()
+
+    mock_management_group_create(mock_azure, {"id": "Test Id"})
+
+    result = mock_azure._create_application(AUTH_CREDENTIALS, application)
+
+    assert result.id == "Test Id"
 
 
 def test_create_atat_admin_user_succeeds(mock_azure: AzureCloudProvider):

tests/mock_azure.py

@@ -10,9 +10,9 @@ AZURE_CONFIG = {
 }
 
 AUTH_CREDENTIALS = {
-    "CLIENT_ID": AZURE_CONFIG["AZURE_CLIENT_ID"],
-    "SECRET_KEY": AZURE_CONFIG["AZURE_SECRET_KEY"],
-    "TENANT_ID": AZURE_CONFIG["AZURE_TENANT_ID"],
+    "client_id": AZURE_CONFIG["AZURE_CLIENT_ID"],
+    "secret_key": AZURE_CONFIG["AZURE_SECRET_KEY"],
+    "tenant_id": AZURE_CONFIG["AZURE_TENANT_ID"],
 }
 
 
@@ -28,6 +28,12 @@ def mock_authorization():
     return Mock(spec=authorization)
 
 
+def mock_managementgroups():
+    from azure.mgmt import managementgroups
+
+    return Mock(spec=managementgroups)
+
+
 def mock_graphrbac():
     import azure.graphrbac as graphrbac
 
@@ -46,6 +52,7 @@ class MockAzureSDK(object):
 
         self.subscription = mock_subscription()
         self.authorization = mock_authorization()
+        self.managementgroups = mock_managementgroups()
         self.graphrbac = mock_graphrbac()
         self.credentials = mock_credentials()
         # may change to a JEDI cloud

tests/routes/applications/test_settings.py

@@ -12,6 +12,7 @@ from atst.domain.application_roles import ApplicationRoles
 from atst.domain.environment_roles import EnvironmentRoles
 from atst.domain.invitations import ApplicationInvitations
 from atst.domain.common import Paginator
+from atst.domain.csp.cloud import GeneralCSPException
 from atst.domain.permission_sets import PermissionSets
 from atst.models.application_role import Status as ApplicationRoleStatus
 from atst.models.environment_role import CSPRole, EnvironmentRole
@@ -748,3 +749,41 @@ def test_handle_update_member(set_g):
     assert len(application.roles) == 1
     assert len(app_role.environment_roles) == 1
     assert app_role.environment_roles[0].environment == env
+
+
+def test_handle_update_member_with_error(set_g, monkeypatch, mock_logger):
+    exception = "An error occurred."
+
+    def _raise_csp_exception(*args, **kwargs):
+        raise GeneralCSPException(exception)
+
+    monkeypatch.setattr(
+        "atst.domain.environments.Environments.update_env_role", _raise_csp_exception
+    )
+
+    user = UserFactory.create()
+    application = ApplicationFactory.create(
+        environments=[{"name": "Naboo"}, {"name": "Endor"}]
+    )
+    (env, env_1) = application.environments
+    app_role = ApplicationRoleFactory(application=application)
+    set_g("current_user", application.portfolio.owner)
+    set_g("portfolio", application.portfolio)
+    set_g("application", application)
+
+    form_data = ImmutableMultiDict(
+        {
+            "environment_roles-0-environment_id": env.id,
+            "environment_roles-0-role": "Basic Access",
+            "environment_roles-0-environment_name": env.name,
+            "environment_roles-1-environment_id": env_1.id,
+            "environment_roles-1-role": NO_ACCESS,
+            "environment_roles-1-environment_name": env_1.name,
+            "perms_env_mgmt": True,
+            "perms_team_mgmt": True,
+            "perms_del_env": True,
+        }
+    )
+    handle_update_member(application.id, app_role.id, form_data)
+
+    assert mock_logger.messages[-1] == exception

tests/utils.py

@@ -40,6 +40,9 @@ class FakeLogger:
     def error(self, msg, *args, **kwargs):
         self._log("error", msg, *args, **kwargs)
 
+    def exception(self, msg, *args, **kwargs):
+        self._log("exception", msg, *args, **kwargs)
+
     def _log(self, _lvl, msg, *args, **kwargs):
         self.messages.append(msg)
         if "extra" in kwargs: