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

resolve conflict with staging

Browse Source
This commit is contained in:
Philip Kalinsky
2020-02-05 12:21:00 -05:00
parent b52d6c89ac c1055e3e78
commit 38e92e427b
56 changed files with 1341 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,6 +5,8 @@ from uuid import uuid4
import pytest
from tests.factories import ApplicationFactory, EnvironmentFactory
from tests.mock_azure import AUTH_CREDENTIALS, mock_azure
import pendulum
import pydantic
from atst.domain.csp.cloud import AzureCloudProvider
from atst.domain.csp.cloud.models import (
@@ -24,10 +26,12 @@ from atst.domain.csp.cloud.models import (
ManagementGroupCSPResponse,
ManagementGroupGetCSPPayload,
ManagementGroupGetCSPResponse,
CostManagementQueryCSPResult,
ProductPurchaseCSPPayload,
ProductPurchaseCSPResult,
ProductPurchaseVerificationCSPPayload,
ProductPurchaseVerificationCSPResult,
ReportingCSPPayload,
SubscriptionCreationCSPPayload,
SubscriptionCreationCSPResult,
SubscriptionVerificationCSPPayload,
@@ -765,3 +769,77 @@ def test_create_subscription_verification(mock_azure: AzureCloudProvider):
payload
)
assert result.subscription_id == "60fbbb72-0516-4253-ab18-c92432ba3230"
def test_get_reporting_data(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.json.return_value = {
"eTag": None,
"id": "providers/Microsoft.Billing/billingAccounts/52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31/billingProfiles/XQDJ-6LB4-BG7-TGB/invoiceSections/P73M-XC7J-PJA-TGB/providers/Microsoft.CostManagement/query/e82d0cda-2ffb-4476-a98a-425c83c216f9",
"location": None,
"name": "e82d0cda-2ffb-4476-a98a-425c83c216f9",
"properties": {
"columns": [
{"name": "PreTaxCost", "type": "Number"},
{"name": "UsageDate", "type": "Number"},
{"name": "InvoiceId", "type": "String"},
{"name": "Currency", "type": "String"},
],
"nextLink": None,
"rows": [],
},
"sku": None,
"type": "Microsoft.CostManagement/query",
}
mock_result.ok = True
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
# Subset of a profile's CSP data that we care about for reporting
csp_data = {
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"billing_profile_properties": {
"invoice_sections": [
{
"invoice_section_id": "providers/Microsoft.Billing/billingAccounts/52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31/billingProfiles/XQDJ-6LB4-BG7-TGB/invoiceSections/P73M-XC7J-PJA-TGB",
}
],
},
}
data: CostManagementQueryCSPResult = mock_azure.get_reporting_data(
ReportingCSPPayload(
from_date=pendulum.now().subtract(years=1).add(days=1).format("YYYY-MM-DD"),
to_date=pendulum.now().format("YYYY-MM-DD"),
**csp_data,
)
)
assert isinstance(data, CostManagementQueryCSPResult)
assert data.name == "e82d0cda-2ffb-4476-a98a-425c83c216f9"
assert len(data.properties.columns) == 4
def test_get_reporting_data_malformed_payload(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.ok = True
mock_azure.sdk.requests.post.return_value = mock_result
mock_azure = mock_get_secret(mock_azure)
# Malformed csp_data payloads that should throw pydantic validation errors
index_error = {
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"billing_profile_properties": {"invoice_sections": [],},
}
key_error = {
"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"billing_profile_properties": {"invoice_sections": [{}],},
}
for malformed_payload in [key_error, index_error]:
with pytest.raises(pydantic.ValidationError):
assert mock_azure.get_reporting_data(
ReportingCSPPayload(
from_date="foo", to_date="bar", **malformed_payload,
)
)

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

@@ -7,6 +7,7 @@ from atst.domain.csp.cloud.models import (
KeyVaultCredentials,
ManagementGroupCSPPayload,
ManagementGroupCSPResponse,
UserCSPPayload,
)
@@ -97,3 +98,26 @@ def test_KeyVaultCredentials_enforce_root_creds():
assert KeyVaultCredentials(
root_tenant_id="an id", root_sp_client_id="C3PO", root_sp_key="beep boop"
)
user_payload = {
"tenant_id": "123",
"display_name": "Han Solo",
"tenant_host_name": "rebelalliance",
"email": "han@moseisley.cantina",
}
def test_UserCSPPayload_mail_nickname():
payload = UserCSPPayload(**user_payload)
assert payload.mail_nickname == f"han.solo"
def test_UserCSPPayload_user_principal_name():
payload = UserCSPPayload(**user_payload)
assert payload.user_principal_name == f"han.solo@rebelalliance.onmicrosoft.com"
def test_UserCSPPayload_password():
payload = UserCSPPayload(**user_payload)
assert payload.password

76
tests/domain/test_application_roles.py
View File

@@ -86,3 +86,79 @@ def test_disable(session):
session.refresh(environment_role)
assert member_role.status == ApplicationRoleStatus.DISABLED
assert environment_role.deleted
def test_get_pending_creation():
# ready Applications belonging to the same Portfolio
portfolio_one = PortfolioFactory.create()
ready_app = ApplicationFactory.create(cloud_id="123", portfolio=portfolio_one)
ready_app2 = ApplicationFactory.create(cloud_id="321", portfolio=portfolio_one)
# ready Application belonging to a new Portfolio
ready_app3 = ApplicationFactory.create(cloud_id="567")
unready_app = ApplicationFactory.create()
# two distinct Users
user_one = UserFactory.create()
user_two = UserFactory.create()
# Two ApplicationRoles belonging to the same User and
# different Applications. These should sort together because
# they are all under the same portfolio (portfolio_one).
role_one = ApplicationRoleFactory.create(
user=user_one, application=ready_app, status=ApplicationRoleStatus.ACTIVE
)
role_two = ApplicationRoleFactory.create(
user=user_one, application=ready_app2, status=ApplicationRoleStatus.ACTIVE
)
# An ApplicationRole belonging to a different User. This will
# be included but sort separately because it belongs to a
# different user.
role_three = ApplicationRoleFactory.create(
user=user_two, application=ready_app, status=ApplicationRoleStatus.ACTIVE
)
# An ApplicationRole belonging to one of the existing users
# but under a different portfolio. It will sort separately.
role_four = ApplicationRoleFactory.create(
user=user_one, application=ready_app3, status=ApplicationRoleStatus.ACTIVE
)
# This ApplicationRole will not be in the results because its
# application is not ready (implicitly, its cloud_id is not
# set.)
ApplicationRoleFactory.create(
user=UserFactory.create(),
application=unready_app,
status=ApplicationRoleStatus.ACTIVE,
)
# This ApplicationRole will not be in the results because it
# does not have a user associated.
ApplicationRoleFactory.create(
user=None, application=ready_app, status=ApplicationRoleStatus.ACTIVE,
)
# This ApplicationRole will not be in the results because its
# status is not ACTIVE.
ApplicationRoleFactory.create(
user=UserFactory.create(),
application=unready_app,
status=ApplicationRoleStatus.DISABLED,
)
app_ids = ApplicationRoles.get_pending_creation()
expected_ids = [[role_one.id, role_two.id], [role_three.id], [role_four.id]]
# Sort them to produce the same order.
assert sorted(app_ids) == sorted(expected_ids)
def test_get_many():
ar1 = ApplicationRoleFactory.create()
ar2 = ApplicationRoleFactory.create()
ApplicationRoleFactory.create()
result = ApplicationRoles.get_many([ar1.id, ar2.id])
assert result == [ar1, ar2]

36
tests/domain/test_environments.py
View File

@@ -1,5 +1,4 @@
import pytest
import pendulum
from uuid import uuid4
from atst.domain.environments import Environments
@@ -14,6 +13,7 @@ from tests.factories import (
EnvironmentRoleFactory,
ApplicationRoleFactory,
)
from tests.utils import EnvQueryTest
def test_create_environments():
@@ -119,40 +119,6 @@ def test_update_does_not_duplicate_names_within_application():
Environments.update(dupe_env, name)
class EnvQueryTest:
@property
def NOW(self):
return pendulum.now()
@property
def YESTERDAY(self):
return self.NOW.subtract(days=1)
@property
def TOMORROW(self):
return self.NOW.add(days=1)
def create_portfolio_with_clins(self, start_and_end_dates, env_data=None):
env_data = env_data or {}
return PortfolioFactory.create(
applications=[
{
"name": "Mos Eisley",
"description": "Where Han shot first",
"environments": [{"name": "thebar", **env_data}],
}
],
task_orders=[
{
"create_clins": [
{"start_date": start_date, "end_date": end_date}
for (start_date, end_date) in start_and_end_dates
]
}
],
)
class TestGetEnvironmentsPendingCreate(EnvQueryTest):
def test_with_expired_clins(self, session):
self.create_portfolio_with_clins([(self.YESTERDAY, self.YESTERDAY)])

49
tests/domain/test_portfolios.py
View File

@@ -26,6 +26,7 @@ from tests.factories import (
PortfolioStateMachineFactory,
get_all_portfolio_permission_sets,
)
from tests.utils import EnvQueryTest
@pytest.fixture(scope="function")
@@ -263,10 +264,44 @@ def test_create_state_machine(portfolio):
assert fsm
def test_get_portfolios_pending_provisioning(session):
for x in range(5):
portfolio = PortfolioFactory.create()
sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
if x == 2:
sm.state = FSMStates.COMPLETED
assert len(Portfolios.get_portfolios_pending_provisioning()) == 4
class TestGetPortfoliosPendingCreate(EnvQueryTest):
def test_finds_unstarted(self):
for x in range(5):
if x == 2:
state = "COMPLETED"
else:
state = "UNSTARTED"
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], state_machine_status=state
)
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 4
def test_finds_created(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], state_machine_status="TENANT_CREATED"
)
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 1
def test_does_not_find_failed(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], state_machine_status="TENANT_FAILED"
)
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 0
def test_with_expired_clins(self):
self.create_portfolio_with_clins([(self.YESTERDAY, self.YESTERDAY)])
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 0
def test_with_active_clins(self):
portfolio = self.create_portfolio_with_clins([(self.YESTERDAY, self.TOMORROW)])
Portfolios.get_portfolios_pending_provisioning(self.NOW) == [portfolio.id]
def test_with_future_clins(self):
self.create_portfolio_with_clins([(self.TOMORROW, self.TOMORROW)])
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 0
def test_with_already_provisioned_env(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], env_data={"cloud_id": uuid4().hex}
)
assert len(Portfolios.get_portfolios_pending_provisioning(self.NOW)) == 0