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

send officer invitations when MO completes TO form

Browse Source
This commit is contained in:
dandds 2019-02-06 09:57:15 -05:00
parent 37df5b4b9c
commit a85487a2d1
7 changed files with 164 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
alembic/versions/c98adf9bb431_record_invitation_status.py Normal file
View File

@ -0,0 +1,32 @@
"""record invitation status
Revision ID: c98adf9bb431
Revises: da9d1c911a52
Create Date: 2019-02-06 09:02:28.617202
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'c98adf9bb431'
down_revision = 'da9d1c911a52'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('task_orders', sa.Column('cor_invite', sa.Boolean(), nullable=True))
op.add_column('task_orders', sa.Column('ko_invite', sa.Boolean(), nullable=True))
op.add_column('task_orders', sa.Column('so_invite', sa.Boolean(), nullable=True))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('task_orders', 'so_invite')
op.drop_column('task_orders', 'ko_invite')
op.drop_column('task_orders', 'cor_invite')
# ### end Alembic commands ###

5
atst/models/task_order.py
View File

@ -2,7 +2,7 @@ from enum import Enum
from datetime import date
import pendulum
from sqlalchemy import Column, Numeric, String, ForeignKey, Date, Integer
from sqlalchemy import Boolean, Column, Numeric, String, ForeignKey, Date, Integer
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.types import ARRAY
from sqlalchemy.orm import relationship
@ -62,16 +62,19 @@ class TaskOrder(Base, mixins.TimestampsMixin):
ko_email = Column(String) # Email
ko_phone_number = Column(String) # Phone Number
ko_dod_id = Column(String) # DOD ID
ko_invite = Column(Boolean)
cor_first_name = Column(String) # First Name
cor_last_name = Column(String) # Last Name
cor_email = Column(String) # Email
cor_phone_number = Column(String) # Phone Number
cor_dod_id = Column(String) # DOD ID
cor_invite = Column(Boolean)
so_first_name = Column(String) # First Name
so_last_name = Column(String) # Last Name
so_email = Column(String) # Email
so_phone_number = Column(String) # Phone Number
so_dod_id = Column(String) # DOD ID
so_invite = Column(Boolean)
pdf_attachment_id = Column(ForeignKey("attachments.id"))
_pdf = relationship("Attachment", foreign_keys=[pdf_attachment_id])
number = Column(String, unique=True) # Task Order Number

59
atst/routes/task_orders/invite.py
View File

@ -3,11 +3,70 @@ from flask import redirect, url_for, g
from . import task_orders_bp
from atst.domain.task_orders import TaskOrders
from atst.utils.flash import formatted_flash as flash
from atst.domain.portfolio_roles import PortfolioRoles
from atst.services.invitation import Invitation as InvitationService
OFFICER_INVITATIONS = [
{
"field": "ko_invite",
"prefix": "ko",
"role": "contracting_officer",
"subject": "Review a task order",
"template": "emails/invitation.txt",
},
{
"field": "cor_invite",
"prefix": "cor",
"role": "contracting_officer_representative",
"subject": "Help with a task order",
"template": "emails/invitation.txt",
},
{
"field": "so_invite",
"prefix": "so",
"role": "security_officer",
"subject": "Review security for a task order",
"template": "emails/invitation.txt",
},
]
def update_officer_invitations(user, task_order):
for officer_type in OFFICER_INVITATIONS:
field = officer_type["field"]
if getattr(task_order, field) and not getattr(task_order, officer_type["role"]):
prefix = officer_type["prefix"]
officer_data = {
field: getattr(task_order, prefix + "_" + field)
for field in [
"first_name",
"last_name",
"email",
"phone_number",
"dod_id",
]
}
officer = TaskOrders.add_officer(
user, task_order, officer_type["role"], officer_data
)
pf_officer_member = PortfolioRoles.get(task_order.portfolio.id, officer.id)
invite_service = InvitationService(
user,
pf_officer_member,
officer_data["email"],
subject=officer_type["subject"],
email_template=officer_type["template"],
)
invite_service.invite()
@task_orders_bp.route("/task_orders/invite/<task_order_id>", methods=["POST"])
def invite(task_order_id):
task_order = TaskOrders.get(g.current_user, task_order_id)
# TODO: only do this if TO is complete
update_officer_invitations(g.current_user, task_order)
portfolio = task_order.portfolio
flash("task_order_congrats", portfolio=portfolio)
return redirect(

63
atst/routes/task_orders/new.py
View File

@ -12,9 +12,7 @@ from flask import (
from . import task_orders_bp
from atst.domain.task_orders import TaskOrders
from atst.domain.portfolios import Portfolios
from atst.domain.portfolio_roles import PortfolioRoles
import atst.forms.task_order as task_order_form
from atst.services.invitation import Invitation as InvitationService
TASK_ORDER_SECTIONS = [
@ -173,7 +171,7 @@ class UpdateTaskOrderWorkflow(ShowTaskOrderWorkflow):
def validate(self):
return self.form.validate()
def _update_task_order(self):
def update(self):
if self.task_order:
if "portfolio_name" in self.form.data:
new_name = self.form.data["portfolio_name"]
@ -189,65 +187,6 @@ class UpdateTaskOrderWorkflow(ShowTaskOrderWorkflow):
self._task_order = TaskOrders.create(portfolio=pf, creator=self.user)
TaskOrders.update(self.user, self.task_order, **self.task_order_form_data)
OFFICER_INVITATIONS = [
{
"field": "ko_invite",
"prefix": "ko",
"role": "contracting_officer",
"subject": "Review a task order",
"template": "emails/invitation.txt",
},
{
"field": "cor_invite",
"prefix": "cor",
"role": "contracting_officer_representative",
"subject": "Help with a task order",
"template": "emails/invitation.txt",
},
{
"field": "so_invite",
"prefix": "so",
"role": "security_officer",
"subject": "Review security for a task order",
"template": "emails/invitation.txt",
},
]
def _update_officer_invitations(self):
for officer_type in self.OFFICER_INVITATIONS:
field = officer_type["field"]
if (
hasattr(self.form, field)
and self.form[field].data
and not getattr(self.task_order, officer_type["role"])
):
prefix = officer_type["prefix"]
officer_data = {
field: getattr(self.task_order, prefix + "_" + field)
for field in [
"first_name",
"last_name",
"email",
"phone_number",
"dod_id",
]
}
officer = TaskOrders.add_officer(
self.user, self.task_order, officer_type["role"], officer_data
)
pf_officer_member = PortfolioRoles.get(self.portfolio.id, officer.id)
invite_service = InvitationService(
self.user,
pf_officer_member,
officer_data["email"],
subject=officer_type["subject"],
email_template=officer_type["template"],
)
invite_service.invite()
def update(self):
self._update_task_order()
self._update_officer_invitations()
return self.task_order

27
tests/routes/portfolios/test_invitations.py
View File

@ -212,27 +212,20 @@ def test_existing_member_invite_resent_to_email_submitted_in_form(
def test_contracting_officer_accepts_invite(monkeypatch, client, user_session):
portfolio = PortfolioFactory.create()
task_order = TaskOrderFactory.create(portfolio=portfolio)
user_info = UserFactory.dictionary()
task_order = TaskOrderFactory.create(
portfolio=portfolio,
ko_first_name=user_info["first_name"],
ko_last_name=user_info["last_name"],
ko_email=user_info["email"],
ko_phone_number=user_info["phone_number"],
ko_dod_id=user_info["dod_id"],
ko_invite=True,
)
# create contracting officer
user_session(portfolio.owner)
client.post(
url_for("task_orders.new", screen=3, task_order_id=task_order.id),
data={
"portfolio_role": "contracting_officer",
"ko_first_name": user_info["first_name"],
"ko_last_name": user_info["last_name"],
"ko_email": user_info["email"],
"ko_phone_number": user_info["phone_number"],
"ko_dod_id": user_info["dod_id"],
"cor_phone_number": user_info["phone_number"],
"so_phone_number": user_info["phone_number"],
"so_dod_id": task_order.so_dod_id,
"cor_dod_id": task_order.cor_dod_id,
"ko_invite": True,
},
)
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
# contracting officer accepts invitation
user = Users.get_by_dod_id(user_info["dod_id"])

59
tests/routes/task_orders/test_invite.py
View File

@ -1,7 +1,7 @@
import pytest
from flask import url_for
from tests.factories import PortfolioFactory, TaskOrderFactory
from tests.factories import PortfolioFactory, TaskOrderFactory, UserFactory
def test_invite(client, user_session):
@ -15,3 +15,60 @@ def test_invite(client, user_session):
"portfolios.view_task_order", portfolio_id=to.portfolio_id, task_order_id=to.id
)
assert redirect in response.headers["Location"]
def test_invite_officers_to_task_order(client, user_session, queue):
task_order = TaskOrderFactory.create(
ko_invite=True, cor_invite=True, so_invite=True
)
portfolio = task_order.portfolio
user_session(portfolio.owner)
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
# owner and three officers are portfolio members
assert len(portfolio.members) == 4
roles = [member.role.name for member in portfolio.members]
# officers exist in roles
assert roles.count("officer") == 3
# email invitations are enqueued
assert len(queue.get_queue()) == 3
# task order has relationship to user for each officer role
assert task_order.contracting_officer.dod_id == task_order.ko_dod_id
assert task_order.contracting_officer_representative.dod_id == task_order.cor_dod_id
assert task_order.security_officer.dod_id == task_order.so_dod_id
def test_add_officer_but_do_not_invite(client, user_session, queue):
task_order = TaskOrderFactory.create(
ko_invite=False, cor_invite=False, so_invite=False
)
portfolio = task_order.portfolio
user_session(portfolio.owner)
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
portfolio = task_order.portfolio
# owner is only portfolio member
assert len(portfolio.members) == 1
# no invitations are enqueued
assert len(queue.get_queue()) == 0
def test_does_not_resend_officer_invitation(client, user_session):
user = UserFactory.create()
contracting_officer = UserFactory.create()
portfolio = PortfolioFactory.create(owner=user)
task_order = TaskOrderFactory.create(
creator=user,
portfolio=portfolio,
ko_first_name=contracting_officer.first_name,
ko_last_name=contracting_officer.last_name,
ko_dod_id=contracting_officer.dod_id,
ko_invite=True,
)
user_session(user)
for i in range(2):
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
assert len(contracting_officer.invitations) == 1

63
tests/routes/task_orders/test_new_task_order.py
View File

@ -235,69 +235,6 @@ def test_cor_data_set_to_user_data_if_am_cor_is_checked(task_order):
assert task_order.cor_dod_id == task_order.creator.dod_id
def test_invite_officers_to_task_order(task_order, queue):
to_data = {
**TaskOrderFactory.dictionary(),
"ko_invite": True,
"cor_invite": True,
"so_invite": True,
}
workflow = UpdateTaskOrderWorkflow(
task_order.creator, to_data, screen=3, task_order_id=task_order.id
)
workflow.update()
portfolio = task_order.portfolio
# owner and three officers are portfolio members
assert len(portfolio.members) == 4
roles = [member.role.name for member in portfolio.members]
# officers exist in roles
assert roles.count("officer") == 3
# email invitations are enqueued
assert len(queue.get_queue()) == 3
# task order has relationship to user for each officer role
assert task_order.contracting_officer.dod_id == to_data["ko_dod_id"]
assert task_order.contracting_officer_representative.dod_id == to_data["cor_dod_id"]
assert task_order.security_officer.dod_id == to_data["so_dod_id"]
def test_add_officer_but_do_not_invite(task_order, queue):
to_data = {
**TaskOrderFactory.dictionary(),
"ko_invite": False,
"cor_invite": False,
"so_invite": False,
}
workflow = UpdateTaskOrderWorkflow(
task_order.creator, to_data, screen=3, task_order_id=task_order.id
)
workflow.update()
portfolio = task_order.portfolio
# owner is only portfolio member
assert len(portfolio.members) == 1
# no invitations are enqueued
assert len(queue.get_queue()) == 0
def test_update_does_not_resend_invitation():
user = UserFactory.create()
contracting_officer = UserFactory.create()
portfolio = PortfolioFactory.create(owner=user)
task_order = TaskOrderFactory.create(
creator=user,
portfolio=portfolio,
ko_first_name=contracting_officer.first_name,
ko_last_name=contracting_officer.last_name,
ko_dod_id=contracting_officer.dod_id,
)
to_data = {**task_order.to_dictionary(), "ko_invite": True}
workflow = UpdateTaskOrderWorkflow(
user, to_data, screen=3, task_order_id=task_order.id
)
for i in range(2):
workflow.update()
assert len(contracting_officer.invitations) == 1
def test_review_task_order_form(client, user_session, task_order):
user_session(task_order.creator)