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

resolve merge conflict with staging

Browse Source
This commit is contained in:
Philip Kalinsky
2020-01-30 15:57:06 -05:00
parent de992eeed8 b444378b0f
commit 7e4340e7e4
62 changed files with 1339 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,15 +1,18 @@
import pytest
import json
from uuid import uuid4
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
from atst.domain.csp.cloud import AzureCloudProvider
from atst.domain.csp.cloud.models import (
AdminRoleDefinitionCSPPayload,
AdminRoleDefinitionCSPResult,
ApplicationCSPPayload,
ApplicationCSPResult,
BaseCSPPayload,
BillingInstructionCSPPayload,
BillingInstructionCSPResult,
BillingProfileCreationCSPPayload,
@@ -26,15 +29,20 @@ from atst.domain.csp.cloud.models import (
TaskOrderBillingCreationCSPResult,
TaskOrderBillingVerificationCSPPayload,
TaskOrderBillingVerificationCSPResult,
TenantAdminOwnershipCSPPayload,
TenantAdminOwnershipCSPResult,
TenantCSPPayload,
TenantCSPResult,
TenantPrincipalAppCSPPayload,
TenantPrincipalAppCSPResult,
TenantPrincipalCredentialCSPPayload,
TenantPrincipalCredentialCSPResult,
TenantPrincipalCSPPayload,
TenantPrincipalCSPResult,
TenantPrincipalOwnershipCSPPayload,
TenantPrincipalOwnershipCSPResult,
)
creds = {
"home_tenant_id": "tenant_id",
"client_id": "client_id",
"secret_key": "secret_key",
}
BILLING_ACCOUNT_NAME = "52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31"
@@ -98,8 +106,10 @@ MOCK_CREDS = {
}
def mock_get_secret(azure, func):
azure.get_secret = func
def mock_get_secret(azure, val=None):
if val is None:
val = json.dumps(MOCK_CREDS)
azure.get_secret = lambda *a, **k: val
return azure
@@ -107,12 +117,12 @@ def mock_get_secret(azure, func):
def test_create_application_succeeds(mock_azure: AzureCloudProvider):
application = ApplicationFactory.create()
mock_management_group_create(mock_azure, {"id": "Test Id"})
mock_azure = mock_get_secret(mock_azure, lambda *a, **k: json.dumps(MOCK_CREDS))
mock_azure = mock_get_secret(mock_azure)
payload = ApplicationCSPPayload(
tenant_id="1234", display_name=application.name, parent_id=str(uuid4())
)
result = mock_azure.create_application(payload)
assert result.id == "Test Id"
@@ -158,10 +168,6 @@ def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
def test_create_tenant(mock_azure: AzureCloudProvider):
mock_azure.sdk.adal.AuthenticationContext.return_value.context.acquire_token_with_client_credentials.return_value = {
"accessToken": "TOKEN"
}
mock_result = Mock()
mock_result.json.return_value = {
"objectId": "0a5f4926-e3ee-4f47-a6e3-8b0a30a40e3d",
@@ -172,7 +178,6 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
mock_azure.sdk.requests.post.return_value = mock_result
payload = TenantCSPPayload(
**dict(
creds=creds,
user_id="admin",
password="JediJan13$coot", # pragma: allowlist secret
domain_name="jediccpospawnedtenant2",
@@ -182,6 +187,7 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
password_recovery_email_address="thomas@promptworks.com",
)
)
mock_azure = mock_get_secret(mock_azure)
result = mock_azure.create_tenant(payload)
body: TenantCSPResult = result.get("body")
assert body.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
@@ -209,7 +215,6 @@ def test_create_billing_profile_creation(mock_azure: AzureCloudProvider):
country="US",
postal_code="19109",
),
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
billing_profile_display_name="Test Billing Profile",
billing_account_name=BILLING_ACCOUNT_NAME,
@@ -260,7 +265,7 @@ def test_validate_billing_profile_creation(mock_azure: AzureCloudProvider):
payload = BillingProfileVerificationCSPPayload(
**dict(
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
billing_profile_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
)
)
@@ -299,7 +304,6 @@ def test_create_billing_profile_tenant_access(mock_azure: AzureCloudProvider):
payload = BillingProfileTenantAccessCSPPayload(
**dict(
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
user_object_id="0a5f4926-e3ee-4f47-a6e3-8b0a30a40e3d",
billing_account_name="7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31",
@@ -331,7 +335,7 @@ def test_create_task_order_billing_creation(mock_azure: AzureCloudProvider):
payload = TaskOrderBillingCreationCSPPayload(
**dict(
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
billing_account_name="7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31",
billing_profile_name="KQWI-W2SU-BG7-TGB",
)
@@ -392,7 +396,7 @@ def test_create_task_order_billing_verification(mock_azure):
payload = TaskOrderBillingVerificationCSPPayload(
**dict(
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
task_order_billing_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
)
)
@@ -427,7 +431,7 @@ def test_create_billing_instruction(mock_azure: AzureCloudProvider):
payload = BillingInstructionCSPPayload(
**dict(
creds=creds,
tenant_id="60ff9d34-82bf-4f21-b565-308ef0533435",
initial_clin_amount=1000.00,
initial_clin_start_date="2020/1/1",
initial_clin_end_date="2020/3/1",
@@ -441,7 +445,6 @@ def test_create_billing_instruction(mock_azure: AzureCloudProvider):
body: BillingInstructionCSPResult = result.get("body")
assert body.reported_clin_name == "TO1:CLIN001"
def test_create_product_purchase(mock_azure: AzureCloudProvider):
mock_azure.sdk.adal.AuthenticationContext.return_value.context.acquire_token_with_client_credentials.return_value = {
"accessToken": "TOKEN"
@@ -458,7 +461,7 @@ def test_create_product_purchase(mock_azure: AzureCloudProvider):
payload = ProductPurchaseCSPPayload(
**dict(
creds=creds,
tenant_id="6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
type="AADPremium",
sku="AADP1",
productProperties={
@@ -519,7 +522,7 @@ def test_create_product_purchase_verification(mock_azure):
payload = ProductPurchaseVerificationCSPPayload(
**dict(
creds=creds,
tenant_id="6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
product_purchase_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
)
)
@@ -527,3 +530,172 @@ def test_create_product_purchase_verification(mock_azure):
result = mock_azure.create_product_purchase_verification(payload)
body: ProductPurchaseVerificationCSPResult = result.get("body")
assert body.premium_purchase_date == "2020-01-30T18:57:05.981Z"
def test_create_tenant_principal_app(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"appId": "appId", "id": "id"}
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
payload = TenantPrincipalAppCSPPayload(
**{"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4"}
)
result: TenantPrincipalAppCSPResult = mock_azure.create_tenant_principal_app(
payload
)
assert result.principal_app_id == "appId"
def test_create_tenant_principal(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"id": "principal_id"}
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
payload = TenantPrincipalCSPPayload(
**{
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"principal_app_id": "appId",
}
)
result: TenantPrincipalCSPResult = mock_azure.create_tenant_principal(payload)
assert result.principal_id == "principal_id"
def test_create_tenant_principal_credential(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"secretText": "new secret key"}
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
payload = TenantPrincipalCredentialCSPPayload(
**{
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"principal_app_id": "appId",
"principal_app_object_id": "appObjId",
}
)
result: TenantPrincipalCredentialCSPResult = mock_azure.create_tenant_principal_credential(
payload
)
assert result.principal_creds_established == True
def test_create_admin_role_definition(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {
"value": [
{"id": "wrongid", "displayName": "Wrong Role"},
{"id": "id", "displayName": "Company Administrator"},
]
}
mock_azure.sdk.requests.get.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
payload = AdminRoleDefinitionCSPPayload(
**{"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4"}
)
result: AdminRoleDefinitionCSPResult = mock_azure.create_admin_role_definition(
payload
)
assert result.admin_role_def_id == "id"
def test_create_tenant_admin_ownership(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"id": "id"}
mock_azure.sdk.requests.put.return_value = mock_result
payload = TenantAdminOwnershipCSPPayload(
**{
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"user_object_id": "971efe4d-1e80-4e39-b3b9-4e5c63ad446d",
}
)
result: TenantAdminOwnershipCSPResult = mock_azure.create_tenant_admin_ownership(
payload
)
assert result.admin_owner_assignment_id == "id"
def test_create_tenant_principal_ownership(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_elevated_management_token",
wraps=mock_azure._get_elevated_management_token,
) as get_elevated_management_token:
get_elevated_management_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"id": "id"}
mock_azure.sdk.requests.put.return_value = mock_result
payload = TenantPrincipalOwnershipCSPPayload(
**{
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"principal_id": "971efe4d-1e80-4e39-b3b9-4e5c63ad446d",
}
)
result: TenantPrincipalOwnershipCSPResult = mock_azure.create_tenant_principal_ownership(
payload
)
assert result.principal_owner_assignment_id == "id"

11
tests/domain/test_portfolio_state_machine.py
View File

@@ -106,10 +106,15 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
FSMStates.BILLING_INSTRUCTION_CREATED,
FSMStates.PRODUCT_PURCHASE_CREATED,
FSMStates.PRODUCT_PURCHASE_VERIFICATION_CREATED,
FSMStates.TENANT_PRINCIPAL_APP_CREATED,
FSMStates.TENANT_PRINCIPAL_CREATED,
FSMStates.TENANT_PRINCIPAL_CREDENTIAL_CREATED,
FSMStates.ADMIN_ROLE_DEFINITION_CREATED,
FSMStates.PRINCIPAL_ADMIN_ROLE_CREATED,
FSMStates.TENANT_ADMIN_OWNERSHIP_CREATED,
FSMStates.TENANT_PRINCIPAL_OWNERSHIP_CREATED,
]
# Should source all creds for portfolio? might be easier to manage than per-step specific ones
creds = {"username": "mock-cloud", "password": "shh"} # pragma: allowlist secret
if portfolio.csp_data is not None:
csp_data = portfolio.csp_data
else:
@@ -152,7 +157,7 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
collected_data = dict(
list(csp_data.items()) + list(portfolio_data.items()) + list(config.items())
)
sm.trigger_next_transition(creds=creds, csp_data=collected_data)
sm.trigger_next_transition(csp_data=collected_data)
assert sm.state == expected_state
if portfolio.csp_data is not None:
csp_data = portfolio.csp_data

20
tests/mock_azure.py
View File

@@ -9,6 +9,9 @@ AZURE_CONFIG = {
"AZURE_TENANT_ID": "MOCK",
"AZURE_POLICY_LOCATION": "policies",
"AZURE_VAULT_URL": "http://vault",
"POWERSHELL_CLIENT_ID": "MOCK",
"AZURE_OWNER_ROLE_DEF_ID": "MOCK",
"AZURE_GRAPH_RESOURCE": "MOCK",
}
AUTH_CREDENTIALS = {
@@ -48,6 +51,12 @@ def mock_credentials():
return Mock(spec=credentials)
def mock_identity():
import azure.identity as identity
return Mock(spec=identity)
def mock_policy():
from azure.mgmt.resource import policy
@@ -72,15 +81,14 @@ def mock_secrets():
return Mock(spec=secrets)
def mock_identity():
import azure.identity as identity
def mock_cloud_details():
from msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD
return Mock(spec=identity)
return AZURE_PUBLIC_CLOUD
class MockAzureSDK(object):
def __init__(self):
from msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD
self.subscription = mock_subscription()
self.authorization = mock_authorization()
@@ -89,11 +97,11 @@ class MockAzureSDK(object):
self.managementgroups = mock_managementgroups()
self.graphrbac = mock_graphrbac()
self.credentials = mock_credentials()
self.identity = mock_identity()
self.policy = mock_policy()
self.secrets = mock_secrets()
self.requests = mock_requests()
# may change to a JEDI cloud
self.cloud = AZURE_PUBLIC_CLOUD
self.cloud = mock_cloud_details()
self.identity = mock_identity()

2
tests/routes/applications/test_init.py
View File

@@ -15,7 +15,7 @@ def test_environment_access_with_env_role(client, user_session):
url_for("applications.access_environment", environment_id=environment.id)
)
assert response.status_code == 302
assert "csp-environment-access" in response.location
assert "portal.azure.com" in response.location
def test_environment_access_with_no_role(client, user_session):

58
tests/routes/task_orders/test_new.py
View File

@@ -458,3 +458,61 @@ def test_task_order_form_shows_errors(client, user_session, task_order):
body = response.data.decode()
assert "There were some errors" in body
assert "Not a valid decimal" in body
def test_update_and_render_next_handles_previous_valid_data(
client, user_session, task_order
):
user_session(task_order.portfolio.owner)
form_data = {"number": "0000000000000"}
original_number = task_order.number
response = client.post(
url_for(
"task_orders.submit_form_step_two_add_number",
task_order_id=task_order.id,
previous=True,
),
data=form_data,
)
assert response.status_code == 302
assert task_order.number == "0000000000000"
assert task_order.number != original_number
def test_update_and_render_next_handles_previous_invalid_data(
client, user_session, task_order
):
clin_list = [
{
"jedi_clin_type": "JEDI_CLIN_1",
"number": "12312",
"start_date": "01/01/2020",
"end_date": "01/01/2021",
"obligated_amount": "5000",
"total_amount": "10000",
},
]
TaskOrders.create_clins(task_order.id, clin_list)
assert len(task_order.clins) == 2
user_session(task_order.portfolio.owner)
form_data = {
"clins-0-jedi_clin_type": "JEDI_CLIN_1",
"clins-0-number": "12312",
"clins-0-start_date": "01/01/2020",
"clins-0-end_date": "01/01/2021",
"clins-0-obligated_amount": "5000",
"clins-0-total_amount": "10000",
"clins-1-jedi_clin_type": "JEDI_CLIN_1",
"clins-1-number": "1212",
}
response = client.post(
url_for(
"task_orders.submit_form_step_three_add_clins",
task_order_id=task_order.id,
previous=True,
),
data=form_data,
)
assert len(task_order.clins) == 2

31
tests/routes/test_root.py
View File

@@ -1,7 +1,36 @@
from tests.factories import UserFactory
from tests.factories import UserFactory, PortfolioFactory
from atst.routes import match_url_pattern
def test_root_redirects_if_user_is_logged_in(client, user_session):
user_session(UserFactory.create())
response = client.get("/", follow_redirects=False)
assert "home" in response.location
def test_match_url_pattern(client):
assert not match_url_pattern(None)
assert match_url_pattern("/home") == "/home"
portfolio = PortfolioFactory()
# matches a URL with an argument
assert (
match_url_pattern(f"/portfolios/{portfolio.id}") # /portfolios/<portfolio_id>
== f"/portfolios/{portfolio.id}"
)
# matches a url with a query string
assert (
match_url_pattern(f"/portfolios/{portfolio.id}?foo=bar")
== f"/portfolios/{portfolio.id}?foo=bar"
)
# matches a URL only with a valid method
assert not match_url_pattern(f"/portfolios/{portfolio.id}/edit")
assert (
match_url_pattern(f"/portfolios/{portfolio.id}/edit", method="POST")
== f"/portfolios/{portfolio.id}/edit"
)
# returns None for URL that doesn't match a view function
assert not match_url_pattern("/pwned")
assert not match_url_pattern("http://www.hackersite.com/pwned")

2
tests/routes/test_users.py
View File

@@ -28,7 +28,7 @@ def test_user_can_update_profile(user_session, client):
def test_user_is_redirected_when_updating_profile(user_session, client):
user = UserFactory.create()
user_session(user)
next_url = "/requests"
next_url = "/home"
user_data = user.to_dictionary()
user_data["date_latest_training"] = user_data["date_latest_training"].strftime(

2
tests/test_access.py
View File

@@ -19,9 +19,7 @@ from atst.app import make_config, make_app
_NO_ACCESS_CHECK_REQUIRED = _NO_LOGIN_REQUIRED + [
"applications.accept_invitation", # available to all users; access control is built into invitation logic
"atst.catch_all", # available to all users
"atst.csp_environment_access", # internal redirect
"atst.home", # available to all users
"atst.jedi_csp_calculator", # internal redirect
"dev.messages", # dev tool
"dev.test_email", # dev tool
"portfolios.accept_invitation", # available to all users; access control is built into invitation logic