150 lines
5.1 KiB
Python
150 lines
5.1 KiB
Python
import pytest
|
|
from uuid import uuid4
|
|
|
|
from atat.domain.environments import Environments
|
|
from atat.domain.environment_roles import EnvironmentRoles
|
|
from atat.domain.exceptions import AlreadyExistsError, DisabledError, NotFoundError
|
|
from atat.models.environment_role import CSPRole, EnvironmentRole
|
|
|
|
from tests.factories import (
|
|
ApplicationFactory,
|
|
PortfolioFactory,
|
|
EnvironmentFactory,
|
|
EnvironmentRoleFactory,
|
|
ApplicationRoleFactory,
|
|
)
|
|
from tests.utils import EnvQueryTest
|
|
|
|
|
|
def test_create_environments():
|
|
application = ApplicationFactory.create()
|
|
environments = Environments.create_many(
|
|
application.portfolio.owner, application, ["Staging", "Production"]
|
|
)
|
|
for env in environments:
|
|
assert env.cloud_id is None
|
|
|
|
|
|
def test_update_env_role():
|
|
env_role = EnvironmentRoleFactory.create(role=CSPRole.ADMIN)
|
|
new_role = CSPRole.BILLING_READ
|
|
Environments.update_env_role(
|
|
env_role.environment, env_role.application_role, new_role
|
|
)
|
|
assert env_role.role == new_role
|
|
|
|
|
|
def test_update_env_role_no_access():
|
|
env_role = EnvironmentRoleFactory.create(role=CSPRole.ADMIN)
|
|
Environments.update_env_role(env_role.environment, env_role.application_role, None)
|
|
|
|
assert not EnvironmentRoles.get(
|
|
env_role.application_role.id, env_role.environment.id
|
|
)
|
|
assert env_role.role is None
|
|
assert env_role.disabled
|
|
|
|
|
|
def test_update_env_role_disabled_role():
|
|
env_role = EnvironmentRoleFactory.create(role=CSPRole.ADMIN)
|
|
Environments.update_env_role(env_role.environment, env_role.application_role, None)
|
|
|
|
# An exception should be raised when a new role is passed to Environments.update_env_role
|
|
with pytest.raises(DisabledError):
|
|
Environments.update_env_role(
|
|
env_role.environment, env_role.application_role, CSPRole.BILLING_READ,
|
|
)
|
|
|
|
assert env_role.role is None
|
|
assert env_role.disabled
|
|
|
|
# An exception should not be raised when no new role is passed
|
|
Environments.update_env_role(env_role.environment, env_role.application_role, None)
|
|
|
|
|
|
def test_get_excludes_deleted():
|
|
env = EnvironmentFactory.create(
|
|
deleted=True, application=ApplicationFactory.create()
|
|
)
|
|
with pytest.raises(NotFoundError):
|
|
Environments.get(env.id)
|
|
|
|
|
|
def test_delete_environment(session):
|
|
env = EnvironmentFactory.create(application=ApplicationFactory.create())
|
|
env_role = EnvironmentRoleFactory.create(
|
|
application_role=ApplicationRoleFactory.create(application=env.application),
|
|
environment=env,
|
|
)
|
|
assert not env.deleted
|
|
assert not env_role.deleted
|
|
Environments.delete(env)
|
|
assert env.deleted
|
|
assert env_role.deleted
|
|
# did not flush
|
|
assert session.dirty
|
|
|
|
Environments.delete(env, commit=True)
|
|
assert env.deleted
|
|
assert env_role.deleted
|
|
# flushed the change
|
|
assert not session.dirty
|
|
|
|
|
|
def test_update_environment():
|
|
environment = EnvironmentFactory.create()
|
|
assert environment.name is not "name 2"
|
|
Environments.update(environment, name="name 2")
|
|
assert environment.name == "name 2"
|
|
|
|
|
|
def test_create_does_not_duplicate_names_within_application():
|
|
application = ApplicationFactory.create()
|
|
name = "Your Environment"
|
|
user = application.portfolio.owner
|
|
|
|
assert Environments.create(user, application, name)
|
|
with pytest.raises(AlreadyExistsError):
|
|
Environments.create(user, application, name)
|
|
|
|
|
|
def test_update_does_not_duplicate_names_within_application():
|
|
application = ApplicationFactory.create()
|
|
name = "Your Environment"
|
|
environment = EnvironmentFactory.create(application=application, name=name)
|
|
dupe_env = EnvironmentFactory.create(application=application)
|
|
user = application.portfolio.owner
|
|
|
|
with pytest.raises(AlreadyExistsError):
|
|
Environments.update(dupe_env, name)
|
|
|
|
|
|
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_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},
|
|
app_data={"cloud_id": uuid4().hex},
|
|
)
|
|
assert len(Environments.get_environments_pending_creation(self.NOW)) == 0
|