Delete unused Applications domain method and invitation service
This commit is contained in:
dandds
parent
fa50c01e48
commit
bd4f2b3dde
@ -7,7 +7,6 @@ from atst.domain.environment_roles import EnvironmentRoles
|
|||||||
from atst.domain.environments import Environments
|
from atst.domain.environments import Environments
|
||||||
from atst.domain.exceptions import NotFoundError
|
from atst.domain.exceptions import NotFoundError
|
||||||
from atst.domain.invitations import ApplicationInvitations
|
from atst.domain.invitations import ApplicationInvitations
|
||||||
from atst.domain.users import Users
|
|
||||||
from atst.models import (
|
from atst.models import (
|
||||||
Application,
|
Application,
|
||||||
ApplicationRole,
|
ApplicationRole,
|
||||||
@ -82,35 +81,6 @@ class Applications(BaseDomainClass):
|
|||||||
db.session.add(application)
|
db.session.add(application)
|
||||||
db.session.commit()
|
db.session.commit()
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def create_member(
|
|
||||||
cls, application, user_data, permission_sets=None, environment_roles_data=None
|
|
||||||
):
|
|
||||||
permission_sets = [] if permission_sets is None else permission_sets
|
|
||||||
environment_roles_data = (
|
|
||||||
[] if environment_roles_data is None else environment_roles_data
|
|
||||||
)
|
|
||||||
|
|
||||||
user = Users.get_or_create_by_dod_id(
|
|
||||||
user_data["dod_id"],
|
|
||||||
first_name=user_data["first_name"],
|
|
||||||
last_name=user_data["last_name"],
|
|
||||||
phone_number=user_data.get("phone_number"),
|
|
||||||
email=user_data["email"],
|
|
||||||
)
|
|
||||||
|
|
||||||
application_role = ApplicationRoles.create(user, application, permission_sets)
|
|
||||||
|
|
||||||
for env_role_data in environment_roles_data:
|
|
||||||
role = env_role_data.get("role")
|
|
||||||
if role:
|
|
||||||
environment = Environments.get(env_role_data.get("environment_id"))
|
|
||||||
Environments.add_member(
|
|
||||||
environment, application_role, env_role_data.get("role")
|
|
||||||
)
|
|
||||||
|
|
||||||
return application_role
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def invite(
|
def invite(
|
||||||
cls,
|
cls,
|
||||||
|
|||||||
@ -1,94 +0,0 @@
|
|||||||
from flask import render_template
|
|
||||||
|
|
||||||
from atst.domain.invitations import PortfolioInvitations, ApplicationInvitations
|
|
||||||
from atst.queue import queue
|
|
||||||
from atst.domain.task_orders import TaskOrders
|
|
||||||
from atst.domain.portfolio_roles import PortfolioRoles
|
|
||||||
from atst.models import ApplicationRole, PortfolioRole
|
|
||||||
|
|
||||||
OFFICER_INVITATIONS = {
|
|
||||||
"ko_invite": {
|
|
||||||
"role": "contracting_officer",
|
|
||||||
"subject": "Review a task order",
|
|
||||||
"template": "emails/portfolio/invitation.txt",
|
|
||||||
},
|
|
||||||
"cor_invite": {
|
|
||||||
"role": "contracting_officer_representative",
|
|
||||||
"subject": "Help with a task order",
|
|
||||||
"template": "emails/portfolio/invitation.txt",
|
|
||||||
},
|
|
||||||
"so_invite": {
|
|
||||||
"role": "security_officer",
|
|
||||||
"subject": "Review security for a task order",
|
|
||||||
"template": "emails/portfolio/invitation.txt",
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def update_officer_invitations(user, task_order):
|
|
||||||
for invite_type in dict.keys(OFFICER_INVITATIONS):
|
|
||||||
invite_opts = OFFICER_INVITATIONS[invite_type]
|
|
||||||
|
|
||||||
if getattr(task_order, invite_type) and not getattr(
|
|
||||||
task_order, invite_opts["role"]
|
|
||||||
):
|
|
||||||
officer_data = task_order.officer_dictionary(invite_opts["role"])
|
|
||||||
officer = TaskOrders.add_officer(
|
|
||||||
task_order, invite_opts["role"], officer_data
|
|
||||||
)
|
|
||||||
pf_officer_member = PortfolioRoles.get(task_order.portfolio.id, officer.id)
|
|
||||||
invite_service = Invitation(
|
|
||||||
user,
|
|
||||||
pf_officer_member,
|
|
||||||
officer_data["email"],
|
|
||||||
subject=invite_opts["subject"],
|
|
||||||
email_template=invite_opts["template"],
|
|
||||||
)
|
|
||||||
invite_service.invite()
|
|
||||||
|
|
||||||
|
|
||||||
class Invitation:
|
|
||||||
def __init__(self, inviter, member, email, subject="", email_template=None):
|
|
||||||
self.inviter = inviter
|
|
||||||
self.member = member
|
|
||||||
self.email = email
|
|
||||||
self.subject = subject
|
|
||||||
self.email_template = email_template
|
|
||||||
|
|
||||||
if isinstance(member, PortfolioRole):
|
|
||||||
self.email_template = (
|
|
||||||
self.email_template or "emails/portfolio/invitation.txt"
|
|
||||||
)
|
|
||||||
self.subject = (
|
|
||||||
self.subject or "{} has invited you to a JEDI cloud portfolio"
|
|
||||||
)
|
|
||||||
self.domain_class = PortfolioInvitations
|
|
||||||
elif isinstance(member, ApplicationRole):
|
|
||||||
self.email_template = (
|
|
||||||
self.email_template or "emails/application/invitation.txt"
|
|
||||||
)
|
|
||||||
self.subject = (
|
|
||||||
self.subject or "{} has invited you to a JEDI cloud application"
|
|
||||||
)
|
|
||||||
self.domain_class = ApplicationInvitations
|
|
||||||
|
|
||||||
def invite(self):
|
|
||||||
invite = self._create_invite()
|
|
||||||
self._send_invite_email(invite.token)
|
|
||||||
|
|
||||||
return invite
|
|
||||||
|
|
||||||
def _create_invite(self):
|
|
||||||
user = self.member.user
|
|
||||||
return self.domain_class.create(
|
|
||||||
self.inviter,
|
|
||||||
self.member,
|
|
||||||
{"email": self.email, "dod_id": user.dod_id},
|
|
||||||
commit=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
def _send_invite_email(self, token):
|
|
||||||
body = render_template(
|
|
||||||
self.email_template, owner=self.inviter.full_name, token=token
|
|
||||||
)
|
|
||||||
queue.send_mail([self.email], self.subject.format(self.inviter.full_name), body)
|
|
||||||
@ -107,32 +107,6 @@ def test_delete_application(session):
|
|||||||
assert not session.dirty
|
assert not session.dirty
|
||||||
|
|
||||||
|
|
||||||
def test_create_member():
|
|
||||||
application = ApplicationFactory.create()
|
|
||||||
env1 = EnvironmentFactory.create(application=application)
|
|
||||||
env2 = EnvironmentFactory.create(application=application)
|
|
||||||
user_data = UserFactory.dictionary()
|
|
||||||
permission_set_names = [PermissionSets.EDIT_APPLICATION_TEAM]
|
|
||||||
|
|
||||||
member_role = Applications.create_member(
|
|
||||||
application,
|
|
||||||
user_data,
|
|
||||||
permission_set_names,
|
|
||||||
environment_roles_data=[
|
|
||||||
{"environment_id": env1.id, "role": CSPRole.BASIC_ACCESS.value},
|
|
||||||
{"environment_id": env2.id, "role": None},
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
assert member_role.user.dod_id == user_data["dod_id"]
|
|
||||||
# view application AND edit application team
|
|
||||||
assert len(member_role.permission_sets) == 2
|
|
||||||
|
|
||||||
env_roles = member_role.environment_roles
|
|
||||||
assert len(env_roles) == 1
|
|
||||||
assert env_roles[0].environment == env1
|
|
||||||
|
|
||||||
|
|
||||||
def test_for_user():
|
def test_for_user():
|
||||||
user = UserFactory.create()
|
user = UserFactory.create()
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
|
|||||||
@ -1,31 +0,0 @@
|
|||||||
from tests.factories import (
|
|
||||||
ApplicationFactory,
|
|
||||||
ApplicationRoleFactory,
|
|
||||||
UserFactory,
|
|
||||||
PortfolioFactory,
|
|
||||||
PortfolioRoleFactory,
|
|
||||||
)
|
|
||||||
|
|
||||||
from atst.services.invitation import Invitation
|
|
||||||
|
|
||||||
|
|
||||||
def test_invite_portfolio_member(queue):
|
|
||||||
inviter = UserFactory.create()
|
|
||||||
new_member = UserFactory.create()
|
|
||||||
portfolio = PortfolioFactory.create(owner=inviter)
|
|
||||||
ws_member = PortfolioRoleFactory.create(user=new_member, portfolio=portfolio)
|
|
||||||
invite_service = Invitation(inviter, ws_member, new_member.email)
|
|
||||||
new_invitation = invite_service.invite()
|
|
||||||
assert new_invitation == new_member.portfolio_invitations[0]
|
|
||||||
assert len(queue.get_queue()) == 1
|
|
||||||
|
|
||||||
|
|
||||||
def test_invite_application_member(queue):
|
|
||||||
inviter = UserFactory.create()
|
|
||||||
new_member = UserFactory.create()
|
|
||||||
application = ApplicationFactory.create()
|
|
||||||
member = ApplicationRoleFactory.create(user=new_member, application=application)
|
|
||||||
invite_service = Invitation(inviter, member, new_member.email)
|
|
||||||
new_invitation = invite_service.invite()
|
|
||||||
assert new_invitation == new_member.application_invitations[0]
|
|
||||||
assert len(queue.get_queue()) == 1
|
|
||||||
Loading…
x
Reference in New Issue
Block a user