Merge portfolio members routes with invitations.

`portfolios.create_member` now just sends an invitation, so it should be
with the invitation routes. This also de-duplicates the function for
sending a portfolio invitation email.
This commit is contained in:
dandds 2019-06-03 12:55:15 -04:00
parent 5434443b02
commit 4f8e9cddc8
7 changed files with 106 additions and 138 deletions

View File

@ -4,7 +4,6 @@ from operator import attrgetter
portfolios_bp = Blueprint("portfolios", __name__) portfolios_bp = Blueprint("portfolios", __name__)
from . import index from . import index
from . import members
from . import invitations from . import invitations
from . import admin from . import admin
from atst.utils.context_processors import portfolio as portfolio_context_processor from atst.utils.context_processors import portfolio as portfolio_context_processor

View File

@ -1,20 +1,23 @@
from flask import g, redirect, url_for, render_template from flask import g, redirect, url_for, render_template, request as http_request
from . import portfolios_bp from . import portfolios_bp
from atst.domain.authz.decorator import user_can_access_decorator as user_can
from atst.domain.exceptions import AlreadyExistsError
from atst.domain.invitations import PortfolioInvitations from atst.domain.invitations import PortfolioInvitations
from atst.domain.portfolios import Portfolios
from atst.models import Permissions
from atst.queue import queue from atst.queue import queue
from atst.utils.flash import formatted_flash as flash from atst.utils.flash import formatted_flash as flash
from atst.domain.authz.decorator import user_can_access_decorator as user_can import atst.forms.portfolio_member as member_forms
from atst.models.permissions import Permissions
def send_invite_email(owner_name, token, new_member_email): def send_portfolio_invitation(invitee_email, inviter_name, token):
body = render_template( body = render_template(
"emails/portfolio/invitation.txt", owner=owner_name, token=token "emails/portfolio/invitation.txt", owner=inviter_name, token=token
) )
queue.send_mail( queue.send_mail(
[new_member_email], [invitee_email],
"{} has invited you to a JEDI Cloud Portfolio".format(owner_name), "{} has invited you to a JEDI cloud portfolio".format(inviter_name),
body, body,
) )
@ -51,7 +54,7 @@ def revoke_invitation(portfolio_id, portfolio_token):
@user_can(Permissions.EDIT_PORTFOLIO_USERS, message="resend invitation") @user_can(Permissions.EDIT_PORTFOLIO_USERS, message="resend invitation")
def resend_invitation(portfolio_id, portfolio_token): def resend_invitation(portfolio_id, portfolio_token):
invite = PortfolioInvitations.resend(g.current_user, portfolio_token) invite = PortfolioInvitations.resend(g.current_user, portfolio_token)
send_invite_email(g.current_user.full_name, invite.token, invite.email) send_portfolio_invitation(invite.email, g.current_user.full_name, invite.token)
flash("resend_portfolio_invitation", user_name=invite.user_name) flash("resend_portfolio_invitation", user_name=invite.user_name)
return redirect( return redirect(
url_for( url_for(
@ -61,3 +64,38 @@ def resend_invitation(portfolio_id, portfolio_token):
_anchor="portfolio-members", _anchor="portfolio-members",
) )
) )
@portfolios_bp.route("/portfolios/<portfolio_id>/members/new", methods=["POST"])
@user_can(Permissions.CREATE_PORTFOLIO_USERS, message="create new portfolio member")
def invite_member(portfolio_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
form = member_forms.NewForm(http_request.form)
if form.validate():
try:
invite = Portfolios.invite(portfolio, g.current_user, form.update_data)
send_portfolio_invitation(
invite.email, g.current_user.full_name, invite.token
)
flash(
"new_portfolio_member", user_name=invite.user_name, portfolio=portfolio
)
except AlreadyExistsError:
return render_template(
"error.html", message="There was an error processing your request."
)
else:
pass
# TODO: flash error message
return redirect(
url_for(
"portfolios.admin",
portfolio_id=portfolio_id,
fragment="portfolio-members",
_anchor="portfolio-members",
)
)

View File

@ -1,57 +0,0 @@
from flask import render_template, request as http_request, g, redirect, url_for
from . import portfolios_bp
from atst.domain.exceptions import AlreadyExistsError
from atst.domain.portfolios import Portfolios
import atst.forms.portfolio_member as member_forms
from atst.domain.authz.decorator import user_can_access_decorator as user_can
from atst.models.permissions import Permissions
from atst.utils.flash import formatted_flash as flash
from atst.queue import queue
def send_portfolio_invitation(invitee_email, inviter_name, token):
body = render_template(
"emails/portfolio/invitation.txt", owner=inviter_name, token=token
)
queue.send_mail(
[invitee_email],
"{} has invited you to a JEDI cloud portfolio".format(inviter_name),
body,
)
@portfolios_bp.route("/portfolios/<portfolio_id>/members/new", methods=["POST"])
@user_can(Permissions.CREATE_PORTFOLIO_USERS, message="create new portfolio member")
def create_member(portfolio_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
form = member_forms.NewForm(http_request.form)
if form.validate():
try:
invite = Portfolios.invite(portfolio, g.current_user, form.update_data)
send_portfolio_invitation(
invite.email, g.current_user.full_name, invite.token
)
flash(
"new_portfolio_member", user_name=invite.user_name, portfolio=portfolio
)
except AlreadyExistsError:
return render_template(
"error.html", message="There was an error processing your request."
)
else:
pass
# TODO: flash error message
return redirect(
url_for(
"portfolios.admin",
portfolio_id=portfolio_id,
fragment="portfolio-members",
_anchor="portfolio-members",
)
)

View File

@ -78,6 +78,6 @@
{{ MultiStepModalForm( {{ MultiStepModalForm(
'add-port-mem', 'add-port-mem',
member_form, member_form,
url_for("portfolios.create_member", portfolio_id=portfolio.id), url_for("portfolios.invite_member", portfolio_id=portfolio.id),
[step_one, step_two], [step_one, step_two],
) }} ) }}

View File

@ -1,18 +1,12 @@
import pytest
import datetime import datetime
from flask import url_for from flask import url_for
from tests.factories import (
UserFactory,
PortfolioFactory,
PortfolioRoleFactory,
PortfolioInvitationFactory,
TaskOrderFactory,
)
from atst.domain.portfolios import Portfolios from atst.domain.portfolios import Portfolios
from atst.models import InvitationStatus, PortfolioRoleStatus from atst.models import InvitationStatus, PortfolioRoleStatus
from atst.domain.users import Users
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
from atst.queue import queue
from tests.factories import *
def test_existing_member_accepts_valid_invite(client, user_session): def test_existing_member_accepts_valid_invite(client, user_session):
@ -92,7 +86,7 @@ def test_user_who_has_not_accepted_portfolio_invite_cannot_view(client, user_ses
# create user in portfolio with invitation # create user in portfolio with invitation
user_session(portfolio.owner) user_session(portfolio.owner)
response = client.post( response = client.post(
url_for("portfolios.create_member", portfolio_id=portfolio.id), url_for("portfolios.invite_member", portfolio_id=portfolio.id),
data=user.to_dictionary(), data=user.to_dictionary(),
) )
@ -263,3 +257,55 @@ def test_existing_member_invite_resent_to_email_submitted_in_form(
assert user.email != "example@example.com" assert user.email != "example@example.com"
assert send_mail_job.func.__func__.__name__ == "_send_mail" assert send_mail_job.func.__func__.__name__ == "_send_mail"
assert send_mail_job.args[0] == ["example@example.com"] assert send_mail_job.args[0] == ["example@example.com"]
_DEFAULT_PERMS_FORM_DATA = {
"permission_sets-perms_app_mgmt": PermissionSets.VIEW_PORTFOLIO_APPLICATION_MANAGEMENT,
"permission_sets-perms_funding": PermissionSets.VIEW_PORTFOLIO_FUNDING,
"permission_sets-perms_reporting": PermissionSets.VIEW_PORTFOLIO_REPORTS,
"permission_sets-perms_portfolio_mgmt": PermissionSets.VIEW_PORTFOLIO_ADMIN,
}
def test_user_with_permission_has_add_member_link(client, user_session):
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
response = client.get(url_for("portfolios.admin", portfolio_id=portfolio.id))
assert response.status_code == 200
assert (
url_for("portfolios.invite_member", portfolio_id=portfolio.id).encode()
in response.data
)
def test_invite_member(client, user_session, session):
user_data = UserFactory.dictionary()
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
queue_length = len(queue.get_queue())
response = client.post(
url_for("portfolios.invite_member", portfolio_id=portfolio.id),
data={
"user_data-dod_id": user_data.get("dod_id"),
"user_data-first_name": user_data.get("first_name"),
"user_data-last_name": user_data.get("last_name"),
"user_data-email": user_data.get("email"),
**_DEFAULT_PERMS_FORM_DATA,
},
follow_redirects=True,
)
assert response.status_code == 200
full_name = "{} {}".format(user_data.get("first_name"), user_data.get("last_name"))
assert full_name in response.data.decode()
invitation = (
session.query(PortfolioInvitation)
.filter_by(dod_id=user_data.get("dod_id"))
.one()
)
assert invitation.role.portfolio == portfolio
assert len(queue.get_queue()) == queue_length + 1
assert len(invitation.role.permission_sets) == 5

View File

@ -1,58 +0,0 @@
from flask import url_for
from atst.domain.permission_sets import PermissionSets
from atst.models import PortfolioInvitation
from atst.queue import queue
from tests.factories import UserFactory, PortfolioFactory
_DEFAULT_PERMS_FORM_DATA = {
"permission_sets-perms_app_mgmt": PermissionSets.VIEW_PORTFOLIO_APPLICATION_MANAGEMENT,
"permission_sets-perms_funding": PermissionSets.VIEW_PORTFOLIO_FUNDING,
"permission_sets-perms_reporting": PermissionSets.VIEW_PORTFOLIO_REPORTS,
"permission_sets-perms_portfolio_mgmt": PermissionSets.VIEW_PORTFOLIO_ADMIN,
}
def test_user_with_permission_has_add_member_link(client, user_session):
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
response = client.get(url_for("portfolios.admin", portfolio_id=portfolio.id))
assert response.status_code == 200
assert (
url_for("portfolios.create_member", portfolio_id=portfolio.id).encode()
in response.data
)
def test_create_member(client, user_session, session):
user_data = UserFactory.dictionary()
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
queue_length = len(queue.get_queue())
response = client.post(
url_for("portfolios.create_member", portfolio_id=portfolio.id),
data={
"user_data-dod_id": user_data.get("dod_id"),
"user_data-first_name": user_data.get("first_name"),
"user_data-last_name": user_data.get("last_name"),
"user_data-email": user_data.get("email"),
**_DEFAULT_PERMS_FORM_DATA,
},
follow_redirects=True,
)
assert response.status_code == 200
full_name = "{} {}".format(user_data.get("first_name"), user_data.get("last_name"))
assert full_name in response.data.decode()
invitation = (
session.query(PortfolioInvitation)
.filter_by(dod_id=user_data.get("dod_id"))
.one()
)
assert invitation.role.portfolio == portfolio
assert len(queue.get_queue()) == queue_length + 1
assert len(invitation.role.permission_sets) == 5

View File

@ -197,14 +197,14 @@ def test_applications_update_team_env_roles(post_url_assert_status):
post_url_assert_status(rando, url, 404) post_url_assert_status(rando, url, 404)
# portfolios.create_member # portfolios.invite_member
def test_portfolios_create_member_access(post_url_assert_status): def test_portfolios_invite_member_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_ADMIN) ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_ADMIN)
owner = user_with() owner = user_with()
rando = user_with() rando = user_with()
portfolio = PortfolioFactory.create(owner=owner) portfolio = PortfolioFactory.create(owner=owner)
url = url_for("portfolios.create_member", portfolio_id=portfolio.id) url = url_for("portfolios.invite_member", portfolio_id=portfolio.id)
post_url_assert_status(ccpo, url, 302) post_url_assert_status(ccpo, url, 302)
post_url_assert_status(owner, url, 302) post_url_assert_status(owner, url, 302)
post_url_assert_status(rando, url, 404) post_url_assert_status(rando, url, 404)