diff --git a/atst/domain/environments.py b/atst/domain/environments.py index 5e7102b1..69ba8ab3 100644 --- a/atst/domain/environments.py +++ b/atst/domain/environments.py @@ -6,6 +6,7 @@ from atst.models.environment import Environment from atst.models.environment_role import EnvironmentRole from atst.models.application import Application from atst.domain.environment_roles import EnvironmentRoles +from atst.domain.users import Users from .exceptions import NotFoundError @@ -62,39 +63,51 @@ class Environments(object): return env @classmethod - def update_environment_roles(cls, portfolio_role, ids_and_roles): + def update_env_role(cls, environment, user, new_role): updated = False - for id_and_role in ids_and_roles: - new_role = id_and_role["role"] - environment = Environments.get(id_and_role["id"]) - - if new_role is None: - role_deleted = EnvironmentRoles.delete( - portfolio_role.user.id, environment.id + if new_role is None or new_role == "No access": + role_deleted = EnvironmentRoles.delete(user.id, environment.id) + if role_deleted: + updated = True + else: + env_role = EnvironmentRoles.get(user.id, environment.id) + if env_role and env_role.role != new_role: + env_role.role = new_role + updated = True + db.session.add(env_role) + elif not env_role: + env_role = EnvironmentRoles.create( + user=user, environment=environment, role=new_role ) - if role_deleted: - updated = True - else: - env_role = EnvironmentRoles.get( - portfolio_role.user.id, id_and_role["id"] - ) - if env_role and env_role.role != new_role: - env_role.role = new_role - updated = True - db.session.add(env_role) - elif not env_role: - env_role = EnvironmentRoles.create( - user=portfolio_role.user, environment=environment, role=new_role - ) - updated = True - db.session.add(env_role) + updated = True + db.session.add(env_role) if updated: db.session.commit() return updated + @classmethod + def update_env_roles_by_environment(cls, environment_id, team_roles): + environment = Environments.get(environment_id) + + for member in team_roles: + new_role = member["role"] + user = Users.get(member["user_id"]) + Environments.update_env_role( + environment=environment, user=user, new_role=new_role + ) + + @classmethod + def update_env_roles_by_member(cls, member, env_roles): + for env_roles in env_roles: + new_role = env_roles["role"] + environment = Environments.get(env_roles["id"]) + Environments.update_env_role( + environment=environment, user=member, new_role=new_role + ) + @classmethod def revoke_access(cls, environment, target_user): EnvironmentRoles.delete(environment.id, target_user.id) diff --git a/tests/domain/test_environments.py b/tests/domain/test_environments.py index b572a7e5..18027fd8 100644 --- a/tests/domain/test_environments.py +++ b/tests/domain/test_environments.py @@ -1,9 +1,11 @@ import pytest +import random from atst.domain.environments import Environments from atst.domain.environment_roles import EnvironmentRoles from atst.domain.portfolio_roles import PortfolioRoles from atst.domain.exceptions import NotFoundError +from atst.models.environment_role import CSPRole from tests.factories import ( ApplicationFactory, @@ -21,143 +23,111 @@ def test_create_environments(): assert env.cloud_id is not None -def test_create_environment_role_creates_cloud_id(session): - owner = UserFactory.create() - developer = UserFactory.create() +def test_update_env_role(): + env_role = EnvironmentRoleFactory.create(role=CSPRole.BASIC_ACCESS.value) + new_role = CSPRole.TECHNICAL_READ.value - portfolio = PortfolioFactory.create( - owner=owner, - members=[{"user": developer, "role_name": "developer"}], - applications=[ - {"name": "application1", "environments": [{"name": "application1 prod"}]} - ], + assert Environments.update_env_role(env_role.environment, env_role.user, new_role) + assert env_role.role == new_role + + +def test_update_env_role_no_access(): + env_role = EnvironmentRoleFactory.create(role=CSPRole.BASIC_ACCESS.value) + + assert Environments.update_env_role( + env_role.environment, env_role.user, "No access" + ) + assert not EnvironmentRoles.get(env_role.user.id, env_role.environment.id) + + +def test_update_env_role_no_change(): + env_role = EnvironmentRoleFactory.create(role=CSPRole.BASIC_ACCESS.value) + new_role = CSPRole.BASIC_ACCESS.value + + assert not Environments.update_env_role( + env_role.environment, env_role.user, new_role ) - env = portfolio.applications[0].environments[0] - new_role = [{"id": env.id, "role": "developer"}] - portfolio_role = portfolio.members[0] - assert not portfolio_role.user.cloud_id - assert Environments.update_environment_roles(portfolio_role, new_role) - - assert portfolio_role.user.cloud_id is not None +def test_update_env_role_creates_cloud_id_for_new_member(session): + user = UserFactory.create() + env = EnvironmentFactory.create() + assert not user.cloud_id + assert Environments.update_env_role(env, user, CSPRole.TECHNICAL_READ.value) + assert EnvironmentRoles.get(user.id, env.id) + assert user.cloud_id is not None -def test_update_environment_roles(): - owner = UserFactory.create() - developer = UserFactory.create() - - portfolio = PortfolioFactory.create( - owner=owner, - members=[{"user": developer, "role_name": "developer"}], - applications=[ - { - "name": "application1", - "environments": [ - { - "name": "application1 dev", - "members": [{"user": developer, "role_name": "devlops"}], - }, - { - "name": "application1 staging", - "members": [{"user": developer, "role_name": "developer"}], - }, - {"name": "application1 prod"}, - ], - } - ], +def test_update_env_roles_by_environment(): + environment = EnvironmentFactory.create() + env_role_1 = EnvironmentRoleFactory.create( + environment=environment, role=CSPRole.BASIC_ACCESS.value + ) + env_role_2 = EnvironmentRoleFactory.create( + environment=environment, role=CSPRole.NETWORK_ADMIN.value + ) + env_role_3 = EnvironmentRoleFactory.create( + environment=environment, role=CSPRole.TECHNICAL_READ.value ) - dev_env = portfolio.applications[0].environments[0] - staging_env = portfolio.applications[0].environments[1] - new_ids_and_roles = [ - {"id": dev_env.id, "role": "billing_admin"}, - {"id": staging_env.id, "role": "developer"}, + team_roles = [ + { + "user_id": env_role_1.user.id, + "name": env_role_1.user.full_name, + "role": CSPRole.BUSINESS_READ.value, + }, + { + "user_id": env_role_2.user.id, + "name": env_role_2.user.full_name, + "role": CSPRole.NETWORK_ADMIN.value, + }, + { + "user_id": env_role_3.user.id, + "name": env_role_3.user.full_name, + "role": "No access", + }, ] - portfolio_role = portfolio.members[0] - assert Environments.update_environment_roles(portfolio_role, new_ids_and_roles) - new_dev_env_role = EnvironmentRoles.get(portfolio_role.user.id, dev_env.id) - staging_env_role = EnvironmentRoles.get(portfolio_role.user.id, staging_env.id) - - assert new_dev_env_role.role == "billing_admin" - assert staging_env_role.role == "developer" + Environments.update_env_roles_by_environment(environment.id, team_roles) + assert env_role_1.role == CSPRole.BUSINESS_READ.value + assert env_role_2.role == CSPRole.NETWORK_ADMIN.value + assert not EnvironmentRoles.get(env_role_3.user.id, environment.id) -def test_remove_environment_role(): - owner = UserFactory.create() - developer = UserFactory.create() - portfolio = PortfolioFactory.create( - owner=owner, - members=[{"user": developer, "role_name": "developer"}], - applications=[ +def test_update_env_roles_by_member(): + user = UserFactory.create() + application = ApplicationFactory.create( + environments=[ { - "name": "application1", - "environments": [ - { - "name": "application1 dev", - "members": [{"user": developer, "role_name": "devops"}], - }, - { - "name": "application1 staging", - "members": [{"user": developer, "role_name": "developer"}], - }, - { - "name": "application1 uat", - "members": [ - {"user": developer, "role_name": "financial_auditor"} - ], - }, - {"name": "application1 prod"}, - ], - } - ], + "name": "dev", + "members": [{"user": user, "role_name": CSPRole.BUSINESS_READ.value}], + }, + { + "name": "staging", + "members": [{"user": user, "role_name": CSPRole.BUSINESS_READ.value}], + }, + {"name": "prod"}, + { + "name": "testing", + "members": [{"user": user, "role_name": CSPRole.BUSINESS_READ.value}], + }, + ] ) - application = portfolio.applications[0] - now_ba = application.environments[0].id - now_none = application.environments[1].id - still_fa = application.environments[2].id - - new_environment_roles = [ - {"id": now_ba, "role": "billing_auditor"}, - {"id": now_none, "role": None}, + dev, staging, prod, testing = application.environments + env_roles = [ + {"id": dev.id, "role": CSPRole.NETWORK_ADMIN.value}, + {"id": staging.id, "role": CSPRole.BUSINESS_READ.value}, + {"id": prod.id, "role": CSPRole.TECHNICAL_READ.value}, + {"id": testing.id, "role": "No access"}, ] - portfolio_role = PortfolioRoles.get(portfolio.id, developer.id) - assert Environments.update_environment_roles(portfolio_role, new_environment_roles) + Environments.update_env_roles_by_member(user, env_roles) - assert portfolio_role.num_environment_roles == 2 - assert EnvironmentRoles.get(developer.id, now_ba).role == "billing_auditor" - assert EnvironmentRoles.get(developer.id, now_none) is None - assert EnvironmentRoles.get(developer.id, still_fa).role == "financial_auditor" - - -def test_no_update_to_environment_roles(): - owner = UserFactory.create() - developer = UserFactory.create() - - portfolio = PortfolioFactory.create( - owner=owner, - members=[{"user": developer, "role_name": "developer"}], - applications=[ - { - "name": "application1", - "environments": [ - { - "name": "application1 dev", - "members": [{"user": developer, "role_name": "devops"}], - } - ], - } - ], - ) - - dev_env = portfolio.applications[0].environments[0] - new_ids_and_roles = [{"id": dev_env.id, "role": "devops"}] - - portfolio_role = PortfolioRoles.get(portfolio.id, developer.id) - assert not Environments.update_environment_roles(portfolio_role, new_ids_and_roles) + assert EnvironmentRoles.get(user.id, dev.id).role == CSPRole.NETWORK_ADMIN.value + assert EnvironmentRoles.get(user.id, staging.id).role == CSPRole.BUSINESS_READ.value + assert EnvironmentRoles.get(user.id, prod.id).role == CSPRole.TECHNICAL_READ.value + assert not EnvironmentRoles.get(user.id, testing.id) def test_get_scoped_environments(db): diff --git a/tests/factories.py b/tests/factories.py index 12b4f62b..a9ac984e 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -194,6 +194,7 @@ class EnvironmentFactory(Base): model = Environment name = factory.Faker("domain_word") + application = factory.SubFactory(ApplicationFactory) @classmethod def _create(cls, model_class, *args, **kwargs): diff --git a/tests/models/test_portfolio_role.py b/tests/models/test_portfolio_role.py index 3d9ac0ec..4b5e6601 100644 --- a/tests/models/test_portfolio_role.py +++ b/tests/models/test_portfolio_role.py @@ -123,8 +123,8 @@ def test_has_env_role_history(session): env_role = EnvironmentRoleFactory.create( user=user, environment=environment, role="developer" ) - Environments.update_environment_roles( - portfolio_role, [{"role": "admin", "id": environment.id}] + Environments.update_env_roles_by_member( + user, [{"role": "admin", "id": environment.id}] ) changed_events = ( session.query(AuditEvent)