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

Merge pull request #1062 from dod-ccpo/env-provisioning-task

Browse Source 
Environment provisioning background jobs
This commit is contained in:
richard-dds
2019-09-16 09:58:18 -04:00
committed by GitHub
parent 1fe755bb69 6100b6e479
commit 79c8e4fc63
22 changed files with 613 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/domain/test_applications.py
View File

@@ -21,7 +21,7 @@ from tests.factories import (
def test_create_application_with_multiple_environments():
portfolio = PortfolioFactory.create()
application = Applications.create(
portfolio, "My Test Application", "Test", ["dev", "prod"]
portfolio.owner, portfolio, "My Test Application", "Test", ["dev", "prod"]
)
assert application.portfolio == portfolio

132
tests/domain/test_environments.py
View File

@@ -1,4 +1,6 @@
import pytest
import pendulum
from uuid import uuid4
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
@@ -7,7 +9,6 @@ from atst.models.environment_role import CSPRole
from tests.factories import (
ApplicationFactory,
UserFactory,
PortfolioFactory,
EnvironmentFactory,
EnvironmentRoleFactory,
@@ -15,12 +16,13 @@ from tests.factories import (
)
@pytest.mark.skip(reason="Reinstate and update once jobs api is up")
def test_create_environments():
application = ApplicationFactory.create()
environments = Environments.create_many(application, ["Staging", "Production"])
environments = Environments.create_many(
application.portfolio.owner, application, ["Staging", "Production"]
)
for env in environments:
assert env.cloud_id is not None
assert env.cloud_id is None
def test_update_env_role():
@@ -87,3 +89,125 @@ def test_update_environment():
assert environment.name is not "name 2"
Environments.update(environment, name="name 2")
assert environment.name == "name 2"
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)])
assert len(Environments.get_environments_pending_creation(self.NOW)) == 0
def test_with_active_clins(self, session):
portfolio = self.create_portfolio_with_clins([(self.YESTERDAY, self.TOMORROW)])
Environments.get_environments_pending_creation(self.NOW) == [
portfolio.applications[0].environments[0].id
]
def test_with_future_clins(self, session):
self.create_portfolio_with_clins([(self.TOMORROW, self.TOMORROW)])
assert len(Environments.get_environments_pending_creation(self.NOW)) == 0
def test_with_already_provisioned_env(self, session):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)], env_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
)
class TestGetEnvironmentsPendingBaselineCreation(EnvQueryTest):
def test_with_provisioned_environment(self):
self.create_portfolio_with_clins(
[(self.YESTERDAY, self.TOMORROW)],
{
"cloud_id": uuid4().hex,
"root_user_info": {"foo": "bar"},
"baseline_info": {"foo": "bar"},
},
)
assert (
len(Environments.get_environments_pending_baseline_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": {"foo": "bar"},
"baseline_info": None,
},
)
assert (
len(Environments.get_environments_pending_baseline_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": {"foo": "bar"}},
)
assert (
len(Environments.get_environments_pending_baseline_creation(self.NOW)) == 0
)

18
tests/domain/test_portfolios.py
View File

@@ -71,7 +71,11 @@ def test_update_portfolio_role_role(portfolio, portfolio_owner):
def test_scoped_portfolio_for_admin_missing_view_apps_perms(portfolio_owner, portfolio):
Applications.create(
portfolio, "My Application 2", "My application 2", ["dev", "staging", "prod"]
portfolio.owner,
portfolio,
"My Application 2",
"My application 2",
["dev", "staging", "prod"],
)
restricted_admin = UserFactory.create()
PortfolioRoleFactory.create(
@@ -90,7 +94,11 @@ def test_scoped_portfolio_returns_all_applications_for_portfolio_admin(
):
for _ in range(5):
Applications.create(
portfolio, "My Application", "My application", ["dev", "staging", "prod"]
portfolio.owner,
portfolio,
"My Application",
"My application",
["dev", "staging", "prod"],
)
admin = UserFactory.create()
@@ -109,7 +117,11 @@ def test_scoped_portfolio_returns_all_applications_for_portfolio_owner(
):
for _ in range(5):
Applications.create(
portfolio, "My Application", "My application", ["dev", "staging", "prod"]
portfolio.owner,
portfolio,
"My Application",
"My application",
["dev", "staging", "prod"],
)
scoped_portfolio = Portfolios.get(portfolio_owner, portfolio.id)

6
tests/domain/test_task_orders.py
View File

@@ -139,7 +139,7 @@ def test_update_adds_clins():
def test_update_does_not_duplicate_clins():
task_order = TaskOrderFactory.create(
number="3453453456", create_clins=["123", "456"]
number="3453453456", create_clins=[{"number": "123"}, {"number": "456"}]
)
clins = [
{
@@ -171,7 +171,9 @@ def test_update_does_not_duplicate_clins():
def test_delete_task_order_with_clins(session):
task_order = TaskOrderFactory.create(create_clins=[1, 2, 3])
task_order = TaskOrderFactory.create(
create_clins=[{"number": 1}, {"number": 2}, {"number": 3}]
)
TaskOrders.delete(task_order.id)
assert not session.query(