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

merge staging

Browse Source
This commit is contained in:
Philip Kalinsky 2020-02-06 10:39:08 -05:00
commit 1da7d42eb7
45 changed files with 912 additions and 717 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
alembic/versions/0039308c6351_remove_root_user_info_from_environment.py Normal file
View File

@ -0,0 +1,28 @@
"""Remove root_user_info from Environment
Revision ID: 0039308c6351
Revises: 17da2a475429
Create Date: 2020-02-04 14:37:06.814645
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '0039308c6351' # pragma: allowlist secret
down_revision = '17da2a475429' # pragma: allowlist secret
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('environments', 'root_user_info')
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('environments', sa.Column('root_user_info', postgresql.JSONB(astext_type=sa.Text()), autoincrement=False, nullable=True))
# ### end Alembic commands ###

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

@ -11,13 +11,9 @@ from .exceptions import (
UserProvisioningException,
ConnectionException,
UnknownServerException,
SecretException,
)
from .models import (
SubscriptionCreationCSPPayload,
SubscriptionCreationCSPResult,
SubscriptionVerificationCSPPayload,
SuscriptionVerificationCSPResult,
AdminRoleDefinitionCSPPayload,
AdminRoleDefinitionCSPResult,
ApplicationCSPPayload,
@ -30,14 +26,21 @@ from .models import (
BillingProfileTenantAccessCSPResult,
BillingProfileVerificationCSPPayload,
BillingProfileVerificationCSPResult,
CostManagementQueryCSPResult,
EnvironmentCSPPayload,
EnvironmentCSPResult,
KeyVaultCredentials,
ManagementGroupCSPResponse,
PrincipalAdminRoleCSPPayload,
PrincipalAdminRoleCSPResult,
ProductPurchaseCSPPayload,
ProductPurchaseCSPResult,
ProductPurchaseVerificationCSPPayload,
ProductPurchaseVerificationCSPResult,
PrincipalAdminRoleCSPPayload,
PrincipalAdminRoleCSPResult,
ReportingCSPPayload,
SubscriptionCreationCSPPayload,
SubscriptionCreationCSPResult,
SubscriptionVerificationCSPPayload,
SuscriptionVerificationCSPResult,
TaskOrderBillingCreationCSPPayload,
TaskOrderBillingCreationCSPResult,
TaskOrderBillingVerificationCSPPayload,
@ -59,7 +62,6 @@ from .models import (
)
from .policy import AzurePolicyManager
# This needs to be a fully pathed role definition identifier, not just a UUID
# TODO: Extract these from sdk msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD
AZURE_SKU_ID = "0001" # probably a static sku specific to ATAT/JEDI
@ -122,11 +124,19 @@ class AzureCloudProvider(CloudProviderInterface):
)
try:
return secret_client.set_secret(secret_key, secret_value)
<<<<<<< HEAD
except self.azure_exceptions.HttpResponseError:
=======
except self.sdk.exceptions.HttpResponseError as exc:
>>>>>>> staging
app.logger.error(
f"Could not SET secret in Azure keyvault for key {secret_key}.",
exc_info=1,
)
raise SecretException(
f"Could not SET secret in Azure keyvault for key {secret_key}.",
exc.message,
)
def get_secret(self, secret_key):
credential = self._get_client_secret_credential_obj()
@ -135,67 +145,39 @@ class AzureCloudProvider(CloudProviderInterface):
)
try:
return secret_client.get_secret(secret_key).value
<<<<<<< HEAD
except self.azure_exceptions.HttpResponseError:
=======
except self.sdk.exceptions.HttpResponseError:
>>>>>>> staging
app.logger.error(
f"Could not GET secret in Azure keyvault for key {secret_key}.",
exc_info=1,
)
raise SecretException(
f"Could not GET secret in Azure keyvault for key {secret_key}.",
exc.message,
)
def create_environment(self, auth_credentials: Dict, user, environment):
# since this operation would only occur within a tenant, should we source the tenant
# via lookup from environment once we've created the portfolio csp data schema
# something like this:
# environment_tenant = environment.application.portfolio.csp_data.get('tenant_id', None)
# though we'd probably source the whole credentials for these calls from the portfolio csp
# data, as it would have to be where we store the creds for the at-at user within the portfolio tenant
# credentials = self._get_credential_obj(environment.application.portfolio.csp_data.get_creds())
credentials = self._get_credential_obj(self._root_creds)
display_name = f"{environment.application.name}_{environment.name}_{environment.id}" # proposed format
management_group_id = "?" # management group id chained from environment
parent_id = "?" # from environment.application
management_group = self._create_management_group(
credentials, management_group_id, display_name, parent_id,
def create_environment(self, payload: EnvironmentCSPPayload):
creds = self._source_creds(payload.tenant_id)
credentials = self._get_credential_obj(
{
"client_id": creds.tenant_sp_client_id,
"secret_key": creds.tenant_sp_key,
"tenant_id": creds.tenant_id,
},
resource=self.sdk.cloud.endpoints.resource_manager,
)
return ManagementGroupCSPResponse(**management_group)
def create_atat_admin_user(
self, auth_credentials: Dict, csp_environment_id: str
) -> Dict:
root_creds = self._root_creds
credentials = self._get_credential_obj(root_creds)
sub_client = self.sdk.subscription.SubscriptionClient(credentials)
subscription = sub_client.subscriptions.get(csp_environment_id)
managment_principal = self._get_management_service_principal()
auth_client = self.sdk.authorization.AuthorizationManagementClient(
response = self._create_management_group(
credentials,
# TODO: Determine which subscription this needs to point at
# Once we're in a multi-sub environment
subscription.id,
payload.management_group_name,
payload.display_name,
payload.parent_id,
)
# Create role assignment for
role_assignment_id = str(uuid4())
role_assignment_create_params = auth_client.role_assignments.models.RoleAssignmentCreateParameters(
role_definition_id=REMOTE_ROOT_ROLE_DEF_ID,
principal_id=managment_principal.id,
)
auth_client.role_assignments.create(
scope=f"/subscriptions/{subscription.id}/",
role_assignment_name=role_assignment_id,
parameters=role_assignment_create_params,
)
return {
"csp_user_id": managment_principal.object_id,
"credentials": managment_principal.password_credentials,
"role_name": role_assignment_id,
}
return EnvironmentCSPResult(**response)
def create_application(self, payload: ApplicationCSPPayload):
creds = self._source_creds(payload.tenant_id)
@ -822,17 +804,6 @@ class AzureCloudProvider(CloudProviderInterface):
return self._ok()
def create_billing_alerts(self, TBD):
# TODO: Add azure-mgmt-consumption for Budget and Notification entities/operations
# TODO: Determine how to auth against that API using the SDK, doesn't seeem possible at the moment
# TODO: billing alerts are registered as Notifications on Budget objects, which have start/end dates
# TODO: determine what the keys in the Notifications dict are supposed to be
# we may need to rotate budget objects when new TOs/CLINs are reported?
# we likely only want the budget ID, can be updated or replaced?
response = {"id": "id"}
return self._ok({"budget_id": response["id"]})
def _get_management_service_principal(self):
# we really should be using graph.microsoft.com, but i'm getting
# "expired token" errors for that
@ -1094,3 +1065,41 @@ class AzureCloudProvider(CloudProviderInterface):
hashed = sha256_hex(tenant_id)
raw_creds = self.get_secret(hashed)
return KeyVaultCredentials(**json.loads(raw_creds))
def get_reporting_data(self, payload: ReportingCSPPayload):
"""
Queries the Cost Management API for an invoice section's raw reporting data
We query at the invoiceSection scope. The full scope path is passed in
with the payload at the `invoice_section_id` key.
"""
creds = self._source_tenant_creds(payload.tenant_id)
token = self._get_sp_token(
payload.tenant_id, creds.tenant_sp_client_id, creds.tenant_sp_key
)
if not token:
raise AuthenticationException("Could not retrieve tenant access token")
headers = {"Authorization": f"Bearer {token}"}
request_body = {
"type": "Usage",
"timeframe": "Custom",
"timePeriod": {"from": payload.from_date, "to": payload.to_date,},
"dataset": {
"granularity": "Daily",
"aggregation": {"totalCost": {"name": "PreTaxCost", "function": "Sum"}},
"grouping": [{"type": "Dimension", "name": "InvoiceId"}],
},
}
cost_mgmt_url = (
f"/providers/Microsoft.CostManagement/query?api-version=2019-11-01"
)
result = self.sdk.requests.post(
f"{self.sdk.cloud.endpoints.resource_manager}{payload.invoice_section_id}{cost_mgmt_url}",
json=request_body,
headers=headers,
)
if result.ok:
return CostManagementQueryCSPResult(**result.json())

29
atst/domain/csp/cloud/cloud_provider_interface.py
View File

@ -11,7 +11,7 @@ class CloudProviderInterface:
def root_creds(self) -> Dict:
raise NotImplementedError()
def create_environment(self, auth_credentials: Dict, user, environment) -> str:
def create_environment(self, payload):
"""Create a new environment in the CSP.
Arguments:
@ -31,33 +31,6 @@ class CloudProviderInterface:
"""
raise NotImplementedError()
def create_atat_admin_user(
self, auth_credentials: Dict, csp_environment_id: str
) -> Dict:
"""Creates a new, programmatic user in the CSP. Grants this user full permissions to administer
the CSP.
Arguments:
auth_credentials -- Object containing CSP account credentials
csp_environment_id -- ID of the CSP Environment the admin user should be created in
Returns:
object: Object representing new remote admin user, including credentials
Something like:
{
"user_id": string,
"credentials": dict, # structure TBD based on csp
}
Raises:
AuthenticationException: Problem with the credentials
AuthorizationException: Credentials not authorized for current action(s)
ConnectionException: Issue with the CSP API connection
UnknownServerException: Unknown issue on the CSP side
UserProvisioningException: Problem creating the root user
"""
raise NotImplementedError()
def create_or_update_user(
self, auth_credentials: Dict, user_info, csp_role_id: str
) -> str:

14
atst/domain/csp/cloud/exceptions.py
View File

@ -118,3 +118,17 @@ class BaselineProvisionException(GeneralCSPException):
return "Could not complete baseline provisioning for environment ({}): {}".format(
self.env_identifier, self.reason
)
class SecretException(GeneralCSPException):
"""A problem occurred with setting or getting secrets"""
def __init__(self, tenant_id, reason):
self.tenant_id = tenant_id
self.reason = reason
@property
def message(self):
return "Could not get or set secret for ({}): {}".format(
self.tenant_id, self.reason
)

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

@ -4,9 +4,7 @@ from .cloud_provider_interface import CloudProviderInterface
from .exceptions import (
AuthenticationException,
AuthorizationException,
BaselineProvisionException,
ConnectionException,
EnvironmentCreationException,
GeneralCSPException,
UnknownServerException,
UserProvisioningException,
@ -25,16 +23,21 @@ from .models import (
BillingProfileTenantAccessCSPResult,
BillingProfileVerificationCSPPayload,
BillingProfileVerificationCSPResult,
CostManagementQueryCSPResult,
CostManagementQueryProperties,
ProductPurchaseCSPPayload,
ProductPurchaseCSPResult,
ProductPurchaseVerificationCSPPayload,
ProductPurchaseVerificationCSPResult,
PrincipalAdminRoleCSPPayload,
PrincipalAdminRoleCSPResult,
ReportingCSPPayload,
SubscriptionCreationCSPPayload,
SubscriptionCreationCSPResult,
SubscriptionVerificationCSPPayload,
SuscriptionVerificationCSPResult,
EnvironmentCSPPayload,
EnvironmentCSPResult,
TaskOrderBillingCreationCSPPayload,
TaskOrderBillingCreationCSPResult,
TaskOrderBillingVerificationCSPPayload,
@ -91,34 +94,6 @@ class MockCloudProvider(CloudProviderInterface):
def get_secret(self, secret_key: str, default=dict()):
return default
def create_environment(self, auth_credentials, user, environment):
self._authorize(auth_credentials)
self._delay(1, 5)
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(
self.ENV_CREATE_FAILURE_PCT,
EnvironmentCreationException(
environment.id, "Could not create environment."
),
)
csp_environment_id = self._id()
self._delay(1, 5)
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(
self.ATAT_ADMIN_CREATE_FAILURE_PCT,
BaselineProvisionException(
csp_environment_id, "Could not create environment baseline."
),
)
self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
return csp_environment_id
def create_subscription(self, payload: SubscriptionCreationCSPPayload):
return self.create_subscription_creation(payload)
@ -142,23 +117,6 @@ class MockCloudProvider(CloudProviderInterface):
subscription_id="subscriptions/60fbbb72-0516-4253-ab18-c92432ba3230"
)
def create_atat_admin_user(self, auth_credentials, csp_environment_id):
self._authorize(auth_credentials)
self._delay(1, 5)
self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
self._maybe_raise(
self.ATAT_ADMIN_CREATE_FAILURE_PCT,
UserProvisioningException(
csp_environment_id, "atat_admin", "Could not create admin user."
),
)
self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
return {"id": self._id(), "credentials": self._auth_credentials}
def create_tenant(self, payload: TenantCSPPayload):
"""
payload is an instance of TenantCSPPayload data class
@ -477,6 +435,13 @@ class MockCloudProvider(CloudProviderInterface):
id=f"{AZURE_MGMNT_PATH}{payload.management_group_name}"
)
def create_environment(self, payload: EnvironmentCSPPayload):
self._maybe_raise(self.UNAUTHORIZED_RATE, GeneralCSPException)
return EnvironmentCSPResult(
id=f"{AZURE_MGMNT_PATH}{payload.management_group_name}"
)
def create_user(self, payload: UserCSPPayload):
self._maybe_raise(self.UNAUTHORIZED_RATE, GeneralCSPException)
@ -487,3 +452,25 @@ class MockCloudProvider(CloudProviderInterface):
def update_tenant_creds(self, tenant_id, secret):
return secret
def get_reporting_data(self, payload: ReportingCSPPayload):
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)
object_id = str(uuid4())
properties = CostManagementQueryProperties(
**dict(
columns=[
{"name": "PreTaxCost", "type": "Number"},
{"name": "UsageDate", "type": "Number"},
{"name": "InvoiceId", "type": "String"},
{"name": "Currency", "type": "String"},
],
rows=[],
)
)
return CostManagementQueryCSPResult(
**dict(name=object_id, properties=properties,)
)

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

@ -358,6 +358,14 @@ class ApplicationCSPResult(ManagementGroupCSPResponse):
pass
class EnvironmentCSPPayload(ManagementGroupCSPPayload):
pass
class EnvironmentCSPResult(ManagementGroupCSPResponse):
pass
class KeyVaultCredentials(BaseModel):
root_sp_client_id: Optional[str]
root_sp_key: Optional[str]
@ -499,3 +507,34 @@ class UserCSPPayload(BaseCSPPayload):
class UserCSPResult(AliasModel):
id: str
class QueryColumn(AliasModel):
name: str
type: str
class CostManagementQueryProperties(AliasModel):
columns: List[QueryColumn]
rows: List[Optional[list]]
class CostManagementQueryCSPResult(AliasModel):
name: str
properties: CostManagementQueryProperties
class ReportingCSPPayload(BaseCSPPayload):
invoice_section_id: str
from_date: str
to_date: str
@root_validator(pre=True)
def extract_invoice_section(cls, values):
try:
values["invoice_section_id"] = values["billing_profile_properties"][
"invoice_sections"
][0]["invoice_section_id"]
return values
except (KeyError, IndexError):
raise ValueError("Invoice section ID not present in payload")

8
atst/domain/environment_roles.py
View File

@ -106,9 +106,11 @@ class EnvironmentRoles(object):
def disable(cls, environment_role_id):
environment_role = EnvironmentRoles.get_by_id(environment_role_id)
if environment_role.csp_user_id and not environment_role.environment.is_pending:
credentials = environment_role.environment.csp_credentials
app.csp.cloud.disable_user(credentials, environment_role.csp_user_id)
if environment_role.csp_user_id and not environment_role.environment.cloud_id:
tenant_id = environment_role.environment.application.portfolio.csp_data.get(
"tenant_id"
)
app.csp.cloud.disable_user(tenant_id, environment_role.csp_user_id)
environment_role.status = EnvironmentRole.Status.DISABLED
db.session.add(environment_role)

17
atst/domain/environments.py
View File

@ -124,18 +124,9 @@ class Environments(object):
Any environment with an active CLIN that doesn't yet have a `cloud_id`.
"""
results = (
cls.base_provision_query(now).filter(Environment.cloud_id == None).all()
cls.base_provision_query(now)
.filter(Application.cloud_id != None)
.filter(Environment.cloud_id.is_(None))
.all()
)
return [id_ for id_, in results]
@classmethod
def get_environments_pending_atat_user_creation(cls, now) -> List[UUID]:
"""
Any environment with an active CLIN that has a cloud_id but no `root_user_info`.
"""
results = (
cls.base_provision_query(now)
.filter(Environment.cloud_id != None)
.filter(Environment.root_user_info == None)
).all()
return [id_ for id_, in results]

22
atst/domain/portfolios/portfolios.py
View File

@ -15,6 +15,8 @@ from atst.models import (
Permissions,
PortfolioRole,
PortfolioRoleStatus,
TaskOrder,
CLIN,
)
from .query import PortfoliosQuery, PortfolioStateMachinesQuery
@ -144,7 +146,7 @@ class Portfolios(object):
return db.session.query(Portfolio.id)
@classmethod
def get_portfolios_pending_provisioning(cls) -> List[UUID]:
def get_portfolios_pending_provisioning(cls, now) -> List[UUID]:
"""
Any portfolio with a corresponding State Machine that is either:
not started yet,
@ -153,22 +155,18 @@ class Portfolios(object):
"""
results = (
cls.base_provision_query()
db.session.query(Portfolio.id)
.join(PortfolioStateMachine)
.join(TaskOrder)
.join(CLIN)
.filter(Portfolio.deleted == False)
.filter(CLIN.start_date <= now)
.filter(CLIN.end_date > now)
.filter(
or_(
PortfolioStateMachine.state == FSMStates.UNSTARTED,
PortfolioStateMachine.state == FSMStates.FAILED,
PortfolioStateMachine.state == FSMStates.TENANT_FAILED,
PortfolioStateMachine.state.like("%CREATED"),
)
)
)
return [id_ for id_, in results]
# db.session.query(PortfolioStateMachine).\
# filter(
# or_(
# PortfolioStateMachine.state==FSMStates.UNSTARTED,
# PortfolioStateMachine.state==FSMStates.UNSTARTED,
# )
# ).all()

78
atst/jobs.py
View File

@ -1,18 +1,21 @@
from flask import current_app as app
import pendulum
from flask import current_app as app
from atst.database import db
from atst.queue import celery
from atst.models import JobFailure
from atst.domain.csp.cloud.exceptions import GeneralCSPException
from atst.domain.csp.cloud import CloudProviderInterface
from atst.domain.application_roles import ApplicationRoles
from atst.domain.applications import Applications
from atst.domain.csp.cloud import CloudProviderInterface
from atst.domain.csp.cloud.exceptions import GeneralCSPException
from atst.domain.csp.cloud.models import (
ApplicationCSPPayload,
EnvironmentCSPPayload,
UserCSPPayload,
)
from atst.domain.environments import Environments
from atst.domain.portfolios import Portfolios
from atst.domain.application_roles import ApplicationRoles
from atst.models import JobFailure
from atst.models.utils import claim_for_update, claim_many_for_update
from atst.utils.localization import translate
from atst.domain.csp.cloud.models import ApplicationCSPPayload, UserCSPPayload
from atst.queue import celery
class RecordFailure(celery.Task):
@ -109,45 +112,17 @@ def do_create_environment(csp: CloudProviderInterface, environment_id=None):
with claim_for_update(environment) as environment:
if environment.cloud_id is not None:
# TODO: Return value for this?
return
user = environment.creator
# we'll need to do some checking in this job for cases where it's retrying
# when a failure occured after some successful steps
# (e.g. if environment.cloud_id is not None, then we can skip first step)
# credentials either from a given user or pulled from config?
# if using global creds, do we need to log what user authorized action?
atat_root_creds = csp.root_creds()
# user is needed because baseline root account in the environment will
# be assigned to the requesting user, open question how to handle duplicate
# email addresses across new environments
csp_environment_id = csp.create_environment(atat_root_creds, user, environment)
environment.cloud_id = csp_environment_id
db.session.add(environment)
db.session.commit()
body = render_email(
"emails/application/environment_ready.txt", {"environment": environment}
)
app.mailer.send(
[environment.creator.email], translate("email.environment_ready"), body
csp_details = environment.application.portfolio.csp_data
parent_id = environment.application.cloud_id
tenant_id = csp_details.get("tenant_id")
payload = EnvironmentCSPPayload(
tenant_id=tenant_id, display_name=environment.name, parent_id=parent_id
)
def do_create_atat_admin_user(csp: CloudProviderInterface, environment_id=None):
environment = Environments.get(environment_id)
with claim_for_update(environment) as environment:
atat_root_creds = csp.root_creds()
atat_remote_root_user = csp.create_atat_admin_user(
atat_root_creds, environment.cloud_id
)
environment.root_user_info = atat_remote_root_user
env_result = csp.create_environment(payload)
environment.cloud_id = env_result.id
db.session.add(environment)
db.session.commit()
@ -191,19 +166,12 @@ def create_environment(self, environment_id=None):
do_work(do_create_environment, self, app.csp.cloud, environment_id=environment_id)
@celery.task(bind=True, base=RecordFailure)
def create_atat_admin_user(self, environment_id=None):
do_work(
do_create_atat_admin_user, self, app.csp.cloud, environment_id=environment_id
)
@celery.task(bind=True)
def dispatch_provision_portfolio(self):
"""
Iterate over portfolios with a corresponding State Machine that have not completed.
"""
for portfolio_id in Portfolios.get_portfolios_pending_provisioning():
for portfolio_id in Portfolios.get_portfolios_pending_provisioning(pendulum.now()):
provision_portfolio.delay(portfolio_id=portfolio_id)
@ -225,11 +193,3 @@ def dispatch_create_environment(self):
pendulum.now()
):
create_environment.delay(environment_id=environment_id)
@celery.task(bind=True)
def dispatch_create_atat_admin_user(self):
for environment_id in Environments.get_environments_pending_atat_user_creation(
pendulum.now()
):
create_atat_admin_user.delay(environment_id=environment_id)

28
atst/models/environment.py
View File

@ -1,11 +1,9 @@
from sqlalchemy import Column, ForeignKey, String, UniqueConstraint
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import JSONB
from enum import Enum
from atst.models.base import Base
import atst.models.mixins as mixins
import atst.models.types as types
from atst.models.base import Base
class Environment(
@ -30,7 +28,6 @@ class Environment(
creator = relationship("User")
cloud_id = Column(String)
root_user_info = Column(JSONB(none_as_null=True))
roles = relationship(
"EnvironmentRole",
@ -44,10 +41,6 @@ class Environment(
),
)
class ProvisioningStatus(Enum):
PENDING = "pending"
COMPLETED = "completed"
@property
def users(self):
return {r.application_role.user for r in self.roles}
@ -68,17 +61,6 @@ class Environment(
def portfolio_id(self):
return self.application.portfolio_id
@property
def provisioning_status(self) -> ProvisioningStatus:
if self.cloud_id is None or self.root_user_info is None:
return self.ProvisioningStatus.PENDING
else:
return self.ProvisioningStatus.COMPLETED
@property
def is_pending(self):
return self.provisioning_status == self.ProvisioningStatus.PENDING
def __repr__(self):
return "<Environment(name='{}', num_users='{}', application='{}', portfolio='{}', id='{}')>".format(
self.name,
@ -91,11 +73,3 @@ class Environment(
@property
def history(self):
return self.get_changes()
@property
def csp_credentials(self):
return (
self.root_user_info.get("credentials")
if self.root_user_info is not None
else None
)

40
atst/models/portfolio_state_machine.py
View File

@ -179,34 +179,30 @@ class PortfolioStateMachine(
try:
func_name = f"create_{stage}"
response = getattr(self.csp, func_name)(payload_data)
if self.portfolio.csp_data is None:
self.portfolio.csp_data = {}
self.portfolio.csp_data.update(response.dict())
db.session.add(self.portfolio)
db.session.commit()
except PydanticValidationError as exc:
app.logger.error(
f"Failed to cast response to valid result class {self.__repr__()}:",
exc_info=1,
)
app.logger.info(exc.json())
print(exc.json())
app.logger.info(payload_data)
# TODO: Ensure that failing the stage does not preclude a Celery retry
self.fail_stage(stage)
# TODO: catch and handle general CSP exception here
except (ConnectionException, UnknownServerException) as exc:
app.logger.error(
f"CSP api call. Caught exception for {self.__repr__()}.", exc_info=1,
)
# TODO: Ensure that failing the stage does not preclude a Celery retry
self.fail_stage(stage)
if response.get("status") == "error":
self.fail_stage(stage)
elif response.get("status") == "ok":
try:
if self.portfolio.csp_data is None:
self.portfolio.csp_data = {}
self.portfolio.csp_data.update(response.dict())
db.session.add(self.portfolio)
db.session.commit()
except PydanticValidationError as exc:
app.logger.error(
f"Failed to cast response to valid result class {self.__repr__()}:",
exc_info=1,
)
app.logger.info(exc.json())
print(exc.json())
app.logger.info(payload_data)
self.fail_stage(stage)
self.finish_stage(stage)
self.finish_stage(stage)
def is_csp_data_valid(self, event):
"""

4
atst/queue.py
View File

@ -19,10 +19,6 @@ def update_celery(celery, app):
"task": "atst.jobs.dispatch_create_environment",
"schedule": 60,
},
"beat-dispatch_create_atat_admin_user": {
"task": "atst.jobs.dispatch_create_atat_admin_user",
"schedule": 60,
},
"beat-dispatch_create_user": {
"task": "atst.jobs.dispatch_create_user",
"schedule": 60,

6
atst/routes/applications/settings.py
View File

@ -39,7 +39,6 @@ def get_environments_obj_for_app(application):
{
"id": env.id,
"name": env.name,
"pending": env.is_pending,
"edit_form": EditEnvironmentForm(obj=env),
"member_count": len(env.roles),
"members": sorted(
@ -467,9 +466,10 @@ def revoke_invite(application_id, application_role_id):
if invite.is_pending:
ApplicationInvitations.revoke(invite.token)
flash(
"application_invite_revoked",
"invite_revoked",
resource="Application",
user_name=app_role.user_name,
application_name=g.application.name,
resource_name=g.application.name,
)
else:
flash(

8
atst/routes/portfolios/invitations.py
View File

@ -37,8 +37,14 @@ def accept_invitation(portfolio_token):
)
@user_can(Permissions.EDIT_PORTFOLIO_USERS, message="revoke invitation")
def revoke_invitation(portfolio_id, portfolio_token):
PortfolioInvitations.revoke(portfolio_token)
invite = PortfolioInvitations.revoke(portfolio_token)
flash(
"invite_revoked",
resource="Portfolio",
user_name=invite.user_name,
resource_name=g.portfolio.name,
)
return redirect(
url_for(
"portfolios.admin",

10
atst/utils/flash.py
View File

@ -33,11 +33,6 @@ MESSAGES = {
"message": "flash.application_invite.resent.message",
"category": "success",
},
"application_invite_revoked": {
"title": "flash.application_invite.revoked.title",
"message": "flash.application_invite.revoked.message",
"category": "success",
},
"application_member_removed": {
"title": "flash.application_member.removed.title",
"message": "flash.application_member.removed.message",
@ -103,6 +98,11 @@ MESSAGES = {
"message": None,
"category": "warning",
},
"invite_revoked": {
"title": "flash.invite_revoked.title",
"message": "flash.invite_revoked.message",
"category": "success",
},
"logged_out": {
"title": "flash.logged_out.title",
"message": "flash.logged_out.message",

21
js/components/sidenav_toggler.js
View File

@ -1,30 +1,21 @@
import ExpandSidenavMixin from '../mixins/expand_sidenav'
import ToggleMixin from '../mixins/toggle'
const cookieName = 'expandSidenav'
export default {
name: 'sidenav-toggler',
mixins: [ToggleMixin],
mixins: [ExpandSidenavMixin, ToggleMixin],
props: {
defaultVisible: {
type: Boolean,
default: function() {
if (document.cookie.match(cookieName)) {
return !!document.cookie.match(cookieName + ' *= *true')
} else {
return true
}
},
},
mounted: function() {
this.$parent.$emit('sidenavToggle', this.isVisible)
},
methods: {
toggle: function(e) {
e.preventDefault()
this.isVisible = !this.isVisible
document.cookie = cookieName + '=' + this.isVisible + '; path=/'
document.cookie = this.cookieName + '=' + this.isVisible + '; path=/'
this.$parent.$emit('sidenavToggle', this.isVisible)
},
},
}

12
js/index.js
View File

@ -32,12 +32,14 @@ import ToForm from './components/forms/to_form'
import ClinFields from './components/clin_fields'
import PopDateRange from './components/pop_date_range'
import ToggleMenu from './components/toggle_menu'
import ExpandSidenav from './mixins/expand_sidenav'
Vue.config.productionTip = false
Vue.use(VTooltip)
Vue.mixin(Modal)
Vue.mixin(ExpandSidenav)
const app = new Vue({
el: '#app-root',
@ -67,6 +69,12 @@ const app = new Vue({
ToggleMenu,
},
data: function() {
return {
sidenavExpanded: this.defaultVisible,
}
},
mounted: function() {
this.$on('modalOpen', data => {
if (data['isOpen']) {
@ -105,6 +113,10 @@ const app = new Vue({
}
})
})
this.$on('sidenavToggle', data => {
this.sidenavExpanded = data
})
},
delimiters: ['!{', '}'],

15
js/mixins/expand_sidenav.js Normal file
View File

@ -0,0 +1,15 @@
export default {
props: {
cookieName: 'expandSidenav',
defaultVisible: {
type: Boolean,
default: function() {
if (document.cookie.match(this.cookieName)) {
return !!document.cookie.match(this.cookieName + ' *= *true')
} else {
return true
}
},
},
},
}

1
styles/atat.scss
View File

@ -47,3 +47,4 @@
@import "sections/application_edit";
@import "sections/reports";
@import "sections/task_order";
@import "sections/ccpo";

36
styles/components/_alerts.scss
View File

@ -11,6 +11,7 @@
.usa-alert {
padding-bottom: 2.4rem;
max-width: $max-panel-width;
}
@mixin alert {
@ -97,38 +98,3 @@
}
}
}
.alert {
@include alert;
@include alert-level("info");
&.alert--success {
@include alert-level("success");
.alert__actions {
.icon-link {
@include icon-link-color($color-green, $color-white);
}
}
}
&.alert--warning {
@include alert-level("warning");
.alert__actions {
.icon-link {
@include icon-link-color($color-gold-dark, $color-white);
}
}
}
&.alert--error {
@include alert-level("error");
.alert__actions {
.icon-link {
@include icon-link-color($color-red, $color-white);
}
}
}
}

20
styles/components/_error_page.scss
View File

@ -1,8 +1,13 @@
.error-page {
max-width: 475px;
margin: auto;
max-width: $max-page-width;
.panel {
box-shadow: none;
background-color: unset;
border: none;
max-width: 475px;
margin: auto;
&__heading {
text-align: center;
padding: $gap 0;
@ -15,17 +20,6 @@
margin-bottom: $gap;
}
}
&__body {
padding: $gap * 2;
margin: 0;
hr {
width: 80%;
margin: auto;
margin-bottom: $gap * 3;
}
}
}
.icon {

2
styles/components/_topbar.scss
View File

@ -12,7 +12,7 @@
flex-direction: row;
align-items: stretch;
justify-content: space-between;
max-width: 1190px;
max-width: $max-page-width;
a {
color: $color-white;

1
styles/core/_variables.scss
View File

@ -19,6 +19,7 @@ $sidenav-collapsed-width: 10rem;
$max-panel-width: 90rem;
$home-pg-icon-width: 6rem;
$large-spacing: 4rem;
$max-page-width: $max-panel-width + $sidenav-expanded-width + $large-spacing;
/*
* USWDS Variables

35
styles/elements/_action_group.scss
View File

@ -32,22 +32,35 @@
}
.action-group-footer {
@extend .action-group;
&:last-child {
margin-bottom: 0;
}
margin-top: 0;
margin-bottom: 0;
padding-top: $gap;
padding-bottom: $gap;
padding-right: $gap * 4;
position: fixed;
bottom: $footer-height;
left: 0;
background: white;
right: 0;
padding-right: $gap * 4;
border-top: 1px solid $color-gray-lighter;
width: 100%;
z-index: 1;
width: 100%;
&.action-group-footer--expand-offset {
padding-left: $sidenav-expanded-width;
}
&.action-group-footer--collapse-offset {
padding-left: $sidenav-collapsed-width;
}
.action-group-footer--container {
@extend .action-group;
margin-top: 0;
margin-bottom: 0;
margin-left: $large-spacing;
max-width: $max-panel-width;
&:last-child {
margin-bottom: 0;
}
}
}

3
styles/sections/_ccpo.scss Normal file
View File

@ -0,0 +1,3 @@
.ccpo-panel-container {
max-width: $max-panel-width;
}

20
templates/applications/new/step_1.html
View File

@ -39,14 +39,18 @@
</div>
<span class="action-group-footer">
{% block next_button %}
{{ SaveButton(text=('portfolios.applications.new.step_1_button_text' | translate)) }}
{% endblock %}
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
Cancel
</a>
</span>
<div
class="action-group-footer"
v-bind:class="{'action-group-footer--expand-offset': this.$root.sidenavExpanded, 'action-group-footer--collapse-offset': !this.$root.sidenavExpanded}">
<div class="action-group-footer--container">
{% block next_button %}
{{ SaveButton(text=('portfolios.applications.new.step_1_button_text' | translate)) }}
{% endblock %}
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
Cancel
</a>
</div>
</div>
</form>
</base-form>

28
templates/applications/new/step_2.html
View File

@ -58,20 +58,24 @@
{{ Icon("plus") }}
</button>
</div>
</div>
</div>
</div>
<div
class="action-group-footer"
v-bind:class="{'action-group-footer--expand-offset': this.$root.sidenavExpanded, 'action-group-footer--collapse-offset': !this.$root.sidenavExpanded}">
<div class="action-group-footer--container">
{% block next_button %}
{{ SaveButton(text=('portfolios.applications.new.step_2_button_text' | translate)) }}
{% endblock %}
<a class="usa-button usa-button-secondary" href="{{ url_for('applications.view_new_application_step_1', application_id=application.id) }}">
Previous
</a>
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
Cancel
</a>
</div>
</div>
</div>
<span class="action-group-footer">
{% block next_button %}
{{ SaveButton(text=('portfolios.applications.new.step_2_button_text' | translate)) }}
{% endblock %}
<a class="usa-button usa-button-secondary" href="{{ url_for('applications.view_new_application_step_1', application_id=application.id) }}">
Previous
</a>
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
Cancel
</a>
</span>
</form>
</application-environments>

26
templates/applications/new/step_3.html
View File

@ -25,16 +25,20 @@
action_update="applications.update_new_application_step_3") }}
<span class="action-group-footer">
<a class="usa-button" href="{{ url_for('applications.settings', application_id=application_id) }}">
{{ "portfolios.applications.new.step_3_button_text" | translate }}
</a>
<a class="usa-button usa-button-secondary" href="{{ url_for('applications.view_new_application_step_2', application_id=application.id) }}">
{{ "common.previous" | translate }}
</a>
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
{{ "common.cancel" | translate }}
</a>
</span>
<div
class="action-group-footer"
v-bind:class="{'action-group-footer--expand-offset': this.$root.sidenavExpanded, 'action-group-footer--collapse-offset': !this.$root.sidenavExpanded}">
<div class="action-group-footer--container">
<a class="usa-button" href="{{ url_for('applications.settings', application_id=application_id) }}">
{{ "portfolios.applications.new.step_3_button_text" | translate }}
</a>
<a class="usa-button usa-button-secondary" href="{{ url_for('applications.view_new_application_step_2', application_id=application.id) }}">
{{ "common.previous" | translate }}
</a>
<a href="{{ url_for('applications.portfolio_applications', portfolio_id=portfolio.id) }}">
{{ "common.cancel" | translate }}
</a>
</div>
</div>
{% endblock %}

32
templates/ccpo/add_user.html
View File

@ -4,21 +4,23 @@
{% from "components/text_input.html" import TextInput %}
{% block content %}
<base-form inline-template>
<form id="add-ccpo-user-form" action="{{ url_for('ccpo.submit_new_user') }}" method="POST">
{{ form.csrf_token }}
<h1>{{ "ccpo.form.add_user_title" | translate }}</h1>
<div class='form-row'>
<div class='form-col form-col--two-thirds'>
{{ TextInput(form.dod_id, validation='dodId', optional=False) }}
</div>
<div class="form-col form-col--third">
<div class='action-group'>
{{ SaveButton(text="common.next"|translate, element="input", additional_classes="action-group__action", form="add-ccpo-user-form") }}
<a class='action-group__action icon-link icon-link--default' href="{{ url_for('ccpo.users') }}">{{ "common.cancel" | translate }}</a>
<div class="ccpo-panel-container">
<base-form inline-template>
<form id="add-ccpo-user-form" action="{{ url_for('ccpo.submit_new_user') }}" method="POST">
{{ form.csrf_token }}
<h1>{{ "ccpo.form.add_user_title" | translate }}</h1>
<div class='form-row'>
<div class='form-col form-col--two-thirds'>
{{ TextInput(form.dod_id, validation='dodId', optional=False) }}
</div>
<div class="form-col form-col--third">
<div class='action-group'>
{{ SaveButton(text="common.next"|translate, element="input", additional_classes="action-group__action", form="add-ccpo-user-form") }}
<a class='action-group__action icon-link icon-link--default' href="{{ url_for('ccpo.users') }}">{{ "common.cancel" | translate }}</a>
</div>
</div>
</div>
</div>
</form>
</base-form>
</form>
</base-form>
</div>
{% endblock %}

56
templates/ccpo/confirm_user.html
View File

@ -3,31 +3,33 @@
{% from "components/text_input.html" import TextInput %}
{% block content %}
{% if new_user %}
<h3>{{ 'ccpo.form.confirm_user_title' | translate }}</h3>
<form id="add-ccpo-user-form" action="{{ url_for('ccpo.confirm_new_user') }}" method="POST">
{{ form.csrf_token }}
<input type="hidden" name="dod_id" value="{{ form.dod_id.data }}">
<div>
<p>
{{ "ccpo.form.confirm_user_text" | translate }}
</p>
<p>
{{ new_user.full_name }}
</p>
<p>
{{ new_user.email }}
</p>
</div>
<div class='action-group'>
<input
type='submit'
class='action-group__action usa-button'
value='{{ "ccpo.form.confirm_button" | translate }}'>
<a class='action-group__action icon-link icon-link--default' href="{{ url_for('ccpo.users') }}">
{{ "common.cancel" | translate }}
</a>
</div>
</form>
{% endif %}
<div class="ccpo-panel-container">
{% if new_user %}
<h3>{{ 'ccpo.form.confirm_user_title' | translate }}</h3>
<form id="add-ccpo-user-form" action="{{ url_for('ccpo.confirm_new_user') }}" method="POST">
{{ form.csrf_token }}
<input type="hidden" name="dod_id" value="{{ form.dod_id.data }}">
<div>
<p>
{{ "ccpo.form.confirm_user_text" | translate }}
</p>
<p>
{{ new_user.full_name }}
</p>
<p>
{{ new_user.email }}
</p>
</div>
<div class='action-group'>
<input
type='submit'
class='action-group__action usa-button'
value='{{ "ccpo.form.confirm_button" | translate }}'>
<a class='action-group__action icon-link icon-link--default' href="{{ url_for('ccpo.users') }}">
{{ "common.cancel" | translate }}
</a>
</div>
</form>
{% endif %}
</div>
{% endblock %}

136
templates/ccpo/users.html
View File

@ -6,78 +6,80 @@
{% from "components/modal.html" import Modal %}
{% block content %}
<div class='col'>
<div class="h2">
{{ "ccpo.users_title" | translate }}
</div>
<div class="ccpo-panel-container">
<div class='col'>
<div class="h2">
{{ "ccpo.users_title" | translate }}
</div>
{% include "fragments/flash.html" %}
<table>
<thead>
<tr>
<th>{{ "common.name" | translate }}</th>
<th>{{ "common.email" | translate }}</th>
<th>{{ "common.dod_id" | translate }}</th>
{% if user_can(permissions.DELETE_CCPO_USER) %}
<th></th>
{% endif %}
</tr>
</thead>
<tbody>
{% for user, form in users_info %}
{% set modal_id = "disable_ccpo_user_{}".format(user.dod_id) %}
{% set disable_button_class = 'button-danger-outline' %}
{% if user == g.current_user %}
{% set disable_button_class = "usa-button-disabled" %}
{% endif %}
{% include "fragments/flash.html" %}
<table>
<thead>
<tr>
<td>{{ user.full_name }}</td>
<td>{{ user.email }}</td>
<td>{{ user.dod_id }}</td>
<th>{{ "common.name" | translate }}</th>
<th>{{ "common.email" | translate }}</th>
<th>{{ "common.dod_id" | translate }}</th>
{% if user_can(permissions.DELETE_CCPO_USER) %}
<td>
<a v-on:click="openModal('{{ modal_id }}')" class='usa-button {{ disable_button_class }}'>
{{ "common.disable" | translate }}
</a>
</td>
<th></th>
{% endif %}
</tr>
{% endfor %}
</tbody>
</table>
</thead>
<tbody>
{% for user, form in users_info %}
{% set modal_id = "disable_ccpo_user_{}".format(user.dod_id) %}
{% set disable_button_class = 'button-danger-outline' %}
{% if user == g.current_user %}
{% set disable_button_class = "usa-button-disabled" %}
{% endif %}
<tr>
<td>{{ user.full_name }}</td>
<td>{{ user.email }}</td>
<td>{{ user.dod_id }}</td>
{% if user_can(permissions.DELETE_CCPO_USER) %}
<td>
<a v-on:click="openModal('{{ modal_id }}')" class='usa-button {{ disable_button_class }}'>
{{ "common.disable" | translate }}
</a>
</td>
{% endif %}
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% if user_can(permissions.CREATE_CCPO_USER) %}
<a class="icon-link" href="{{ url_for('ccpo.add_new_user')}}">
{{ "ccpo.add_user" | translate }} {{ Icon("plus") }}
</a>
{% endif %}
{% if user_can(permissions.DELETE_CCPO_USER) %}
{% for user, form in users_info %}
{% set modal_id = "disable_ccpo_user_{}".format(user.dod_id) %}
{% call Modal(name=modal_id) %}
<h1>Disable CCPO User</h1>
<hr>
{{
Alert(
title=("components.modal.destructive_title" | translate),
message=("ccpo.disable_user.alert_message" | translate({"user_name": user.full_name})),
level="warning"
)
}}
{{
DeleteConfirmation(
modal_id=modal_id,
delete_text='Remove Access',
delete_action=(url_for('ccpo.remove_access', user_id=user.id)),
form=form,
confirmation_text='remove'
)
}}
{% endcall %}
{% endfor %}
{% endif %}
</div>
{% if user_can(permissions.CREATE_CCPO_USER) %}
<a class="icon-link" href="{{ url_for('ccpo.add_new_user')}}">
{{ "ccpo.add_user" | translate }} {{ Icon("plus") }}
</a>
{% endif %}
{% if user_can(permissions.DELETE_CCPO_USER) %}
{% for user, form in users_info %}
{% set modal_id = "disable_ccpo_user_{}".format(user.dod_id) %}
{% call Modal(name=modal_id) %}
<h1>Disable CCPO User</h1>
<hr>
{{
Alert(
title=("components.modal.destructive_title" | translate),
message=("ccpo.disable_user.alert_message" | translate({"user_name": user.full_name})),
level="warning"
)
}}
{{
DeleteConfirmation(
modal_id=modal_id,
delete_text='Remove Access',
delete_action=(url_for('ccpo.remove_access', user_id=user.id)),
form=form,
confirmation_text='remove'
)
}}
{% endcall %}
{% endfor %}
{% endif %}
{% endblock %}

2
templates/error.html
View File

@ -5,6 +5,7 @@
{% block content %}
<main class="usa-section usa-content error-page">
<div class="panel">
<div class="panel__heading">
{{ Icon('cloud', classes="icon--red icon--large")}}
<hr>
@ -17,6 +18,7 @@
{%- endif %}
</p>
</div>
</div>
</main>
{% endblock %}

35
templates/error_base.html
View File

@ -10,29 +10,30 @@
<link rel="icon" type="image/x-icon" href="{{ url_for('static', filename='img/favicon.ico') }} " />
</head>
<body class="{% if g.modalOpen %} modalOpen{% endif %}">
<div id='app-root'>
{% block template_vars %}{% endblock %}
{% block template_vars %}{% endblock %}
{% include 'components/usa_header.html' %}
{% include 'components/usa_header.html' %}
{% include 'navigation/topbar.html' %}
{% include 'navigation/topbar.html' %}
<div class='global-layout'>
<div class='global-layout'>
<div class='global-panel-container'>
{% block sidenav %}{% endblock %}
<div class='global-panel-container'>
{% block sidenav %}{% endblock %}
{% block content %}
these are not the droids you are looking for
{% endblock %}
{% block content %}
these are not the droids you are looking for
{% endblock %}
</div>
</div>
{% include 'footer.html' %}
{% block modal %}{% endblock %}
{% assets "js_all" %}
<script src="{{ ASSET_URL }}"></script>
{% endassets %}
</div>
{% include 'footer.html' %}
{% block modal %}{% endblock %}
{% assets "js_all" %}
<script src="{{ ASSET_URL }}"></script>
{% endassets %}
</body>
</html>

20
templates/portfolios/new/step_1.html
View File

@ -15,6 +15,7 @@
<p>{{ "portfolios.header" | translate }}</p>
<h1>{{ 'portfolios.new.title' | translate }}</h1>
</div>
</div>
{{ StickyCTA(text="portfolios.new.cta_step_1"|translate, context=("portfolios.new.sticky_header_context"|translate({"step": "1"}) )) }}
<base-form inline-template>
<div class="row">
@ -38,13 +39,18 @@
{{ "forms.portfolio.defense_component.help_text" | translate | safe }}
</div>
</div>
<div class='action-group-footer'>
{% block next_button %}
{{ SaveButton(text=('portfolios.new.save' | translate), form="portfolio-create", element="input") }}
{% endblock %}
<a class="usa-button usa-button-secondary" href="{{ url_for('atst.home') }}">
Cancel
</a>
<div
class='action-group-footer'
v-bind:class="{'action-group-footer--expand-offset': this.$root.sidenavExpanded, 'action-group-footer--collapse-offset': !this.$root.sidenavExpanded}">
<div class="action-group-footer--container">
{% block next_button %}
{{ SaveButton(text=('portfolios.new.save' | translate), form="portfolio-create", element="input") }}
{% endblock %}
<a class="usa-button usa-button-secondary" href="{{ url_for('atst.home') }}">
Cancel
</a>
</div>
</div>
</form>
</div>
</base-form>

10
templates/task_orders/builder_base.html
View File

@ -31,7 +31,10 @@
<div class="task-order">
{% block to_builder_form_field %}{% endblock %}
</div>
<span class="action-group-footer">
<div
class="action-group-footer"
v-bind:class="{'action-group-footer--expand-offset': this.$root.sidenavExpanded, 'action-group-footer--collapse-offset': !this.$root.sidenavExpanded}">
<div class="action-group-footer--container">
{% block next_button %}
<input
type="submit"
@ -58,14 +61,13 @@
</a>
{%- endif %}
{% endif %}
<a
v-on:click="openModal('cancel')"
class="action-group__action icon-link">
{{ "common.cancel" | translate }}
</a>
</span>
</div>
</div>
</form>
</to-form>
{% endblock %}

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

@ -2,6 +2,8 @@ import json
from unittest.mock import Mock, patch
from uuid import uuid4
import pendulum
import pydantic
import pytest
from tests.factories import ApplicationFactory, EnvironmentFactory
from tests.mock_azure import AUTH_CREDENTIALS, mock_azure
@ -20,10 +22,16 @@ from atst.domain.csp.cloud.models import (
BillingProfileTenantAccessCSPResult,
BillingProfileVerificationCSPPayload,
BillingProfileVerificationCSPResult,
CostManagementQueryCSPResult,
EnvironmentCSPPayload,
EnvironmentCSPResult,
PrincipalAdminRoleCSPPayload,
PrincipalAdminRoleCSPResult,
ProductPurchaseCSPPayload,
ProductPurchaseCSPResult,
ProductPurchaseVerificationCSPPayload,
ProductPurchaseVerificationCSPResult,
ReportingCSPPayload,
SubscriptionCreationCSPPayload,
SubscriptionCreationCSPResult,
SubscriptionVerificationCSPPayload,
@ -44,6 +52,7 @@ from atst.domain.csp.cloud.models import (
TenantPrincipalCSPResult,
TenantPrincipalOwnershipCSPPayload,
TenantPrincipalOwnershipCSPResult,
UserCSPPayload,
)
BILLING_ACCOUNT_NAME = "52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31"
@ -57,12 +66,14 @@ def mock_management_group_create(mock_azure, spec_dict):
def test_create_environment_succeeds(mock_azure: AzureCloudProvider):
environment = EnvironmentFactory.create()
mock_management_group_create(mock_azure, {"id": "Test Id"})
result = mock_azure.create_environment(
AUTH_CREDENTIALS, environment.creator, environment
mock_azure = mock_get_secret(mock_azure)
payload = EnvironmentCSPPayload(
tenant_id="1234", display_name=environment.name, parent_id=str(uuid4())
)
result = mock_azure.create_environment(payload)
assert result.id == "Test Id"
@ -97,20 +108,6 @@ def test_create_application_succeeds(mock_azure: AzureCloudProvider):
assert result.id == "Test Id"
def test_create_atat_admin_user_succeeds(mock_azure: AzureCloudProvider):
environment_id = str(uuid4())
csp_user_id = str(uuid4)
mock_azure.sdk.graphrbac.GraphRbacManagementClient.return_value.service_principals.create.return_value.object_id = (
csp_user_id
)
result = mock_azure.create_atat_admin_user(AUTH_CREDENTIALS, environment_id)
assert result.get("csp_user_id") == csp_user_id
def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
subscription_id = str(uuid4())
management_group_id = str(uuid4())
@ -162,6 +159,27 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
assert body.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
def test_create_tenant_fails(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.json.return_value = {"error": "body"}
mock_result.status_code = 403
mock_azure.sdk.requests.post.return_value = mock_result
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",
)
)
mock_azure = mock_get_secret(mock_azure)
result = mock_azure.create_tenant(payload)
assert result.get("status") == "error"
def test_create_billing_profile_creation(mock_azure: AzureCloudProvider):
mock_azure.sdk.adal.AuthenticationContext.return_value.context.acquire_token_with_client_credentials.return_value = {
"accessToken": "TOKEN"
@ -573,10 +591,10 @@ def test_create_tenant_principal_credential(mock_azure: AzureCloudProvider):
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"
"_get_tenant_admin_token",
wraps=mock_azure._get_tenant_admin_token,
) as get_tenant_admin_token:
get_tenant_admin_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
@ -657,6 +675,35 @@ def test_create_tenant_principal_ownership(mock_azure: AzureCloudProvider):
assert result.principal_owner_assignment_id == "id"
def test_create_principal_admin_role(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_tenant_admin_token",
wraps=mock_azure._get_tenant_admin_token,
) as get_tenant_admin_token:
get_tenant_admin_token.return_value = "my fake token"
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"id": "id"}
mock_azure.sdk.requests.post.return_value = mock_result
payload = PrincipalAdminRoleCSPPayload(
**{
"tenant_id": uuid4().hex,
"principal_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4",
"admin_role_def_id": uuid4().hex,
}
)
result: PrincipalAdminRoleCSPResult = mock_azure.create_principal_admin_role(
payload
)
assert result.principal_assignment_id == "id"
def test_create_subscription_creation(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
@ -718,3 +765,176 @@ 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,
)
)
def test_get_secret(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_client_secret_credential_obj",
wraps=mock_azure._get_client_secret_credential_obj,
) as _get_client_secret_credential_obj:
_get_client_secret_credential_obj.return_value = {}
mock_azure.sdk.secrets.SecretClient.return_value.get_secret.return_value.value = (
"my secret"
)
assert mock_azure.get_secret("secret key") == "my secret"
def test_set_secret(mock_azure: AzureCloudProvider):
with patch.object(
AzureCloudProvider,
"_get_client_secret_credential_obj",
wraps=mock_azure._get_client_secret_credential_obj,
) as _get_client_secret_credential_obj:
_get_client_secret_credential_obj.return_value = {}
mock_azure.sdk.secrets.SecretClient.return_value.set_secret.return_value = (
"my secret"
)
assert mock_azure.set_secret("secret key", "secret_value") == "my secret"
def test_create_active_directory_user(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.ok = True
mock_result.json.return_value = {"id": "id"}
mock_azure.sdk.requests.post.return_value = mock_result
payload = UserCSPPayload(
tenant_id=uuid4().hex,
display_name="Test Testerson",
tenant_host_name="testtenant",
email="test@testerson.test",
password="asdfghjkl", # pragma: allowlist secret
)
result = mock_azure._create_active_directory_user("token", payload)
assert result.id == "id"
def test_update_active_directory_user_email(mock_azure: AzureCloudProvider):
mock_result = Mock()
mock_result.ok = True
mock_azure.sdk.requests.patch.return_value = mock_result
payload = UserCSPPayload(
tenant_id=uuid4().hex,
display_name="Test Testerson",
tenant_host_name="testtenant",
email="test@testerson.test",
password="asdfghjkl", # pragma: allowlist secret
)
result = mock_azure._update_active_directory_user_email(
"token", uuid4().hex, payload
)
assert result
def test_create_user(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"
mock_result_create = Mock()
mock_result_create.ok = True
mock_result_create.json.return_value = {"id": "id"}
mock_azure.sdk.requests.post.return_value = mock_result_create
mock_result_update = Mock()
mock_result_update.ok = True
mock_azure.sdk.requests.patch.return_value = mock_result_update
payload = UserCSPPayload(
tenant_id=uuid4().hex,
display_name="Test Testerson",
tenant_host_name="testtenant",
email="test@testerson.test",
password="asdfghjkl", # pragma: allowlist secret
)
result = mock_azure.create_user(payload)
assert result.id == "id"

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

@ -1,6 +1,7 @@
import pytest
from atst.domain.csp import MockCloudProvider
from atst.domain.csp.cloud.models import EnvironmentCSPPayload, EnvironmentCSPResult
from tests.factories import EnvironmentFactory, EnvironmentRoleFactory, UserFactory
@ -14,20 +15,17 @@ def mock_csp():
def test_create_environment(mock_csp: MockCloudProvider):
environment = EnvironmentFactory.create()
user = UserFactory.create()
environment_id = mock_csp.create_environment(CREDENTIALS, user, environment)
assert isinstance(environment_id, str)
def test_create_admin_user(mock_csp: MockCloudProvider):
admin_user = mock_csp.create_atat_admin_user(CREDENTIALS, "env_id")
assert isinstance(admin_user["id"], str)
assert isinstance(admin_user["credentials"], dict)
def test_create_environment_baseline(mock_csp: MockCloudProvider):
baseline = mock_csp.create_atat_admin_user(CREDENTIALS, "env_id")
assert isinstance(baseline, dict)
environment.application.cloud_id = "parent_id"
environment.application.portfolio.csp_data = {"tenant_id": "fake"}
payload = EnvironmentCSPPayload(
**dict(
tenant_id=environment.application.portfolio.csp_data.get("tenant_id"),
display_name=environment.name,
parent_id=environment.application.cloud_id,
)
)
result = mock_csp.create_environment(payload)
assert isinstance(result, EnvironmentCSPResult)
def test_create_or_update_user(mock_csp: MockCloudProvider):

10
tests/domain/test_environment_roles.py
View File

@ -93,27 +93,25 @@ def test_disable_completed(application_role, environment):
def test_disable_checks_env_provisioning_status(session):
environment = EnvironmentFactory.create()
assert environment.is_pending
assert not environment.cloud_id
env_role1 = EnvironmentRoleFactory.create(environment=environment)
env_role1 = EnvironmentRoles.disable(env_role1.id)
assert env_role1.disabled
environment.cloud_id = "cloud-id"
environment.root_user_info = {"credentials": "credentials"}
session.add(environment)
session.commit()
session.refresh(environment)
assert not environment.is_pending
assert environment.cloud_id
env_role2 = EnvironmentRoleFactory.create(environment=environment)
env_role2 = EnvironmentRoles.disable(env_role2.id)
assert env_role2.disabled
def test_disable_checks_env_role_provisioning_status():
environment = EnvironmentFactory.create(
cloud_id="cloud-id", root_user_info={"credentials": "credentials"}
)
environment = EnvironmentFactory.create(cloud_id="cloud-id")
environment.application.portfolio.csp_data = {"tenant_id": uuid4().hex}
env_role1 = EnvironmentRoleFactory.create(environment=environment)
assert not env_role1.csp_user_id
env_role1 = EnvironmentRoles.disable(env_role1.id)

75
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)])
@ -168,37 +134,16 @@ class TestGetEnvironmentsPendingCreate(EnvQueryTest):
self.create_portfolio_with_clins([(self.TOMORROW, self.TOMORROW)])
assert len(Environments.get_environments_pending_creation(self.NOW)) == 0
def test_with_already_provisioned_app(self, session):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], app_data={"cloud_id": uuid4().hex}
)
assert len(Environments.get_environments_pending_creation(self.NOW)) == 1
def test_with_already_provisioned_env(self, session):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], env_data={"cloud_id": uuid4().hex}
[(self.YESTERDAY, self.TOMORROW)],
env_data={"cloud_id": uuid4().hex},
app_data={"cloud_id": uuid4().hex},
)
assert len(Environments.get_environments_pending_creation(self.NOW)) == 0
class TestGetEnvironmentsPendingAtatUserCreation(EnvQueryTest):
def test_with_provisioned_environment(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)],
{"cloud_id": uuid4().hex, "root_user_info": {}},
)
assert (
len(Environments.get_environments_pending_atat_user_creation(self.NOW)) == 0
)
def test_with_unprovisioned_environment(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)],
{"cloud_id": uuid4().hex, "root_user_info": None},
)
assert (
len(Environments.get_environments_pending_atat_user_creation(self.NOW)) == 1
)
def test_with_unprovisioned_expired_clins_environment(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.YESTERDAY)],
{"cloud_id": uuid4().hex, "root_user_info": None},
)
assert (
len(Environments.get_environments_pending_atat_user_creation(self.NOW)) == 0
)

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

22
tests/models/test_environments.py
View File

@ -51,28 +51,6 @@ def test_audit_event_for_environment_deletion(session):
assert after
@pytest.mark.parametrize(
"env_data,expected_status",
[
[
{"cloud_id": None, "root_user_info": None},
Environment.ProvisioningStatus.PENDING,
],
[
{"cloud_id": 1, "root_user_info": None},
Environment.ProvisioningStatus.PENDING,
],
[
{"cloud_id": 1, "root_user_info": {}},
Environment.ProvisioningStatus.COMPLETED,
],
],
)
def test_environment_provisioning_status(env_data, expected_status):
environment = EnvironmentFactory.create(**env_data)
assert environment.provisioning_status == expected_status
def test_environment_roles_do_not_include_deleted():
member_list = [
{"role_name": CSPRole.ADMIN},

61
tests/test_jobs.py
View File

@ -12,14 +12,12 @@ from atst.jobs import (
dispatch_create_environment,
dispatch_create_application,
dispatch_create_user,
dispatch_create_atat_admin_user,
dispatch_provision_portfolio,
create_environment,
do_create_user,
do_provision_portfolio,
do_create_environment,
do_create_application,
do_create_atat_admin_user,
)
from tests.factories import (
EnvironmentFactory,
@ -94,6 +92,10 @@ tomorrow = now.add(days=1)
def test_create_environment_job(session, csp):
environment = EnvironmentFactory.create()
environment.application.cloud_id = "parentId"
environment.application.portfolio.csp_data = {"tenant_id": "fake"}
session.add(environment)
session.commit()
do_create_environment(csp, environment.id)
session.refresh(environment)
@ -149,19 +151,11 @@ def test_create_user_job(session, csp):
assert app_role.cloud_id
def test_create_atat_admin_user(csp, session):
environment = EnvironmentFactory.create(cloud_id="something")
do_create_atat_admin_user(csp, environment.id)
session.refresh(environment)
assert environment.root_user_info
def test_dispatch_create_environment(session, monkeypatch):
# Given that I have a portfolio with an active CLIN and two environments,
# one of which is deleted
portfolio = PortfolioFactory.create(
applications=[{"environments": [{}, {}]}],
applications=[{"environments": [{}, {}], "cloud_id": uuid4().hex}],
task_orders=[
{
"create_clins": [
@ -227,36 +221,9 @@ def test_dispatch_create_user(monkeypatch):
mock.delay.assert_called_once_with(application_role_ids=[app_role.id])
def test_dispatch_create_atat_admin_user(session, monkeypatch):
portfolio = PortfolioFactory.create(
applications=[
{"environments": [{"cloud_id": uuid4().hex, "root_user_info": None}]}
],
task_orders=[
{
"create_clins": [
{
"start_date": pendulum.now().subtract(days=1),
"end_date": pendulum.now().add(days=1),
}
]
}
],
)
mock = Mock()
monkeypatch.setattr("atst.jobs.create_atat_admin_user", mock)
environment = portfolio.applications[0].environments[0]
dispatch_create_atat_admin_user.run()
mock.delay.assert_called_once_with(environment_id=environment.id)
def test_create_environment_no_dupes(session, celery_app, celery_worker):
portfolio = PortfolioFactory.create(
applications=[
{"environments": [{"cloud_id": uuid4().hex, "root_user_info": {}}]}
],
applications=[{"environments": [{"cloud_id": uuid4().hex}]}],
task_orders=[
{
"create_clins": [
@ -286,9 +253,19 @@ def test_create_environment_no_dupes(session, celery_app, celery_worker):
assert environment.claimed_until == None
def test_dispatch_provision_portfolio(
csp, session, portfolio, celery_app, celery_worker, monkeypatch
):
def test_dispatch_provision_portfolio(csp, monkeypatch):
portfolio = PortfolioFactory.create(
task_orders=[
{
"create_clins": [
{
"start_date": pendulum.now().subtract(days=1),
"end_date": pendulum.now().add(days=1),
}
]
}
],
)
sm = PortfolioStateMachineFactory.create(portfolio=portfolio)
mock = Mock()
monkeypatch.setattr("atst.jobs.provision_portfolio", mock)

46
tests/utils.py
View File

@ -5,9 +5,12 @@ from unittest.mock import Mock
from OpenSSL import crypto
from cryptography.hazmat.backends import default_backend
from flask import template_rendered
import pendulum
from atst.utils.notification_sender import NotificationSender
import tests.factories as factories
@contextmanager
def captured_templates(app):
@ -62,3 +65,46 @@ def make_crl_list(x509_obj, x509_path):
issuer = x509_obj.issuer.public_bytes(default_backend())
filename = os.path.basename(x509_path)
return [(filename, issuer.hex())]
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,
app_data=None,
state_machine_status=None,
):
env_data = env_data or {}
app_data = app_data or {}
return factories.PortfolioFactory.create(
state=state_machine_status,
applications=[
{
"name": "Mos Eisley",
"description": "Where Han shot first",
"environments": [{"name": "thebar", **env_data}],
**app_data,
}
],
task_orders=[
{
"create_clins": [
{"start_date": start_date, "end_date": end_date}
for (start_date, end_date) in start_and_end_dates
]
}
],
)

6
translations.yaml
View File

@ -128,9 +128,6 @@ flash:
message: There was an error processing the invitation for {user_name} from {application_name}
resent:
message: "{email} has been sent an invitation to access this Application"
revoked:
title: Application invitation revoked
message: You have successfully revoked the invite for {user_name} from {application_name}
application_member:
removed:
title: Team member removed from application
@ -166,6 +163,9 @@ flash:
errors:
title: There were some errors
message: Please see below.
invite_revoked:
title: "{resource} invitation revoked"
message: "You have successfully revoked the invite for {user_name} from {resource_name}"
login_required_message: After you log in, you will be redirected to your destination page.
login_required_title: Log in required
logged_out: