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

New backend flow for application invitations.

Browse Source 
Like portfolio invitations, now a user is not associated with an
application role until they accept the associated invitation.
- domain method for inviting user to application
- change application route for inviting a member
- ApplicationRole model knows user name from invitation
This commit is contained in:
dandds 2019-06-04 10:27:18 -04:00
parent a2d1c470c1
commit fa50c01e48
6 changed files with 141 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
atst/domain/applications.py
View File

@ -1,13 +1,20 @@
from sqlalchemy.orm.exc import NoResultFound
from atst.database import db
from . import BaseDomainClass
from atst.database import db
from atst.domain.application_roles import ApplicationRoles
from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.environments import Environments
from atst.domain.exceptions import NotFoundError
from atst.domain.invitations import ApplicationInvitations
from atst.domain.users import Users
from atst.models import Application, ApplicationRole, ApplicationRoleStatus
from atst.models import (
Application,
ApplicationRole,
ApplicationRoleStatus,
EnvironmentRole,
)
from atst.utils import first_or_none
class Applications(BaseDomainClass):
@ -104,6 +111,52 @@ class Applications(BaseDomainClass):
return application_role
@classmethod
def invite(
cls,
application,
inviter,
user_data,
permission_sets_names=None,
environment_roles_data=None,
):
permission_sets_names = permission_sets_names or []
permission_sets = ApplicationRoles._permission_sets_for_names(
permission_sets_names
)
app_role = ApplicationRole(
application=application, permission_sets=permission_sets
)
db.session.add(app_role)
for env_role_data in environment_roles_data:
env_role_name = env_role_data.get("role")
environment_id = env_role_data.get("environment_id")
if env_role_name is not None:
# pylint: disable=cell-var-from-loop
environment = first_or_none(
lambda e: str(e.id) == str(environment_id), application.environments
)
if environment is None:
raise NotFoundError("environment")
else:
env_role = EnvironmentRole(
application_role=app_role,
environment=environment,
role=env_role_name,
)
db.session.add(env_role)
invitation = ApplicationInvitations.create(
inviter=inviter, role=app_role, member_data=user_data
)
db.session.add(invitation)
db.session.commit()
return invitation
@classmethod
def remove_member(cls, application, user_id):
application_role = ApplicationRoles.get(

9
atst/models/application_role.py
View File

@ -59,12 +59,17 @@ class ApplicationRole(
),
)
@property
def latest_invitation(self):
if self.invitations:
return self.invitations[-1]
@property
def user_name(self):
if self.user:
return self.user.full_name
else:
return None
elif self.latest_invitation:
return self.latest_invitation.user_name
def __repr__(self):
return "<ApplicationRole(application='{}', user_id='{}', id='{}', permissions={})>".format(

29
atst/routes/applications/team.py
View File

@ -13,8 +13,8 @@ from atst.domain.users import Users
from atst.forms.application_member import NewForm as NewMemberForm
from atst.forms.team import TeamForm
from atst.models import Permissions
from atst.services.invitation import Invitation as InvitationService
from atst.utils.flash import formatted_flash as flash
from atst.queue import queue
def get_form_permission_value(member, edit_perm_set):
@ -125,6 +125,17 @@ def update_team(application_id):
return (render_team_page(application), 400)
def send_application_invitation(invitee_email, inviter_name, token):
body = render_template(
"emails/application/invitation.txt", owner=inviter_name, token=token
)
queue.send_mail(
[invitee_email],
"{} has invited you to a JEDI cloud application".format(inviter_name),
body,
)
@applications_bp.route("/application/<application_id>/members/new", methods=["POST"])
@user_can(
Permissions.CREATE_APPLICATION_MEMBER, message="create new application member"
@ -135,19 +146,19 @@ def create_member(application_id):
if form.validate():
try:
member = Applications.create_member(
application,
form.user_data.data,
permission_sets=form.permission_sets.data,
invite = Applications.invite(
application=application,
inviter=g.current_user,
user_data=form.user_data.data,
permission_sets_names=form.permission_sets.data,
environment_roles_data=form.environment_roles.data,
)
invite_service = InvitationService(
g.current_user, member, form.user_data.data.get("email")
send_application_invitation(
invite.email, g.current_user.full_name, invite.token
)
invite_service.invite()
flash("new_portfolio_member", new_member=member)
flash("new_application_member", user_name=invite.user_name)
except AlreadyExistsError:
return render_template(

2
atst/utils/flash.py
View File

@ -67,7 +67,7 @@ MESSAGES = {
"new_application_member": {
"title_template": translate("flash.success"),
"message_template": """
<p>{{ "flash.new_application_member" | translate({ "user_name": new_member.user_name }) }}</p>
<p>{{ "flash.new_application_member" | translate({ "user_name": user_name }) }}</p>
""",
"category": "success",
},

45
tests/domain/test_applications.py
View File

@ -189,3 +189,48 @@ def test_remove_member():
)
is None
)
def test_invite():
application = ApplicationFactory.create()
env1 = EnvironmentFactory.create(application=application)
env2 = EnvironmentFactory.create(application=application)
user_data = UserFactory.dictionary()
permission_sets_names = [PermissionSets.EDIT_APPLICATION_TEAM]
invitation = Applications.invite(
application=application,
inviter=application.portfolio.owner,
user_data=user_data,
permission_sets_names=permission_sets_names,
environment_roles_data=[
{"environment_id": env1.id, "role": CSPRole.BASIC_ACCESS.value},
{"environment_id": env2.id, "role": None},
],
)
member_role = invitation.role
assert invitation.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_invite_to_nonexistent_environment():
application = ApplicationFactory.create()
env1 = EnvironmentFactory.create(application=application)
user_data = UserFactory.dictionary()
with pytest.raises(NotFoundError):
Applications.invite(
application=application,
inviter=application.portfolio.owner,
user_data=user_data,
environment_roles_data=[
{"environment_id": env1.id, "role": CSPRole.BASIC_ACCESS.value},
{"environment_id": uuid4(), "role": CSPRole.BASIC_ACCESS.value},
],
)

19
tests/routes/applications/test_team.py
View File

@ -1,10 +1,10 @@
import pytest
import uuid
from flask import url_for
from atst.domain.permission_sets import PermissionSets
from atst.models import CSPRole
from atst.forms.data import ENV_ROLE_NO_ACCESS as NO_ACCESS
from atst.queue import queue
from tests.factories import *
@ -145,7 +145,8 @@ def test_update_team_revoke_environment_access(client, user_session, db, session
assert not session.query(env_role_exists).scalar()
def test_create_member(client, user_session):
def test_create_member(client, user_session, session):
queue_length = len(queue.get_queue())
user = UserFactory.create()
application = ApplicationFactory.create(
environments=[{"name": "Naboo"}, {"name": "Endor"}]
@ -179,14 +180,18 @@ def test_create_member(client, user_session):
_external=True,
)
assert response.location == expected_url
assert len(user.application_roles) == 1
assert user.application_roles[0].application == application
environment_roles = [
er for ar in user.application_roles for er in ar.environment_roles
]
assert len(application.roles) == 1
environment_roles = application.roles[0].environment_roles
assert len(environment_roles) == 1
assert environment_roles[0].environment == env
invitation = (
session.query(ApplicationInvitation).filter_by(dod_id=user.dod_id).one()
)
assert invitation.role.application == application
assert len(queue.get_queue()) == queue_length + 1
def test_remove_member_success(client, user_session):
user = UserFactory.create()