Merge branch 'staging' into azure-config-values
This commit is contained in:
@@ -1,6 +1,8 @@
|
||||
import json
|
||||
from unittest.mock import Mock, patch
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from tests.factories import ApplicationFactory, EnvironmentFactory
|
||||
from tests.mock_azure import AUTH_CREDENTIALS, mock_azure
|
||||
|
||||
@@ -8,6 +10,8 @@ from atst.domain.csp.cloud import AzureCloudProvider
|
||||
from atst.domain.csp.cloud.models import (
|
||||
AdminRoleDefinitionCSPPayload,
|
||||
AdminRoleDefinitionCSPResult,
|
||||
ApplicationCSPPayload,
|
||||
ApplicationCSPResult,
|
||||
BaseCSPPayload,
|
||||
BillingInstructionCSPPayload,
|
||||
BillingInstructionCSPResult,
|
||||
@@ -73,8 +77,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
|
||||
)
|
||||
|
||||
|
||||
@@ -90,12 +94,30 @@ def test_create_environment_succeeds(mock_azure: AzureCloudProvider):
|
||||
assert result.id == "Test Id"
|
||||
|
||||
|
||||
# mock the get_secret so it returns a JSON string
|
||||
MOCK_CREDS = {
|
||||
"tenant_id": str(uuid4()),
|
||||
"tenant_sp_client_id": str(uuid4()),
|
||||
"tenant_sp_key": "1234",
|
||||
}
|
||||
|
||||
|
||||
def mock_get_secret(azure, func):
|
||||
azure.get_secret = func
|
||||
|
||||
return azure
|
||||
|
||||
|
||||
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)
|
||||
mock_azure = mock_get_secret(mock_azure, lambda *a, **k: json.dumps(MOCK_CREDS))
|
||||
|
||||
payload = ApplicationCSPPayload(
|
||||
tenant_id="1234", display_name=application.name, parent_id=str(uuid4())
|
||||
)
|
||||
result = mock_azure.create_application(payload)
|
||||
|
||||
assert result.id == "Test Id"
|
||||
|
||||
|
99
tests/domain/cloud/test_models.py
Normal file
99
tests/domain/cloud/test_models.py
Normal file
@@ -0,0 +1,99 @@
|
||||
import pytest
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from atst.domain.csp.cloud.models import (
|
||||
AZURE_MGMNT_PATH,
|
||||
KeyVaultCredentials,
|
||||
ManagementGroupCSPPayload,
|
||||
ManagementGroupCSPResponse,
|
||||
)
|
||||
|
||||
|
||||
def test_ManagementGroupCSPPayload_management_group_name():
|
||||
# supplies management_group_name when absent
|
||||
payload = ManagementGroupCSPPayload(
|
||||
tenant_id="any-old-id",
|
||||
display_name="Council of Naboo",
|
||||
parent_id="Galactic_Senate",
|
||||
)
|
||||
assert payload.management_group_name
|
||||
# validates management_group_name
|
||||
with pytest.raises(ValidationError):
|
||||
payload = ManagementGroupCSPPayload(
|
||||
tenant_id="any-old-id",
|
||||
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".ljust(95, "1")
|
||||
|
||||
assert len(name) > 90
|
||||
payload = ManagementGroupCSPPayload(
|
||||
tenant_id="any-old-id",
|
||||
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".ljust(95, "1")
|
||||
assert len(name) > 90
|
||||
payload = ManagementGroupCSPPayload(
|
||||
tenant_id="any-old-id", 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(
|
||||
tenant_id="any-old-id",
|
||||
display_name="Council of Naboo",
|
||||
parent_id="Galactic_Senate",
|
||||
)
|
||||
assert payload.parent_id == full_path
|
||||
# keeps full path
|
||||
payload = ManagementGroupCSPPayload(
|
||||
tenant_id="any-old-id", 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
|
||||
|
||||
|
||||
def test_KeyVaultCredentials_enforce_admin_creds():
|
||||
with pytest.raises(ValidationError):
|
||||
KeyVaultCredentials(tenant_id="an id", tenant_admin_username="C3PO")
|
||||
assert KeyVaultCredentials(
|
||||
tenant_id="an id",
|
||||
tenant_admin_username="C3PO",
|
||||
tenant_admin_password="beep boop",
|
||||
)
|
||||
|
||||
|
||||
def test_KeyVaultCredentials_enforce_sp_creds():
|
||||
with pytest.raises(ValidationError):
|
||||
KeyVaultCredentials(tenant_id="an id", tenant_sp_client_id="C3PO")
|
||||
assert KeyVaultCredentials(
|
||||
tenant_id="an id", tenant_sp_client_id="C3PO", tenant_sp_key="beep boop"
|
||||
)
|
||||
|
||||
|
||||
def test_KeyVaultCredentials_enforce_root_creds():
|
||||
with pytest.raises(ValidationError):
|
||||
KeyVaultCredentials(root_tenant_id="an id", root_sp_client_id="C3PO")
|
||||
assert KeyVaultCredentials(
|
||||
root_tenant_id="an id", root_sp_client_id="C3PO", root_sp_key="beep boop"
|
||||
)
|
@@ -1,3 +1,4 @@
|
||||
from datetime import datetime, timedelta
|
||||
import pytest
|
||||
from uuid import uuid4
|
||||
|
||||
@@ -196,3 +197,20 @@ def test_update_does_not_duplicate_names_within_portfolio():
|
||||
|
||||
with pytest.raises(AlreadyExistsError):
|
||||
Applications.update(dupe_application, {"name": name})
|
||||
|
||||
|
||||
def test_get_applications_pending_creation():
|
||||
now = datetime.now()
|
||||
later = now + timedelta(minutes=30)
|
||||
|
||||
portfolio1 = PortfolioFactory.create(state="COMPLETED")
|
||||
app_ready = ApplicationFactory.create(portfolio=portfolio1)
|
||||
|
||||
app_done = ApplicationFactory.create(portfolio=portfolio1, cloud_id="123456")
|
||||
|
||||
portfolio2 = PortfolioFactory.create(state="UNSTARTED")
|
||||
app_not_ready = ApplicationFactory.create(portfolio=portfolio2)
|
||||
|
||||
uuids = Applications.get_applications_pending_creation()
|
||||
|
||||
assert [app_ready.id] == uuids
|
||||
|
@@ -71,7 +71,7 @@ def test_update_adds_clins():
|
||||
|
||||
def test_update_does_not_duplicate_clins():
|
||||
task_order = TaskOrderFactory.create(
|
||||
number="3453453456", create_clins=[{"number": "123"}, {"number": "456"}]
|
||||
number="3453453456123", create_clins=[{"number": "123"}, {"number": "456"}]
|
||||
)
|
||||
clins = [
|
||||
{
|
||||
@@ -93,7 +93,7 @@ def test_update_does_not_duplicate_clins():
|
||||
]
|
||||
task_order = TaskOrders.update(
|
||||
task_order_id=task_order.id,
|
||||
number="0000000000",
|
||||
number="0000000000000",
|
||||
clins=clins,
|
||||
pdf={"filename": "sample.pdf", "object_name": "1234567"},
|
||||
)
|
||||
@@ -170,3 +170,11 @@ def test_update_enforces_unique_number():
|
||||
dupe_task_order = TaskOrderFactory.create()
|
||||
with pytest.raises(AlreadyExistsError):
|
||||
TaskOrders.update(dupe_task_order.id, task_order.number, [], None)
|
||||
|
||||
|
||||
def test_allows_alphanumeric_number():
|
||||
portfolio = PortfolioFactory.create()
|
||||
valid_to_numbers = ["1234567890123", "ABC1234567890"]
|
||||
|
||||
for number in valid_to_numbers:
|
||||
assert TaskOrders.create(portfolio.id, number, [], None)
|
||||
|
Reference in New Issue
Block a user