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

merge staging. update create_create_billing_owner azure csp tests

Browse Source
This commit is contained in:
Philip Kalinsky 2020-02-11 16:57:37 -05:00
commit efe042fac3
8 changed files with 505 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

289
alembic/versions/9f2813487e00_add_billing_owner_to_state_machine.py Normal file
View File

@ -0,0 +1,289 @@
"""add billing owner to state machine
Revision ID: 9f2813487e00
Revises: 418b52c1cedf
Create Date: 2020-02-11 14:56:26.886945
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '9f2813487e00' # pragma: allowlist secret
down_revision = '418b52c1cedf' # pragma: allowlist secret
branch_labels = None
depends_on = None
def upgrade():
op.alter_column(
"portfolio_state_machines",
"state",
type_=sa.Enum(
"UNSTARTED",
"STARTING",
"STARTED",
"COMPLETED",
"FAILED",
"TENANT_CREATED",
"TENANT_IN_PROGRESS",
"TENANT_FAILED",
"BILLING_PROFILE_CREATION_CREATED",
"BILLING_PROFILE_CREATION_IN_PROGRESS",
"BILLING_PROFILE_CREATION_FAILED",
"BILLING_PROFILE_VERIFICATION_CREATED",
"BILLING_PROFILE_VERIFICATION_IN_PROGRESS",
"BILLING_PROFILE_VERIFICATION_FAILED",
"BILLING_PROFILE_TENANT_ACCESS_CREATED",
"BILLING_PROFILE_TENANT_ACCESS_IN_PROGRESS",
"BILLING_PROFILE_TENANT_ACCESS_FAILED",
"TASK_ORDER_BILLING_CREATION_CREATED",
"TASK_ORDER_BILLING_CREATION_IN_PROGRESS",
"TASK_ORDER_BILLING_CREATION_FAILED",
"TASK_ORDER_BILLING_VERIFICATION_CREATED",
"TASK_ORDER_BILLING_VERIFICATION_IN_PROGRESS",
"TASK_ORDER_BILLING_VERIFICATION_FAILED",
"BILLING_INSTRUCTION_CREATED",
"BILLING_INSTRUCTION_IN_PROGRESS",
"BILLING_INSTRUCTION_FAILED",
"PRODUCT_PURCHASE_CREATED",
"PRODUCT_PURCHASE_IN_PROGRESS",
"PRODUCT_PURCHASE_FAILED",
"PRODUCT_PURCHASE_VERIFICATION_CREATED",
"PRODUCT_PURCHASE_VERIFICATION_IN_PROGRESS",
"PRODUCT_PURCHASE_VERIFICATION_FAILED",
"TENANT_PRINCIPAL_APP_CREATED",
"TENANT_PRINCIPAL_APP_IN_PROGRESS",
"TENANT_PRINCIPAL_APP_FAILED",
"TENANT_PRINCIPAL_CREATED",
"TENANT_PRINCIPAL_IN_PROGRESS",
"TENANT_PRINCIPAL_FAILED",
"TENANT_PRINCIPAL_CREDENTIAL_CREATED",
"TENANT_PRINCIPAL_CREDENTIAL_IN_PROGRESS",
"TENANT_PRINCIPAL_CREDENTIAL_FAILED",
"ADMIN_ROLE_DEFINITION_CREATED",
"ADMIN_ROLE_DEFINITION_IN_PROGRESS",
"ADMIN_ROLE_DEFINITION_FAILED",
"PRINCIPAL_ADMIN_ROLE_CREATED",
"PRINCIPAL_ADMIN_ROLE_IN_PROGRESS",
"PRINCIPAL_ADMIN_ROLE_FAILED",
"INITIAL_MGMT_GROUP_CREATED",
"INITIAL_MGMT_GROUP_IN_PROGRESS",
"INITIAL_MGMT_GROUP_FAILED",
"INITIAL_MGMT_GROUP_VERIFICATION_CREATED",
"INITIAL_MGMT_GROUP_VERIFICATION_IN_PROGRESS",
"INITIAL_MGMT_GROUP_VERIFICATION_FAILED",
"TENANT_ADMIN_OWNERSHIP_CREATED",
"TENANT_ADMIN_OWNERSHIP_IN_PROGRESS",
"TENANT_ADMIN_OWNERSHIP_FAILED",
"TENANT_PRINCIPAL_OWNERSHIP_CREATED",
"TENANT_PRINCIPAL_OWNERSHIP_IN_PROGRESS",
"TENANT_PRINCIPAL_OWNERSHIP_FAILED",
"BILLING_OWNER_CREATED",
"BILLING_OWNER_IN_PROGRESS",
"BILLING_OWNER_FAILED",
name="fsmstates",
native_enum=False,
),
existing_type=sa.Enum(
"UNSTARTED",
"STARTING",
"STARTED",
"COMPLETED",
"FAILED",
"TENANT_CREATED",
"TENANT_IN_PROGRESS",
"TENANT_FAILED",
"BILLING_PROFILE_CREATION_CREATED",
"BILLING_PROFILE_CREATION_IN_PROGRESS",
"BILLING_PROFILE_CREATION_FAILED",
"BILLING_PROFILE_VERIFICATION_CREATED",
"BILLING_PROFILE_VERIFICATION_IN_PROGRESS",
"BILLING_PROFILE_VERIFICATION_FAILED",
"BILLING_PROFILE_TENANT_ACCESS_CREATED",
"BILLING_PROFILE_TENANT_ACCESS_IN_PROGRESS",
"BILLING_PROFILE_TENANT_ACCESS_FAILED",
"TASK_ORDER_BILLING_CREATION_CREATED",
"TASK_ORDER_BILLING_CREATION_IN_PROGRESS",
"TASK_ORDER_BILLING_CREATION_FAILED",
"TASK_ORDER_BILLING_VERIFICATION_CREATED",
"TASK_ORDER_BILLING_VERIFICATION_IN_PROGRESS",
"TASK_ORDER_BILLING_VERIFICATION_FAILED",
"BILLING_INSTRUCTION_CREATED",
"BILLING_INSTRUCTION_IN_PROGRESS",
"BILLING_INSTRUCTION_FAILED",
"PRODUCT_PURCHASE_CREATED",
"PRODUCT_PURCHASE_IN_PROGRESS",
"PRODUCT_PURCHASE_FAILED",
"PRODUCT_PURCHASE_VERIFICATION_CREATED",
"PRODUCT_PURCHASE_VERIFICATION_IN_PROGRESS",
"PRODUCT_PURCHASE_VERIFICATION_FAILED",
"TENANT_PRINCIPAL_APP_CREATED",
"TENANT_PRINCIPAL_APP_IN_PROGRESS",
"TENANT_PRINCIPAL_APP_FAILED",
"TENANT_PRINCIPAL_CREATED",
"TENANT_PRINCIPAL_IN_PROGRESS",
"TENANT_PRINCIPAL_FAILED",
"TENANT_PRINCIPAL_CREDENTIAL_CREATED",
"TENANT_PRINCIPAL_CREDENTIAL_IN_PROGRESS",
"TENANT_PRINCIPAL_CREDENTIAL_FAILED",
"ADMIN_ROLE_DEFINITION_CREATED",
"ADMIN_ROLE_DEFINITION_IN_PROGRESS",
"ADMIN_ROLE_DEFINITION_FAILED",
"PRINCIPAL_ADMIN_ROLE_CREATED",
"PRINCIPAL_ADMIN_ROLE_IN_PROGRESS",
"PRINCIPAL_ADMIN_ROLE_FAILED",
"INITIAL_MGMT_GROUP_CREATED",
"INITIAL_MGMT_GROUP_IN_PROGRESS",
"INITIAL_MGMT_GROUP_FAILED",
"INITIAL_MGMT_GROUP_VERIFICATION_CREATED",
"INITIAL_MGMT_GROUP_VERIFICATION_IN_PROGRESS",
"INITIAL_MGMT_GROUP_VERIFICATION_FAILED",
"TENANT_ADMIN_OWNERSHIP_CREATED",
"TENANT_ADMIN_OWNERSHIP_IN_PROGRESS",
"TENANT_ADMIN_OWNERSHIP_FAILED",
"TENANT_PRINCIPAL_OWNERSHIP_CREATED",
"TENANT_PRINCIPAL_OWNERSHIP_IN_PROGRESS",
"TENANT_PRINCIPAL_OWNERSHIP_FAILED",
name="fsmstates",
native_enum=False,
),
existing_nullable=False,
)
def downgrade():
op.alter_column(
"portfolio_state_machines",
"state",
type_=sa.Enum(
"UNSTARTED",
"STARTING",
"STARTED",
"COMPLETED",
"FAILED",
"TENANT_CREATED",
"TENANT_IN_PROGRESS",
"TENANT_FAILED",
"BILLING_PROFILE_CREATION_CREATED",
"BILLING_PROFILE_CREATION_IN_PROGRESS",
"BILLING_PROFILE_CREATION_FAILED",
"BILLING_PROFILE_VERIFICATION_CREATED",
"BILLING_PROFILE_VERIFICATION_IN_PROGRESS",
"BILLING_PROFILE_VERIFICATION_FAILED",
"BILLING_PROFILE_TENANT_ACCESS_CREATED",
"BILLING_PROFILE_TENANT_ACCESS_IN_PROGRESS",
"BILLING_PROFILE_TENANT_ACCESS_FAILED",
"TASK_ORDER_BILLING_CREATION_CREATED",
"TASK_ORDER_BILLING_CREATION_IN_PROGRESS",
"TASK_ORDER_BILLING_CREATION_FAILED",
"TASK_ORDER_BILLING_VERIFICATION_CREATED",
"TASK_ORDER_BILLING_VERIFICATION_IN_PROGRESS",
"TASK_ORDER_BILLING_VERIFICATION_FAILED",
"BILLING_INSTRUCTION_CREATED",
"BILLING_INSTRUCTION_IN_PROGRESS",
"BILLING_INSTRUCTION_FAILED",
"PRODUCT_PURCHASE_CREATED",
"PRODUCT_PURCHASE_IN_PROGRESS",
"PRODUCT_PURCHASE_FAILED",
"PRODUCT_PURCHASE_VERIFICATION_CREATED",
"PRODUCT_PURCHASE_VERIFICATION_IN_PROGRESS",
"PRODUCT_PURCHASE_VERIFICATION_FAILED",
"TENANT_PRINCIPAL_APP_CREATED",
"TENANT_PRINCIPAL_APP_IN_PROGRESS",
"TENANT_PRINCIPAL_APP_FAILED",
"TENANT_PRINCIPAL_CREATED",
"TENANT_PRINCIPAL_IN_PROGRESS",
"TENANT_PRINCIPAL_FAILED",
"TENANT_PRINCIPAL_CREDENTIAL_CREATED",
"TENANT_PRINCIPAL_CREDENTIAL_IN_PROGRESS",
"TENANT_PRINCIPAL_CREDENTIAL_FAILED",
"ADMIN_ROLE_DEFINITION_CREATED",
"ADMIN_ROLE_DEFINITION_IN_PROGRESS",
"ADMIN_ROLE_DEFINITION_FAILED",
"PRINCIPAL_ADMIN_ROLE_CREATED",
"PRINCIPAL_ADMIN_ROLE_IN_PROGRESS",
"PRINCIPAL_ADMIN_ROLE_FAILED",
"INITIAL_MGMT_GROUP_CREATED",
"INITIAL_MGMT_GROUP_IN_PROGRESS",
"INITIAL_MGMT_GROUP_FAILED",
"INITIAL_MGMT_GROUP_VERIFICATION_CREATED",
"INITIAL_MGMT_GROUP_VERIFICATION_IN_PROGRESS",
"INITIAL_MGMT_GROUP_VERIFICATION_FAILED",
"TENANT_ADMIN_OWNERSHIP_CREATED",
"TENANT_ADMIN_OWNERSHIP_IN_PROGRESS",
"TENANT_ADMIN_OWNERSHIP_FAILED",
"TENANT_PRINCIPAL_OWNERSHIP_CREATED",
"TENANT_PRINCIPAL_OWNERSHIP_IN_PROGRESS",
"TENANT_PRINCIPAL_OWNERSHIP_FAILED",
name="fsmstates",
native_enum=False,
),
existing_type=sa.Enum(
"UNSTARTED",
"STARTING",
"STARTED",
"COMPLETED",
"FAILED",
"TENANT_CREATED",
"TENANT_IN_PROGRESS",
"TENANT_FAILED",
"BILLING_PROFILE_CREATION_CREATED",
"BILLING_PROFILE_CREATION_IN_PROGRESS",
"BILLING_PROFILE_CREATION_FAILED",
"BILLING_PROFILE_VERIFICATION_CREATED",
"BILLING_PROFILE_VERIFICATION_IN_PROGRESS",
"BILLING_PROFILE_VERIFICATION_FAILED",
"BILLING_PROFILE_TENANT_ACCESS_CREATED",
"BILLING_PROFILE_TENANT_ACCESS_IN_PROGRESS",
"BILLING_PROFILE_TENANT_ACCESS_FAILED",
"TASK_ORDER_BILLING_CREATION_CREATED",
"TASK_ORDER_BILLING_CREATION_IN_PROGRESS",
"TASK_ORDER_BILLING_CREATION_FAILED",
"TASK_ORDER_BILLING_VERIFICATION_CREATED",
"TASK_ORDER_BILLING_VERIFICATION_IN_PROGRESS",
"TASK_ORDER_BILLING_VERIFICATION_FAILED",
"BILLING_INSTRUCTION_CREATED",
"BILLING_INSTRUCTION_IN_PROGRESS",
"BILLING_INSTRUCTION_FAILED",
"PRODUCT_PURCHASE_CREATED",
"PRODUCT_PURCHASE_IN_PROGRESS",
"PRODUCT_PURCHASE_FAILED",
"PRODUCT_PURCHASE_VERIFICATION_CREATED",
"PRODUCT_PURCHASE_VERIFICATION_IN_PROGRESS",
"PRODUCT_PURCHASE_VERIFICATION_FAILED",
"TENANT_PRINCIPAL_APP_CREATED",
"TENANT_PRINCIPAL_APP_IN_PROGRESS",
"TENANT_PRINCIPAL_APP_FAILED",
"TENANT_PRINCIPAL_CREATED",
"TENANT_PRINCIPAL_IN_PROGRESS",
"TENANT_PRINCIPAL_FAILED",
"TENANT_PRINCIPAL_CREDENTIAL_CREATED",
"TENANT_PRINCIPAL_CREDENTIAL_IN_PROGRESS",
"TENANT_PRINCIPAL_CREDENTIAL_FAILED",
"ADMIN_ROLE_DEFINITION_CREATED",
"ADMIN_ROLE_DEFINITION_IN_PROGRESS",
"ADMIN_ROLE_DEFINITION_FAILED",
"PRINCIPAL_ADMIN_ROLE_CREATED",
"PRINCIPAL_ADMIN_ROLE_IN_PROGRESS",
"PRINCIPAL_ADMIN_ROLE_FAILED",
"INITIAL_MGMT_GROUP_CREATED",
"INITIAL_MGMT_GROUP_IN_PROGRESS",
"INITIAL_MGMT_GROUP_FAILED",
"INITIAL_MGMT_GROUP_VERIFICATION_CREATED",
"INITIAL_MGMT_GROUP_VERIFICATION_IN_PROGRESS",
"INITIAL_MGMT_GROUP_VERIFICATION_FAILED",
"TENANT_ADMIN_OWNERSHIP_CREATED",
"TENANT_ADMIN_OWNERSHIP_IN_PROGRESS",
"TENANT_ADMIN_OWNERSHIP_FAILED",
"TENANT_PRINCIPAL_OWNERSHIP_CREATED",
"TENANT_PRINCIPAL_OWNERSHIP_IN_PROGRESS",
"TENANT_PRINCIPAL_OWNERSHIP_FAILED",
"BILLING_OWNER_CREATED",
"BILLING_OWNER_IN_PROGRESS",
"BILLING_OWNER_FAILED",
name="fsmstates",
native_enum=False,
),
existing_nullable=False,
)

98
atst/domain/csp/cloud/azure_cloud_provider.py
View File

@ -20,6 +20,8 @@ from .models import (
ApplicationCSPResult, ApplicationCSPResult,
BillingInstructionCSPPayload, BillingInstructionCSPPayload,
BillingInstructionCSPResult, BillingInstructionCSPResult,
BillingOwnerCSPPayload,
BillingOwnerCSPResult,
BillingProfileCreationCSPPayload, BillingProfileCreationCSPPayload,
BillingProfileCreationCSPResult, BillingProfileCreationCSPResult,
BillingProfileTenantAccessCSPPayload, BillingProfileTenantAccessCSPPayload,
@ -1060,6 +1062,96 @@ class AzureCloudProvider(CloudProviderInterface):
except self.sdk.requests.exceptions.HTTPError: except self.sdk.requests.exceptions.HTTPError:
raise UnknownServerException("azure application error creating tenant") raise UnknownServerException("azure application error creating tenant")
def create_billing_owner(self, payload: BillingOwnerCSPPayload):
graph_token = self._get_tenant_principal_token(
payload.tenant_id, resource=self.graph_resource
)
if graph_token is None:
raise AuthenticationException(
"Could not resolve graph token for tenant admin"
)
# Step 1: Create an AAD identity for the user
user_result = self._create_active_directory_user(graph_token, payload)
# Step 2: Set the recovery email
self._update_active_directory_user_email(graph_token, user_result.id, payload)
# Step 3: Find the Billing Administrator role ID
billing_admin_role_id = self._get_billing_owner_role(graph_token)
# Step 4: Assign the Billing Administrator role to the new user
self._assign_billing_owner_role(
graph_token, billing_admin_role_id, user_result.id
)
return BillingOwnerCSPResult(billing_owner_id=user_result.id)
def _assign_billing_owner_role(self, graph_token, billing_admin_role_id, user_id):
request_body = {
"roleDefinitionId": billing_admin_role_id,
"principalId": user_id,
"resourceScope": "/",
}
auth_header = {
"Authorization": f"Bearer {graph_token}",
}
url = f"{self.graph_resource}/beta/roleManagement/directory/roleAssignments"
try:
response = self.sdk.requests.post(
url, headers=auth_header, json=request_body
)
response.raise_for_status()
except self.sdk.requests.exceptions.ConnectionError:
app.logger.error(
f"Could not create tenant. Connection Error", exc_info=1,
)
raise ConnectionException("connection error creating tenant")
except self.sdk.requests.exceptions.Timeout:
app.logger.error(
f"Could not create tenant. Request timed out.", exc_info=1,
)
raise ConnectionException("timout error creating tenant")
except self.sdk.requests.exceptions.HTTPError:
raise UnknownServerException("azure application error creating tenant")
if response.ok:
return True
else:
raise UserProvisioningException("Could not assign billing admin role")
def _get_billing_owner_role(self, graph_token):
auth_header = {
"Authorization": f"Bearer {graph_token}",
}
url = f"{self.graph_resource}/v1.0/directoryRoles"
try:
response = self.sdk.requests.get(url, headers=auth_header)
response.raise_for_status()
except self.sdk.requests.exceptions.ConnectionError:
app.logger.error(
f"Could not create tenant. Connection Error", exc_info=1,
)
raise ConnectionException("connection error creating tenant")
except self.sdk.requests.exceptions.Timeout:
app.logger.error(
f"Could not create tenant. Request timed out.", exc_info=1,
)
raise ConnectionException("timout error creating tenant")
except self.sdk.requests.exceptions.HTTPError:
raise UnknownServerException("azure application error creating tenant")
if response.ok:
result = response.json()
for role in result["value"]:
if role["displayName"] == "Billing Administrator":
return role["id"]
else:
raise UserProvisioningException(
"Could not find Billing Administrator role ID; role may not be enabled."
)
def force_tenant_admin_pw_update(self, creds, tenant_owner_id): def force_tenant_admin_pw_update(self, creds, tenant_owner_id):
# use creds to update to force password recovery? # use creds to update to force password recovery?
# not sure what the endpoint/method for this is, yet # not sure what the endpoint/method for this is, yet
@ -1140,7 +1232,7 @@ class AzureCloudProvider(CloudProviderInterface):
return result return result
def _create_active_directory_user(self, graph_token, payload: UserCSPPayload): def _create_active_directory_user(self, graph_token, payload):
request_body = { request_body = {
"accountEnabled": True, "accountEnabled": True,
"displayName": payload.display_name, "displayName": payload.display_name,
@ -1181,9 +1273,7 @@ class AzureCloudProvider(CloudProviderInterface):
except self.sdk.requests.exceptions.HTTPError: except self.sdk.requests.exceptions.HTTPError:
raise UnknownServerException("azure application error creating tenant") raise UnknownServerException("azure application error creating tenant")
def _update_active_directory_user_email( def _update_active_directory_user_email(self, graph_token, user_id, payload):
self, graph_token, user_id, payload: UserCSPPayload
):
request_body = {"otherMails": [payload.email]} request_body = {"otherMails": [payload.email]}
auth_header = { auth_header = {

9
atst/domain/csp/cloud/mock_cloud_provider.py
View File

@ -19,6 +19,8 @@ from .models import (
ApplicationCSPResult, ApplicationCSPResult,
BillingInstructionCSPPayload, BillingInstructionCSPPayload,
BillingInstructionCSPResult, BillingInstructionCSPResult,
BillingOwnerCSPPayload,
BillingOwnerCSPResult,
BillingProfileCreationCSPPayload, BillingProfileCreationCSPPayload,
BillingProfileCreationCSPResult, BillingProfileCreationCSPResult,
BillingProfileTenantAccessCSPResult, BillingProfileTenantAccessCSPResult,
@ -390,6 +392,13 @@ class MockCloudProvider(CloudProviderInterface):
return PrincipalAdminRoleCSPResult(**dict(id="principal_assignment_id")) return PrincipalAdminRoleCSPResult(**dict(id="principal_assignment_id"))
def create_billing_owner(self, payload: BillingOwnerCSPPayload):
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
return BillingOwnerCSPResult(billing_owner_id="foo")
def create_or_update_user(self, auth_credentials, user_info, csp_role_id): def create_or_update_user(self, auth_credentials, user_info, csp_role_id):
self._authorize(auth_credentials) self._authorize(auth_credentials)

35
atst/domain/csp/cloud/models.py
View File

@ -521,10 +521,7 @@ class ProductPurchaseVerificationCSPResult(AliasModel):
premium_purchase_date: str premium_purchase_date: str
class UserCSPPayload(BaseCSPPayload): class UserMixin(BaseModel):
display_name: str
tenant_host_name: str
email: str
password: Optional[str] password: Optional[str]
@property @property
@ -540,6 +537,12 @@ class UserCSPPayload(BaseCSPPayload):
return password or token_urlsafe(16) return password or token_urlsafe(16)
class UserCSPPayload(BaseCSPPayload, UserMixin):
display_name: str
tenant_host_name: str
email: str
class UserCSPResult(AliasModel): class UserCSPResult(AliasModel):
id: str id: str
@ -588,3 +591,27 @@ class ReportingCSPPayload(BaseCSPPayload):
return values return values
except (KeyError, IndexError): except (KeyError, IndexError):
raise ValueError("Invoice section ID not present in payload") raise ValueError("Invoice section ID not present in payload")
class BillingOwnerCSPPayload(BaseCSPPayload, UserMixin):
"""
This class needs to consume data in the shape it's in from the
top-level portfolio CSP data, but return it in the shape
needed for user provisioning.
"""
display_name = "billing_admin"
domain_name: str
password_recovery_email_address: str
@property
def tenant_host_name(self):
return self.domain_name
@property
def email(self):
return self.password_recovery_email_address
class BillingOwnerCSPResult(AliasModel):
billing_owner_id: str

1
atst/models/mixins/state_machines.py
View File

@ -28,6 +28,7 @@ class AzureStages(Enum):
INITIAL_MGMT_GROUP_VERIFICATION = "initial management group verification" INITIAL_MGMT_GROUP_VERIFICATION = "initial management group verification"
TENANT_ADMIN_OWNERSHIP = "tenant admin ownership" TENANT_ADMIN_OWNERSHIP = "tenant admin ownership"
TENANT_PRINCIPAL_OWNERSHIP = "tenant principial ownership" TENANT_PRINCIPAL_OWNERSHIP = "tenant principial ownership"
BILLING_OWNER = "billing owner"
def _build_csp_states(csp_stages): def _build_csp_states(csp_stages):

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

@ -23,6 +23,7 @@ from atst.domain.csp.cloud.models import (
ApplicationCSPResult, ApplicationCSPResult,
BillingInstructionCSPPayload, BillingInstructionCSPPayload,
BillingInstructionCSPResult, BillingInstructionCSPResult,
BillingOwnerCSPPayload,
BillingProfileCreationCSPPayload, BillingProfileCreationCSPPayload,
BillingProfileCreationCSPResult, BillingProfileCreationCSPResult,
BillingProfileTenantAccessCSPPayload, BillingProfileTenantAccessCSPPayload,
@ -1276,6 +1277,49 @@ def test_create_user_role_failure(mock_azure: AzureCloudProvider):
mock_azure.create_user_role(payload) mock_azure.create_user_role(payload)
def test_create_billing_owner(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_tenant_principal_token",
wraps=mock_azure._get_tenant_principal_token,
) as _get_tenant_principal_token:
_get_tenant_principal_token.return_value = "token"
final_result = "1-2-3"
# create_billing_owner does: POST, PATCH, GET, POST
def make_mock_result(return_value=None):
mock_result_create = Mock()
mock_result_create.ok = True
mock_result_create.json.return_value = return_value
return mock_result_create
post_results = [make_mock_result({"id": final_result}), make_mock_result()]
mock_post = lambda *a, **k: post_results.pop(0)
# mock POST so that it pops off results in the order we want
mock_azure.sdk.requests.post = mock_post
# return value for PATCH doesn't matter much
mock_azure.sdk.requests.patch.return_value = make_mock_result()
# return value for GET needs to be a JSON object with a list of role definitions
mock_azure.sdk.requests.get.return_value = make_mock_result(
{"value": [{"displayName": "Billing Administrator", "id": "4567"}]}
)
payload = BillingOwnerCSPPayload(
tenant_id=uuid4().hex,
domain_name="rebelalliance",
password_recovery_email_address="many@bothans.org",
)
result = mock_azure.create_billing_owner(payload)
assert result.billing_owner_id == final_result
def test_update_tenant_creds(mock_azure: AzureCloudProvider): def test_update_tenant_creds(mock_azure: AzureCloudProvider):
with patch.object( with patch.object(
AzureCloudProvider, "set_secret", wraps=mock_azure.set_secret, AzureCloudProvider, "set_secret", wraps=mock_azure.set_secret,

36
tests/domain/cloud/test_models.py
View File

@ -8,6 +8,7 @@ from atst.domain.csp.cloud.models import (
ManagementGroupCSPPayload, ManagementGroupCSPPayload,
ManagementGroupCSPResponse, ManagementGroupCSPResponse,
UserCSPPayload, UserCSPPayload,
BillingOwnerCSPPayload,
) )
@ -141,3 +142,38 @@ def test_UserCSPPayload_user_principal_name():
def test_UserCSPPayload_password(): def test_UserCSPPayload_password():
payload = UserCSPPayload(**user_payload) payload = UserCSPPayload(**user_payload)
assert payload.password assert payload.password
class TestBillingOwnerCSPPayload:
user_payload = {
"tenant_id": "123",
"domain_name": "rebelalliance",
"password_recovery_email_address": "han@moseisley.cantina",
}
def test_display_name(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert payload.display_name == "billing_admin"
def test_tenant_host_name(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert payload.tenant_host_name == self.user_payload["domain_name"]
def test_mail_nickname(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert payload.mail_nickname == "billing_admin"
def test_password(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert payload.password
def test_user_principal_name(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert (
payload.user_principal_name
== f"billing_admin@rebelalliance.onmicrosoft.com"
)
def test_email(self):
payload = BillingOwnerCSPPayload(**self.user_payload)
assert payload.email == self.user_payload["password_recovery_email_address"]

1
tests/domain/test_portfolio_state_machine.py
View File

@ -217,6 +217,7 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
FSMStates.INITIAL_MGMT_GROUP_VERIFICATION_CREATED, FSMStates.INITIAL_MGMT_GROUP_VERIFICATION_CREATED,
FSMStates.TENANT_ADMIN_OWNERSHIP_CREATED, FSMStates.TENANT_ADMIN_OWNERSHIP_CREATED,
FSMStates.TENANT_PRINCIPAL_OWNERSHIP_CREATED, FSMStates.TENANT_PRINCIPAL_OWNERSHIP_CREATED,
FSMStates.BILLING_OWNER_CREATED,
] ]
if portfolio.csp_data is not None: if portfolio.csp_data is not None: