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

Merge branch 'staging' into environment-role-creation

Browse Source
This commit is contained in:
dandds
2020-02-11 12:12:45 -05:00
parent 3f60d3494e d1ee168270
commit 10014e696d
160 changed files with 952 additions and 3915 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
tests/domain/cloud/reports/test_reports.py
View File

@@ -1,58 +1,47 @@
from atst.domain.csp.reports import MockReportingProvider
from atst.domain.csp.reports import prepare_azure_reporting_data
from tests.factories import PortfolioFactory
from decimal import Decimal
import pendulum
def test_get_environment_monthly_totals():
environment = {
"name": "Test Environment",
"spending": {
"this_month": {"JEDI_CLIN_1": 100, "JEDI_CLIN_2": 100},
"last_month": {"JEDI_CLIN_1": 200, "JEDI_CLIN_2": 200},
"total": {"JEDI_CLIN_1": 1000, "JEDI_CLIN_2": 1000},
},
}
totals = MockReportingProvider._get_environment_monthly_totals(environment)
assert totals == {
"name": "Test Environment",
"this_month": 200,
"last_month": 400,
"total": 2000,
}
class TestPrepareAzureData:
start_of_month = pendulum.today(tz="utc").start_of("month").replace(tzinfo=None)
next_month = start_of_month.add(months=1).to_atom_string()
this_month = start_of_month.to_atom_string()
last_month = start_of_month.subtract(months=1).to_atom_string()
two_months_ago = last_month = start_of_month.subtract(months=2).to_atom_string()
def test_estimated_and_invoiced(self):
rows = [
[150.0, self.two_months_ago, "", "USD"],
[100.0, self.last_month, "e0500a4qhw", "USD"],
[50.0, self.this_month, "", "USD"],
[50.0, self.next_month, "", "USD"],
]
output = prepare_azure_reporting_data(rows)
def test_get_application_monthly_totals():
portfolio = PortfolioFactory.create(
applications=[
{"name": "Test Application", "environments": [{"name": "Z"}, {"name": "A"}]}
],
)
application = {
"name": "Test Application",
"environments": [
{
"name": "Z",
"spending": {
"this_month": {"JEDI_CLIN_1": 50, "JEDI_CLIN_2": 50},
"last_month": {"JEDI_CLIN_1": 150, "JEDI_CLIN_2": 150},
"total": {"JEDI_CLIN_1": 250, "JEDI_CLIN_2": 250},
},
},
{
"name": "A",
"spending": {
"this_month": {"JEDI_CLIN_1": 100, "JEDI_CLIN_2": 100},
"last_month": {"JEDI_CLIN_1": 200, "JEDI_CLIN_2": 200},
"total": {"JEDI_CLIN_1": 1000, "JEDI_CLIN_2": 1000},
},
},
],
}
assert output.get("invoiced") == Decimal(250.0)
assert output.get("estimated") == Decimal(100.0)
totals = MockReportingProvider._get_application_monthly_totals(
portfolio, application
)
assert totals["name"] == "Test Application"
assert totals["this_month"] == 300
assert totals["last_month"] == 700
assert totals["total"] == 2500
assert [env["name"] for env in totals["environments"]] == ["A", "Z"]
def test_just_estimated(self):
rows = [
[100.0, self.this_month, "", "USD"],
]
output = prepare_azure_reporting_data(rows)
assert output.get("invoiced") == Decimal(0.0)
assert output.get("estimated") == Decimal(100.0)
def test_just_invoiced(self):
rows = [
[100.0, self.last_month, "", "USD"],
]
output = prepare_azure_reporting_data(rows)
assert output.get("invoiced") == Decimal(100.0)
assert output.get("estimated") == Decimal(0.0)
def test_no_rows(self):
output = prepare_azure_reporting_data([])
assert output.get("invoiced") == Decimal(0.0)
assert output.get("estimated") == Decimal(0.0)

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

@@ -22,9 +22,14 @@ from atst.domain.csp.cloud.models import (
BillingProfileTenantAccessCSPResult,
BillingProfileVerificationCSPPayload,
BillingProfileVerificationCSPResult,
InitialMgmtGroupCSPPayload,
InitialMgmtGroupCSPResult,
InitialMgmtGroupVerificationCSPPayload,
InitialMgmtGroupVerificationCSPResult,
CostManagementQueryCSPResult,
EnvironmentCSPPayload,
EnvironmentCSPResult,
KeyVaultCredentials,
PrincipalAdminRoleCSPPayload,
PrincipalAdminRoleCSPResult,
ProductPurchaseCSPPayload,
@@ -66,6 +71,12 @@ def mock_management_group_create(mock_azure, spec_dict):
)
def mock_management_group_get(mock_azure, spec_dict):
mock_azure.sdk.managementgroups.ManagementGroupsAPI.return_value.management_groups.get.return_value.result.return_value = (
spec_dict
)
def test_create_environment_succeeds(mock_azure: AzureCloudProvider):
environment = EnvironmentFactory.create()
mock_management_group_create(mock_azure, {"id": "Test Id"})
@@ -110,6 +121,41 @@ def test_create_application_succeeds(mock_azure: AzureCloudProvider):
assert result.id == "Test Id"
def test_create_initial_mgmt_group_succeeds(mock_azure: AzureCloudProvider):
application = ApplicationFactory.create()
mock_management_group_create(mock_azure, {"id": "Test Id"})
mock_azure = mock_get_secret(mock_azure)
payload = InitialMgmtGroupCSPPayload(
tenant_id="1234",
display_name=application.name,
management_group_name=str(uuid4()),
)
result: InitialMgmtGroupCSPResult = mock_azure.create_initial_mgmt_group(payload)
assert result.id == "Test Id"
def test_create_initial_mgmt_group_verification_succeeds(
mock_azure: AzureCloudProvider,
):
application = ApplicationFactory.create()
mock_management_group_get(mock_azure, {"id": "Test Id"})
mock_azure = mock_get_secret(mock_azure)
management_group_name = str(uuid4())
payload = InitialMgmtGroupVerificationCSPPayload(
tenant_id="1234", management_group_name=management_group_name
)
result: InitialMgmtGroupVerificationCSPResult = mock_azure.create_initial_mgmt_group_verification(
payload
)
assert result.id == "Test Id"
# assert result.name == management_group_name
def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
subscription_id = str(uuid4())
management_group_id = str(uuid4())
@@ -988,3 +1034,23 @@ def test_create_user_role_failure(mock_azure: AzureCloudProvider):
with pytest.raises(UserProvisioningException):
mock_azure.create_user_role(payload)
def test_update_tenant_creds(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider, "set_secret", wraps=mock_azure.set_secret,
) as set_secret:
set_secret.return_value = None
existing_secrets = {
"tenant_id": "mytenant",
"tenant_admin_username": "admin",
"tenant_admin_password": "foo", # pragma: allowlist secret
}
mock_azure = mock_get_secret(mock_azure, json.dumps(existing_secrets))
mock_new_secrets = KeyVaultCredentials(**MOCK_CREDS)
updated_secret = mock_azure.update_tenant_creds("mytenant", mock_new_secrets)
assert updated_secret == KeyVaultCredentials(
**{**existing_secrets, **MOCK_CREDS}
)

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

@@ -100,6 +100,26 @@ def test_KeyVaultCredentials_enforce_root_creds():
)
def test_KeyVaultCredentials_merge_credentials():
old_secret = KeyVaultCredentials(
tenant_id="foo",
tenant_admin_username="bar",
tenant_admin_password="baz", # pragma: allowlist secret
)
new_secret = KeyVaultCredentials(
tenant_id="foo", tenant_sp_client_id="bip", tenant_sp_key="bop"
)
expected_update = KeyVaultCredentials(
tenant_id="foo",
tenant_admin_username="bar",
tenant_admin_password="baz", # pragma: allowlist secret
tenant_sp_client_id="bip",
tenant_sp_key="bop",
)
assert old_secret.merge_credentials(new_secret) == expected_update
user_payload = {
"tenant_id": "123",
"display_name": "Han Solo",

6
tests/domain/test_portfolio_state_machine.py
View File

@@ -111,6 +111,8 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
FSMStates.TENANT_PRINCIPAL_CREDENTIAL_CREATED,
FSMStates.ADMIN_ROLE_DEFINITION_CREATED,
FSMStates.PRINCIPAL_ADMIN_ROLE_CREATED,
FSMStates.INITIAL_MGMT_GROUP_CREATED,
FSMStates.INITIAL_MGMT_GROUP_VERIFICATION_CREATED,
FSMStates.TENANT_ADMIN_OWNERSHIP_CREATED,
FSMStates.TENANT_PRINCIPAL_OWNERSHIP_CREATED,
]
@@ -131,10 +133,12 @@ def test_fsm_transition_start(mock_cloud_provider, portfolio: Portfolio):
"user_id": user_id,
"password": "jklfsdNCVD83nklds2#202", # pragma: allowlist secret
"domain_name": domain_name,
"display_name": "mgmt group display name",
"management_group_name": "mgmt-group-uuid",
"first_name": ppoc.first_name,
"last_name": 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": "",

35
tests/domain/test_reports.py
View File

@@ -1,8 +1,31 @@
# TODO: Implement when we get real reporting data
def test_expired_task_orders():
pass
import pytest
from atst.domain.reports import Reports
from tests.factories import PortfolioFactory
from decimal import Decimal
# TODO: Implement when we get real reporting data
def test_obligated_funds_by_JEDI_clin():
pass
@pytest.fixture(scope="function")
def portfolio():
portfolio = PortfolioFactory.create()
return portfolio
class TestGetPortfolioSpending:
csp_data = {
"tenant_id": "",
"billing_profile_properties": {
"invoice_sections": [{"invoice_section_id": "",}]
},
}
def test_with_csp_data(self, portfolio):
portfolio.csp_data = self.csp_data
data = Reports.get_portfolio_spending(portfolio)
assert data["invoiced"] == Decimal(1551.0)
assert data["estimated"] == Decimal(500.0)
def test_without_csp_data(self, portfolio):
data = Reports.get_portfolio_spending(portfolio)
assert data["invoiced"] == Decimal(0)
assert data["estimated"] == Decimal(0)

5
tests/domain/test_task_orders.py
View File

@@ -149,11 +149,12 @@ def test_task_order_sort_by_status():
]
sorted_by_status = TaskOrders.sort_by_status(initial_to_list)
assert len(sorted_by_status["Draft"]) == 3
assert len(sorted_by_status["Draft"]) == 4
assert len(sorted_by_status["Active"]) == 1
assert len(sorted_by_status["Upcoming"]) == 1
assert len(sorted_by_status["Expired"]) == 2
assert len(sorted_by_status["Unsigned"]) == 1
with pytest.raises(KeyError):
sorted_by_status["Unsigned"]
assert list(sorted_by_status.keys()) == [status.value for status in SORT_ORDERING]

16
tests/test_app.py
View File

@@ -7,6 +7,7 @@ from atst.app import (
make_crl_validator,
apply_config_from_directory,
apply_config_from_environment,
make_config,
)
@@ -67,3 +68,18 @@ def test_apply_config_from_environment_skips_unknown_settings(
monkeypatch.setenv("FLARF", "MAYO")
apply_config_from_environment(config_object)
assert "FLARF" not in config_object.options("default")
class TestMakeConfig:
def test_redis_ssl_connection(self):
config = make_config({"REDIS_TLS": True})
uri = config.get("REDIS_URI")
assert "rediss" in uri
assert "ssl_cert_reqs" in uri
def test_non_redis_ssl_connection(self):
config = make_config({"REDIS_TLS": False})
uri = config.get("REDIS_URI")
assert "rediss" not in uri
assert "redis" in uri
assert "ssl_cert_reqs" not in uri