Update SQL query to find pending portfolios.
The query to find portfolios that are pending provisioning is updated to check for: - a period of performance that has started - a portfolio state machine that has an UNSTARTED or one of the CREATED states I left several TODOs to ensure that the orchestration functions correctly for portfolio.
This commit is contained in:
@@ -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)])
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user