pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

azure csp unit tests WIP

Browse Source
This commit is contained in:
Philip Kalinsky
2020-02-11 14:11:11 -05:00
parent 50ab045353
commit 441e81ba8d
10 changed files with 355 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
tests/domain/cloud/test_azure_csp.py
View File

@@ -1,13 +1,22 @@
import json
from unittest.mock import Mock, patch
from uuid import uuid4
import pendulum
import pydantic
import pytest
#import requests
from secrets import token_urlsafe
from tests.factories import ApplicationFactory, EnvironmentFactory
from tests.mock_azure import AUTH_CREDENTIALS, mock_azure
from atst.domain.csp.cloud.exceptions import (
AuthenticationException,
UserProvisioningException,
ConnectionException,
UnknownServerException,
SecretException,
)
from atst.domain.csp.cloud import AzureCloudProvider
from atst.domain.csp.cloud.models import (
AdminRoleDefinitionCSPPayload,
@@ -58,6 +67,7 @@ from atst.domain.csp.cloud.models import (
BILLING_ACCOUNT_NAME = "52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31"
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 = (
spec_dict
@@ -93,7 +103,6 @@ def mock_get_secret(azure, val=None):
return azure
def test_create_application_succeeds(mock_azure: AzureCloudProvider):
application = ApplicationFactory.create()
mock_management_group_create(mock_azure, {"id": "Test Id"})
@@ -132,8 +141,8 @@ def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
parameters=mock_policy_definition,
)
def test_create_tenant(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.json.return_value = {
"objectId": "0a5f4926-e3ee-4f47-a6e3-8b0a30a40e3d",
@@ -141,7 +150,13 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
"userId": "1153801116406515559",
}
mock_result.status_code = 200
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure.sdk.requests.post.side_effect = [
mock_azure.sdk.requests.exceptions.ConnectionError,
mock_result
]
payload = TenantCSPPayload(
**dict(
user_id="admin",
@@ -154,10 +169,63 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
)
)
mock_azure = mock_get_secret(mock_azure)
with pytest.raises(ConnectionException):
mock_azure.create_tenant(payload)
result: TenantCSPResult = mock_azure.create_tenant(payload)
assert result.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
def test_create_tenant_req_mock(mock_azure: AzureCloudProvider, requests_mock):
url = f"{mock_azure.sdk.cloud.endpoints.resource_manager}providers/Microsoft.SignUp/createTenant"
requests_mock.register_uri(
"POST", "?".join([url, "api-version=2020-01-01-preview"]),
json={
"objectId": "0a5f4926-e3ee-4f47-a6e3-8b0a30a40e3d",
"tenantId": "60ff9d34-82bf-4f21-b565-308ef0533435",
"userId": "1153801116406515559",
},
)
mock_azure = mock_get_secret(mock_azure)
payload = TenantCSPPayload(
**dict(
user_id="admin",
password="JediJan13$coot", # pragma: allowlist secret
domain_name="jediccpospawnedtenant2",
first_name="Tedry",
last_name="Tenet",
country_code="US",
password_recovery_email_address="thomas@promptworks.com",
)
)
result: TenantCSPResult = mock_azure.create_tenant(payload)
assert result.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
def test_create_tenant_req_mock_fail(mock_azure: AzureCloudProvider, requests_mock):
url = f"{mock_azure.sdk.cloud.endpoints.resource_manager}providers/Microsoft.SignUp/createTenant"
requests_mock.register_uri(
"POST", "?".join([url, "api-version=2020-01-01-preview"]),
exc=requests.exceptions.ConnectionError,
)
mock_azure = mock_get_secret(mock_azure)
payload = TenantCSPPayload(
**dict(
user_id="admin",
password="JediJan13$coot", # pragma: allowlist secret
domain_name="jediccpospawnedtenant2",
first_name="Tedry",
last_name="Tenet",
country_code="US",
password_recovery_email_address="thomas@promptworks.com",
)
)
with pytest.raises(ConnectionException):
assert mock_azure.create_tenant(payload)
# def test_create_tenant_fails(mock_azure: AzureCloudProvider):
# mock_result = Mock()
# mock_result.status_code = 200

47
tests/domain/cloud/test_mock_csp.py
View File

@@ -1,7 +1,14 @@
import pytest
from atst.domain.csp import MockCloudProvider
from atst.domain.csp.cloud.models import EnvironmentCSPPayload, EnvironmentCSPResult
from atst.domain.csp.cloud.models import (
EnvironmentCSPPayload,
EnvironmentCSPResult,
TenantCSPPayload,
TenantCSPResult,
BillingProfileCreationCSPPayload,
BillingProfileCreationCSPResult,
)
from tests.factories import EnvironmentFactory, EnvironmentRoleFactory, UserFactory
@@ -28,6 +35,44 @@ def test_create_environment(mock_csp: MockCloudProvider):
assert isinstance(result, EnvironmentCSPResult)
def test_create_tenant(mock_csp: MockCloudProvider):
payload = TenantCSPPayload(
**dict(
user_id="admin",
password="JediJan13$coot", # pragma: allowlist secret
domain_name="jediccpospawnedtenant2",
first_name="Tedry",
last_name="Tenet",
country_code="US",
password_recovery_email_address="thomas@promptworks.com",
)
)
result = mock_csp.create_tenant(payload)
assert isinstance(result, TenantCSPResult)
def test_create_billing_profile_creation(mock_csp: MockCloudProvider):
payload = BillingProfileCreationCSPPayload(
**dict(
address=dict(
address_line_1="123 S Broad Street, Suite 2400",
company_name="Promptworks",
city="Philadelphia",
region="PA",
country="US",
postal_code="19109",
),
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
billing_profile_display_name="Test Billing Profile",
billing_account_name="123123",
)
)
result = mock_csp.create_billing_profile_creation(payload)
assert isinstance(result, BillingProfileCreationCSPResult)
def test_create_or_update_user(mock_csp: MockCloudProvider):
env_role = EnvironmentRoleFactory.create()
csp_user_id = mock_csp.create_or_update_user(CREDENTIALS, env_role, "csp_role_id")

76
tests/domain/test_portfolio_state_machine.py
View File

@@ -1,6 +1,9 @@
import pytest
import pydantic
import re
from unittest import mock
from enum import Enum
from tests.factories import (
PortfolioStateMachineFactory,
@@ -8,13 +11,29 @@ from tests.factories import (
)
from atst.models import FSMStates, PortfolioStateMachine, TaskOrder
from atst.models.mixins.state_machines import AzureStages, StageStates, compose_state
from atst.domain.csp.cloud.models import BillingProfileCreationCSPPayload
from atst.models.mixins.state_machines import (
AzureStages,
StageStates,
compose_state,
_build_transitions,
_build_csp_states,
)
from atst.models.portfolio import Portfolio
from atst.models.portfolio_state_machine import get_stage_csp_class
from atst.models.portfolio_state_machine import (
get_stage_csp_class,
_stage_to_classname,
_stage_state_to_stage_name,
StateMachineMisconfiguredError,
)
# TODO: Write failure case tests
class AzureStagesTest(Enum):
TENANT = "tenant"
@pytest.fixture(scope="function")
def portfolio():
# TODO: setup clin/to as active/funded/ready
@@ -37,13 +56,34 @@ def test_state_machine_trigger_next_transition(portfolio):
assert sm.current_state == FSMStates.STARTED
def test_state_machine_compose_state(portfolio):
PortfolioStateMachineFactory.create(portfolio=portfolio)
def test_state_machine_compose_state():
assert (
compose_state(AzureStages.TENANT, StageStates.CREATED)
== FSMStates.TENANT_CREATED
)
def test_stage_to_classname():
assert (_stage_to_classname(AzureStages.BILLING_PROFILE_CREATION.name) == "BillingProfileCreation")
def test_get_stage_csp_class():
csp_class = get_stage_csp_class(list(AzureStages)[0].name.lower(), "payload")
assert isinstance(csp_class, pydantic.main.ModelMetaclass)
def test_get_stage_csp_class_import_fail():
with pytest.raises(StateMachineMisconfiguredError):
csp_class = get_stage_csp_class("doesnotexist", "payload")
def test_build_transitions():
states, transitions = _build_transitions(AzureStagesTest)
assert [s.get('name').name for s in states] == ['TENANT_CREATED', 'TENANT_IN_PROGRESS', 'TENANT_FAILED']
assert [s.get('tags') for s in states] == [['TENANT', 'CREATED'], ['TENANT', 'IN_PROGRESS'], ['TENANT', 'FAILED']]
assert [t.get('trigger') for t in transitions] == ['create_tenant', 'finish_tenant', 'fail_tenant', 'resume_progress_tenant']
def test_build_csp_states():
states = _build_csp_states(AzureStagesTest)
assert list(states) == ['UNSTARTED', 'STARTING', 'STARTED', 'COMPLETED', 'FAILED', 'TENANT_CREATED', 'TENANT_IN_PROGRESS', 'TENANT_FAILED']
def test_state_machine_valid_data_classes_for_stages(portfolio):
PortfolioStateMachineFactory.create(portfolio=portfolio)
@@ -52,6 +92,25 @@ def test_state_machine_valid_data_classes_for_stages(portfolio):
assert get_stage_csp_class(stage.name.lower(), "result") is not None
def test_attach_machine(portfolio):
sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
sm.machine = None
sm.attach_machine(stages=AzureStagesTest)
assert list(sm.machine.events) == ['init', 'start', 'reset', 'fail', 'create_tenant', 'finish_tenant', 'fail_tenant', 'resume_progress_tenant']
def test_fail_stage(portfolio):
sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
sm.state = FSMStates.TENANT_IN_PROGRESS
sm.fail_stage('tenant')
assert sm.state == FSMStates.TENANT_FAILED
#import ipdb;ipdb.set_trace()
def test_stage_state_to_stage_name():
#sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
stage = _stage_state_to_stage_name(FSMStates.TENANT_IN_PROGRESS, StageStates.IN_PROGRESS)
assert stage == "tenant"
def test_state_machine_initialization(portfolio):
sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
@@ -121,7 +180,7 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
csp_data = {}
ppoc = portfolio.owner
user_id = f"{ppoc.first_name[0]}{ppoc.last_name}".lower()
user_id = "johndoe" # f"{ppoc.first_name[0]}{ppoc.last_name}".lower()
domain_name = re.sub("[^0-9a-zA-Z]+", "", portfolio.name).lower()
initial_task_order: TaskOrder = portfolio.task_orders[0]
@@ -131,10 +190,10 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
"user_id": user_id,
"password": "jklfsdNCVD83nklds2#202", # pragma: allowlist secret
"domain_name": domain_name,
"first_name": ppoc.first_name,
"last_name": ppoc.last_name,
"first_name": "123", # ppoc.first_name,
"last_name": "123", # ppoc.last_name,
"country_code": "US",
"password_recovery_email_address": ppoc.email,
"password_recovery_email_address": "email@example.com", # ppoc.email,
"address": { # TODO: TBD if we're sourcing this from data or config
"company_name": "",
"address_line_1": "",
@@ -163,3 +222,4 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
csp_data = portfolio.csp_data
else:
csp_data = {}

2
tests/mock_azure.py
View File

@@ -79,7 +79,7 @@ def mock_adal():
def mock_requests():
import requests
return Mock(spec=requests)
return Mock(wraps=requests)
def mock_secrets():