Orchestration for creating app management groups.
This adds: - A Celery beat task for enqueuing application creation tasks - A Celery task for creating the application - Payload and Response dataclasses for creating management groups It also does some incidental cleanup.
This commit is contained in:
@@ -18,6 +18,7 @@ from atst.domain.csp.cloud import (
|
||||
TaskOrderBillingVerificationCSPResult,
|
||||
TenantCSPPayload,
|
||||
TenantCSPResult,
|
||||
ApplicationCSPPayload,
|
||||
)
|
||||
|
||||
from tests.mock_azure import mock_azure, AUTH_CREDENTIALS
|
||||
@@ -67,8 +68,8 @@ def test_create_subscription_succeeds(mock_azure: AzureCloudProvider):
|
||||
|
||||
|
||||
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
|
||||
mock_azure.sdk.managementgroups.ManagementGroupsAPI.return_value.management_groups.create_or_update.return_value.result.return_value = (
|
||||
spec_dict
|
||||
)
|
||||
|
||||
|
||||
@@ -89,7 +90,10 @@ def test_create_application_succeeds(mock_azure: AzureCloudProvider):
|
||||
|
||||
mock_management_group_create(mock_azure, {"id": "Test Id"})
|
||||
|
||||
result = mock_azure._create_application(AUTH_CREDENTIALS, application)
|
||||
payload = ApplicationCSPPayload(
|
||||
creds={}, display_name=application.name, parent_id=str(uuid4())
|
||||
)
|
||||
result = mock_azure.create_application(payload)
|
||||
|
||||
assert result.id == "Test Id"
|
||||
|
||||
@@ -150,7 +154,7 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
|
||||
**dict(
|
||||
creds=creds,
|
||||
user_id="admin",
|
||||
password="JediJan13$coot",
|
||||
password="JediJan13$coot", # pragma: allowlist secret
|
||||
domain_name="jediccpospawnedtenant2",
|
||||
first_name="Tedry",
|
||||
last_name="Tenet",
|
||||
|
72
tests/domain/cloud/test_payloads.py
Normal file
72
tests/domain/cloud/test_payloads.py
Normal file
@@ -0,0 +1,72 @@
|
||||
import pytest
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from atst.domain.csp.cloud import (
|
||||
AZURE_MGMNT_PATH,
|
||||
ManagementGroupCSPPayload,
|
||||
ManagementGroupCSPResponse,
|
||||
)
|
||||
|
||||
|
||||
def test_ManagementGroupCSPPayload_management_group_name():
|
||||
# supplies management_group_name when absent
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={}, display_name="Council of Naboo", parent_id="Galactic_Senate"
|
||||
)
|
||||
assert payload.management_group_name
|
||||
# validates management_group_name
|
||||
with pytest.raises(ValidationError):
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={},
|
||||
management_group_name="council of Naboo 1%^&",
|
||||
display_name="Council of Naboo",
|
||||
parent_id="Galactic_Senate",
|
||||
)
|
||||
# shortens management_group_name to fit
|
||||
name = "council_of_naboo"
|
||||
for _ in range(90):
|
||||
name = f"{name}1"
|
||||
|
||||
assert len(name) > 90
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={},
|
||||
management_group_name=name,
|
||||
display_name="Council of Naboo",
|
||||
parent_id="Galactic_Senate",
|
||||
)
|
||||
assert len(payload.management_group_name) == 90
|
||||
|
||||
|
||||
def test_ManagementGroupCSPPayload_display_name():
|
||||
# shortens display_name to fit
|
||||
name = "Council of Naboo"
|
||||
for _ in range(90):
|
||||
name = f"{name}1"
|
||||
assert len(name) > 90
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={}, display_name=name, parent_id="Galactic_Senate"
|
||||
)
|
||||
assert len(payload.display_name) == 90
|
||||
|
||||
|
||||
def test_ManagementGroupCSPPayload_parent_id():
|
||||
full_path = f"{AZURE_MGMNT_PATH}Galactic_Senate"
|
||||
# adds full path
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={}, display_name="Council of Naboo", parent_id="Galactic_Senate"
|
||||
)
|
||||
assert payload.parent_id == full_path
|
||||
# keeps full path
|
||||
payload = ManagementGroupCSPPayload(
|
||||
creds={}, display_name="Council of Naboo", parent_id=full_path
|
||||
)
|
||||
assert payload.parent_id == full_path
|
||||
|
||||
|
||||
def test_ManagementGroupCSPResponse_id():
|
||||
full_id = "/path/to/naboo-123"
|
||||
response = ManagementGroupCSPResponse(
|
||||
**{"id": "/path/to/naboo-123", "other": "stuff"}
|
||||
)
|
||||
assert response.id == full_id
|
@@ -141,7 +141,6 @@ def test_fsm_transition_start(portfolio: Portfolio):
|
||||
config = {"billing_account_name": "billing_account_name"}
|
||||
|
||||
for expected_state in expected_states:
|
||||
print(expected_state)
|
||||
collected_data = dict(
|
||||
list(csp_data.items()) + list(portfolio_data.items()) + list(config.items())
|
||||
)
|
||||
|
Reference in New Issue
Block a user