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

Merge pull request #855 from dod-ccpo/update-to-table

Browse Source 
Update TO table
This commit is contained in:
leigh-mil 2019-05-31 13:23:16 -04:00 committed by GitHub
commit a9535edf3a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
30 changed files with 220 additions and 2507 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
alembic/versions/d73cba9a4259_remove_columns_from_task_orders.py Normal file
View File

@ -0,0 +1,132 @@
"""remove columns from task orders
Revision ID: d73cba9a4259
Revises: d2390c547dca
Create Date: 2019-05-30 13:47:08.684986
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = 'd73cba9a4259'
down_revision = 'd2390c547dca'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint('task_orders_users_so_id', 'task_orders', type_='foreignkey')
op.drop_constraint('task_orders_users_ko_id', 'task_orders', type_='foreignkey')
op.drop_constraint('task_orders_attachments_csp_attachment_id', 'task_orders', type_='foreignkey')
op.drop_constraint('task_orders_users_cor_id', 'task_orders', type_='foreignkey')
op.drop_constraint('task_orders_dd_254s_id', 'task_orders', type_='foreignkey')
op.drop_column('task_orders', 'custom_clauses')
op.drop_column('task_orders', 'unlimited_level_of_warrant')
op.drop_column('task_orders', 'complexity_other')
op.drop_column('task_orders', 'level_of_warrant')
op.drop_column('task_orders', 'ko_id')
op.drop_column('task_orders', 'so_first_name')
op.drop_column('task_orders', 'ko_invite')
op.drop_column('task_orders', 'team_experience')
op.drop_column('task_orders', 'so_phone_number')
op.drop_column('task_orders', 'cor_phone_number')
op.drop_column('task_orders', 'end_date')
op.drop_column('task_orders', 'cor_first_name')
op.drop_column('task_orders', 'app_migration')
op.drop_column('task_orders', 'performance_length')
op.drop_column('task_orders', 'ko_last_name')
op.drop_column('task_orders', 'so_last_name')
op.drop_column('task_orders', 'cor_dod_id')
op.drop_column('task_orders', 'scope')
op.drop_column('task_orders', 'complexity')
op.drop_column('task_orders', 'dev_team_other')
op.drop_column('task_orders', 'cor_invite')
op.drop_column('task_orders', 'ko_first_name')
op.drop_column('task_orders', 'native_apps')
op.drop_column('task_orders', 'csp_attachment_id')
op.drop_column('task_orders', 'clin_02')
op.drop_column('task_orders', 'cor_id')
op.drop_column('task_orders', 'so_email')
op.drop_column('task_orders', 'ko_email')
op.drop_column('task_orders', 'start_date')
op.drop_column('task_orders', 'cor_last_name')
op.drop_column('task_orders', 'ko_phone_number')
op.drop_column('task_orders', 'so_id')
op.drop_column('task_orders', 'dev_team')
op.drop_column('task_orders', 'clin_01')
op.drop_column('task_orders', 'cor_email')
op.drop_column('task_orders', 'loas')
op.drop_column('task_orders', 'so_dod_id')
op.drop_column('task_orders', 'so_invite')
op.drop_column('task_orders', 'clin_03')
op.drop_column('task_orders', 'dd_254_id')
op.drop_column('task_orders', 'clin_04')
op.drop_column('task_orders', 'ko_dod_id')
op.drop_table('dd_254s')
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('task_orders', sa.Column('ko_dod_id', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('clin_04', sa.NUMERIC(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('dd_254_id', postgresql.UUID(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('clin_03', sa.NUMERIC(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_invite', sa.BOOLEAN(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_dod_id', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('loas', postgresql.ARRAY(sa.VARCHAR()), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_email', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('clin_01', sa.NUMERIC(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('dev_team', postgresql.ARRAY(sa.VARCHAR()), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_id', postgresql.UUID(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_phone_number', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_last_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('start_date', sa.DATE(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_email', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_email', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_id', postgresql.UUID(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('clin_02', sa.NUMERIC(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('csp_attachment_id', postgresql.UUID(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('native_apps', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_first_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_invite', sa.BOOLEAN(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('dev_team_other', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('complexity', postgresql.ARRAY(sa.VARCHAR()), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('scope', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_dod_id', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_last_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_last_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('performance_length', sa.INTEGER(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('app_migration', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_first_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('end_date', sa.DATE(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('cor_phone_number', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_phone_number', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('team_experience', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_invite', sa.BOOLEAN(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('so_first_name', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('ko_id', postgresql.UUID(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('level_of_warrant', sa.NUMERIC(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('complexity_other', sa.VARCHAR(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('unlimited_level_of_warrant', sa.BOOLEAN(), autoincrement=False, nullable=True))
op.add_column('task_orders', sa.Column('custom_clauses', sa.VARCHAR(), autoincrement=False, nullable=True))
op.create_foreign_key('task_orders_users_cor_id', 'task_orders', 'users', ['cor_id'], ['id'])
op.create_foreign_key('task_orders_attachments_csp_attachment_id', 'task_orders', 'attachments', ['csp_attachment_id'], ['id'])
op.create_foreign_key('task_orders_users_ko_id', 'task_orders', 'users', ['ko_id'], ['id'])
op.create_foreign_key('task_orders_users_so_id', 'task_orders', 'users', ['so_id'], ['id'])
op.create_table('dd_254s',
sa.Column('time_created', postgresql.TIMESTAMP(timezone=True), server_default=sa.text('now()'), autoincrement=False, nullable=False),
sa.Column('time_updated', postgresql.TIMESTAMP(timezone=True), server_default=sa.text('now()'), autoincrement=False, nullable=False),
sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
sa.Column('certifying_official', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('certifying_official_title', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('certifying_official_address', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('certifying_official_phone', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('required_distribution', postgresql.ARRAY(sa.VARCHAR()), autoincrement=False, nullable=True),
sa.PrimaryKeyConstraint('id', name='dd_254s_pkey')
)
op.create_foreign_key('task_orders_dd_254s_id', 'task_orders', 'dd_254s', ['dd_254_id'], ['id'])
# ### end Alembic commands ###

33
atst/domain/authz/__init__.py
View File

@ -59,39 +59,6 @@ class Authorization(object):
return True
@classmethod
def is_ko(cls, user, task_order):
return user == task_order.contracting_officer
@classmethod
def is_cor(cls, user, task_order):
return user == task_order.contracting_officer_representative
@classmethod
def is_so(cls, user, task_order):
return user == task_order.security_officer
@classmethod
def check_is_ko_or_cor(cls, user, task_order):
if Authorization.is_ko(user, task_order) or Authorization.is_cor(
user, task_order
):
return True
else:
raise UnauthorizedError(user, "not KO or COR")
@classmethod
def check_is_ko(cls, user, task_order):
if task_order.contracting_officer != user:
message = "review task order {}".format(task_order.id)
raise UnauthorizedError(user, message)
@classmethod
def check_is_so(cls, user, task_order):
if task_order.security_officer != user:
message = "review task order {}".format(task_order.id)
raise UnauthorizedError(user, message)
def user_can_access(user, permission, portfolio=None, application=None, message=None):
if application:

118
atst/domain/task_orders.py
View File

@ -2,10 +2,7 @@ from flask import current_app as app
from atst.database import db
from atst.models.task_order import TaskOrder
from atst.models.dd_254 import DD254
from . import BaseDomainClass
from atst.domain.portfolios import Portfolios
from atst.domain.permission_sets import PermissionSets
class TaskOrderError(Exception):
@ -16,42 +13,9 @@ class TaskOrders(BaseDomainClass):
model = TaskOrder
resource_name = "task_order"
SECTIONS = {
"app_info": [
"portfolio_name",
"scope",
"defense_component",
"app_migration",
"native_apps",
"complexity",
"dev_team",
"team_experience",
],
"funding": [
"performance_length",
"csp_estimate",
"clin_01",
"clin_02",
"clin_03",
"clin_04",
],
"oversight": [
"ko_first_name",
"ko_last_name",
"ko_email",
"ko_phone_number",
"cor_first_name",
"cor_last_name",
"cor_email",
"cor_phone_number",
"so_first_name",
"so_last_name",
"so_email",
"so_phone_number",
],
}
SECTIONS = {"app_info": ["portfolio_name"], "funding": [], "oversight": []}
UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"]
UNCLASSIFIED_FUNDING = []
@classmethod
def create(cls, creator, portfolio):
@ -102,87 +66,9 @@ class TaskOrders(BaseDomainClass):
return True
@classmethod
def can_ko_sign(cls, task_order):
return (
TaskOrders.all_sections_complete(task_order)
and DD254s.is_complete(task_order.dd_254)
and not TaskOrders.is_signed_by_ko(task_order)
)
@classmethod
def is_signed_by_ko(cls, task_order):
return task_order.signer_dod_id is not None
@classmethod
def mission_owner_sections(cls):
section_list = TaskOrders.SECTIONS
if not app.config.get("CLASSIFIED"):
section_list["funding"] = TaskOrders.UNCLASSIFIED_FUNDING
return section_list
OFFICERS = [
"contracting_officer",
"contracting_officer_representative",
"security_officer",
]
@classmethod
def add_officer(cls, task_order, officer_type, officer_data):
if officer_type in TaskOrders.OFFICERS:
portfolio = task_order.portfolio
existing_member = next(
(
member
for member in portfolio.members
if member.user.dod_id == officer_data["dod_id"]
),
None,
)
if existing_member:
portfolio_user = existing_member.user
else:
member = Portfolios.create_member(
portfolio,
{
**officer_data,
"permission_sets": [PermissionSets.EDIT_PORTFOLIO_FUNDING],
},
)
portfolio_user = member.user
setattr(task_order, officer_type, portfolio_user)
db.session.add(task_order)
db.session.commit()
return portfolio_user
else:
raise TaskOrderError(
"{} is not an officer role on task orders".format(officer_type)
)
@classmethod
def add_dd_254(user, task_order, dd_254_data):
dd_254 = DD254(**dd_254_data)
task_order.dd_254 = dd_254
db.session.add(task_order)
db.session.commit()
class DD254s:
# TODO: standin implementation until we have a real download,
# sign, and verify process for the DD 254 PDF
@classmethod
def is_complete(cls, dd254):
if dd254 is None:
return False
for col in DD254.__table__.columns:
if getattr(dd254, col.name) is None:
return False
return True

39
atst/forms/dd_254.py
View File

@ -1,39 +0,0 @@
from wtforms.fields import SelectMultipleField, StringField
from wtforms.fields.html5 import TelField
from wtforms.widgets import ListWidget, CheckboxInput
from wtforms.validators import Required
from atst.forms.validators import PhoneNumber
from .forms import BaseForm
from .data import REQUIRED_DISTRIBUTIONS
from atst.utils.localization import translate
class DD254Form(BaseForm):
certifying_official = StringField(
translate("forms.dd_254.certifying_official.label"),
description=translate("forms.dd_254.certifying_official.description"),
validators=[Required()],
)
certifying_official_title = StringField(
translate("forms.dd_254.certifying_official_title.label"),
validators=[Required()],
)
certifying_official_address = StringField(
translate("forms.dd_254.certifying_official_address.label"),
description=translate("forms.dd_254.certifying_official_address.description"),
validators=[Required()],
)
certifying_official_phone = TelField(
translate("forms.dd_254.certifying_official_phone.label"),
description=translate("forms.dd_254.certifying_official_phone.description"),
validators=[Required(), PhoneNumber()],
)
required_distribution = SelectMultipleField(
translate("forms.dd_254.required_distribution.label"),
choices=REQUIRED_DISTRIBUTIONS,
default="",
widget=ListWidget(prefix_label=False),
option_widget=CheckboxInput(),
)

62
atst/forms/officers.py
View File

@ -1,62 +0,0 @@
from wtforms.fields import StringField, BooleanField
from wtforms.fields.html5 import TelField
from wtforms.validators import Email, Length, Optional
from atst.forms.validators import IsNumber, PhoneNumber
from .forms import BaseForm
from .fields import FormFieldWrapper
class OfficerForm(BaseForm):
first_name = StringField("First Name")
last_name = StringField("Last Name")
email = StringField("Email", validators=[Optional(), Email()])
phone_number = TelField("Phone Number", validators=[PhoneNumber()])
dod_id = StringField("DoD ID", validators=[Optional(), Length(min=10), IsNumber()])
invite = BooleanField("Invite to Task Order Builder")
class EditTaskOrderOfficersForm(BaseForm):
contracting_officer = FormFieldWrapper(OfficerForm)
contracting_officer_representative = FormFieldWrapper(OfficerForm)
security_officer = FormFieldWrapper(OfficerForm)
OFFICER_PREFIXES = {
"contracting_officer": "ko",
"contracting_officer_representative": "cor",
"security_officer": "so",
}
OFFICER_INFO_FIELD_NAMES = [
"first_name",
"last_name",
"email",
"phone_number",
"dod_id",
"invite",
]
def process(self, formdata=None, obj=None, data=None, **kwargs):
if obj:
for name, field in self._fields.items():
if name in self.OFFICER_PREFIXES:
prefix = self.OFFICER_PREFIXES[name]
officer_data = {
field_name: getattr(obj, prefix + "_" + field_name)
for field_name in self.OFFICER_INFO_FIELD_NAMES
}
field.process(formdata=formdata, data=officer_data)
else:
field.process(formdata)
else:
super(EditTaskOrderOfficersForm, self).process(
formdata=formdata, obj=obj, data=data, **kwargs
)
def populate_obj(self, obj):
for name, field in self._fields.items():
if name in self.OFFICER_PREFIXES:
prefix = self.OFFICER_PREFIXES[name]
for field_name in self.OFFICER_INFO_FIELD_NAMES:
setattr(obj, prefix + "_" + field_name, field[field_name].data)

1
atst/models/__init__.py
View File

@ -16,7 +16,6 @@ from .audit_event import AuditEvent
from .portfolio_invitation import PortfolioInvitation
from .application_invitation import ApplicationInvitation
from .task_order import TaskOrder
from .dd_254 import DD254
from .notification_recipient import NotificationRecipient
from .mixins.invites import Status as InvitationStatus

31
atst/models/dd_254.py
View File

@ -1,31 +0,0 @@
from sqlalchemy import Column, String
from sqlalchemy.types import ARRAY
from sqlalchemy.orm import relationship
from atst.models import Base, types, mixins
class DD254(Base, mixins.TimestampsMixin):
__tablename__ = "dd_254s"
id = types.Id()
certifying_official = Column(String)
certifying_official_title = Column(String)
certifying_official_address = Column(String)
certifying_official_phone = Column(String)
required_distribution = Column(ARRAY(String))
task_order = relationship("TaskOrder", uselist=False, backref="task_order")
def to_dictionary(self):
return {
c.name: getattr(self, c.name)
for c in self.__table__.columns
if c.name not in ["id"]
}
def __repr__(self):
return "<DD254(certifying_official='{}', task_order='{}', id='{}')>".format(
self.certifying_official, self.task_order.id, self.id
)

187
atst/models/task_order.py
View File

@ -1,29 +1,13 @@
from enum import Enum
from datetime import date
import pendulum
from sqlalchemy import (
Column,
Numeric,
String,
ForeignKey,
Date,
Integer,
DateTime,
Boolean,
)
from sqlalchemy import Column, DateTime, ForeignKey, String
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.types import ARRAY
from sqlalchemy.orm import relationship
from werkzeug.datastructures import FileStorage
from atst.models import Attachment, Base, types, mixins
# Imports used for mocking TO balance
from atst.domain.csp.reports import MockReportingProvider
from flask import current_app as app
import random
class Status(Enum):
STARTED = "Started"
@ -43,72 +27,11 @@ class TaskOrder(Base, mixins.TimestampsMixin):
user_id = Column(ForeignKey("users.id"))
creator = relationship("User", foreign_keys="TaskOrder.user_id")
ko_id = Column(ForeignKey("users.id"))
contracting_officer = relationship("User", foreign_keys="TaskOrder.ko_id")
cor_id = Column(ForeignKey("users.id"))
contracting_officer_representative = relationship(
"User", foreign_keys="TaskOrder.cor_id"
)
so_id = Column(ForeignKey("users.id"))
security_officer = relationship("User", foreign_keys="TaskOrder.so_id")
dd_254_id = Column(ForeignKey("dd_254s.id"))
dd_254 = relationship("DD254")
scope = Column(String) # Cloud Project Scope
app_migration = Column(String) # App Migration
native_apps = Column(String) # Native Apps
complexity = Column(ARRAY(String)) # Application Complexity
complexity_other = Column(String)
dev_team = Column(ARRAY(String)) # Development Team
dev_team_other = Column(String)
team_experience = Column(String) # Team Experience
start_date = Column(Date) # Period of Performance
end_date = Column(Date)
performance_length = Column(Integer)
csp_attachment_id = Column(ForeignKey("attachments.id"))
_csp_estimate = relationship("Attachment", foreign_keys=[csp_attachment_id])
clin_01 = Column(Numeric(scale=2))
clin_02 = Column(Numeric(scale=2))
clin_03 = Column(Numeric(scale=2))
clin_04 = Column(Numeric(scale=2))
ko_first_name = Column(String) # First Name
ko_last_name = Column(String) # Last Name
ko_email = Column(String) # Email
ko_phone_number = Column(String) # Phone Number
ko_dod_id = Column(String) # DOD ID
ko_invite = Column(Boolean, default=False)
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, default=False)
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, default=False)
pdf_attachment_id = Column(ForeignKey("attachments.id"))
_pdf = relationship("Attachment", foreign_keys=[pdf_attachment_id])
number = Column(String, unique=True) # Task Order Number
loas = Column(ARRAY(String)) # Line of Accounting (LOA)
custom_clauses = Column(String) # Custom Clauses
signer_dod_id = Column(String)
signed_at = Column(DateTime)
level_of_warrant = Column(Numeric(scale=2))
unlimited_level_of_warrant = Column(Boolean, default=False)
@hybrid_property
def csp_estimate(self):
return self._csp_estimate
@csp_estimate.setter
def csp_estimate(self, new_csp_estimate):
self._csp_estimate = self._set_attachment(new_csp_estimate, "_csp_estimate")
@hybrid_property
def pdf(self):
@ -128,14 +51,6 @@ class TaskOrder(Base, mixins.TimestampsMixin):
else:
raise TypeError("Could not set attachment with invalid type")
@property
def is_submitted(self):
return (
self.number is not None
and self.start_date is not None
and self.end_date is not None
)
@property
def is_active(self):
return self.status == Status.ACTIVE
@ -146,19 +61,21 @@ class TaskOrder(Base, mixins.TimestampsMixin):
@property
def status(self):
if self.is_submitted:
now = pendulum.now().date()
if self.start_date > now:
return Status.PENDING
elif self.end_date < now:
return Status.EXPIRED
# TODO: fix task order -- implement correctly using CLINs
# Faked for display purposes
return Status.ACTIVE
else:
return Status.STARTED
@property
def display_status(self):
return self.status.value
def start_date(self):
# TODO: fix task order -- reimplement using CLINs
# Faked for display purposes
return date.today()
@property
def end_date(self):
# TODO: fix task order -- reimplement using CLINs
# Faked for display purposes
return date.today()
@property
def days_to_expiration(self):
@ -167,82 +84,28 @@ class TaskOrder(Base, mixins.TimestampsMixin):
@property
def budget(self):
return sum(
filter(None, [self.clin_01, self.clin_02, self.clin_03, self.clin_04])
)
# TODO: fix task order -- reimplement using CLINs
# Faked for display purposes
return 100000
@property
def balance(self):
# Faking the remaining balance using the stubbed reporting data for A-Wing & B-Wing
if (
self.portfolio_name in MockReportingProvider.REPORT_FIXTURE_MAP
and self.is_active
):
return self.budget - app.csp.reports.get_total_spending(self.portfolio)
# Faking an almost fully spent TO if the TO is expired
if self.is_expired:
return random.randrange(300) / 100 # nosec
# TODO: somehow calculate the remaining balance. For now, assume $0 spent
return self.budget
# TODO: fix task order -- reimplement using CLINs
# Faked for display purposes
return 50
@property
def display_status(self):
return self.status.value
@property
def portfolio_name(self):
return self.portfolio.name
@property
def defense_component(self):
return self.portfolio.defense_component
@property
def is_pending(self):
return self.status == Status.PENDING
@property
def ko_invitable(self):
"""
The MO has indicated that the KO should be invited but we have not sent
an invite and attached the KO user
"""
return self.ko_invite and not self.contracting_officer
@property
def cor_invitable(self):
"""
The MO has indicated that the COR should be invited but we have not sent
an invite and attached the COR user
"""
return self.cor_invite and not self.contracting_officer_representative
@property
def so_invitable(self):
"""
The MO has indicated that the SO should be invited but we have not sent
an invite and attached the SO user
"""
return self.so_invite and not self.security_officer
@property
def officers(self):
return [
self.contracting_officer,
self.contracting_officer_representative,
self.security_officer,
]
_OFFICER_PREFIXES = {
"contracting_officer": "ko",
"contracting_officer_representative": "cor",
"security_officer": "so",
}
_OFFICER_PROPERTIES = ["first_name", "last_name", "phone_number", "email", "dod_id"]
def officer_dictionary(self, officer_type):
prefix = self._OFFICER_PREFIXES[officer_type]
return {
field: getattr(self, "{}_{}".format(prefix, field))
for field in self._OFFICER_PROPERTIES
}
def to_dictionary(self):
return {
"portfolio_name": self.portfolio_name,
@ -254,6 +117,4 @@ class TaskOrder(Base, mixins.TimestampsMixin):
}
def __repr__(self):
return "<TaskOrder(number='{}', budget='{}', end_date='{}', id='{}')>".format(
self.number, self.budget, self.end_date, self.id
)
return "<TaskOrder(number='{}', id='{}')>".format(self.number, self.id)

2
atst/routes/task_orders/__init__.py
View File

@ -4,8 +4,6 @@ task_orders_bp = Blueprint("task_orders", __name__)
from . import index
from . import new
from . import invitations
from . import officer_reviews
from . import signing
from . import downloads
from atst.utils.context_processors import portfolio as portfolio_context_processor

9
atst/routes/task_orders/index.py
View File

@ -3,10 +3,9 @@ from collections import defaultdict
from flask import g, render_template, url_for
from . import task_orders_bp
from atst.domain.authz import Authorization
from atst.domain.authz.decorator import user_can_access_decorator as user_can
from atst.domain.portfolios import Portfolios
from atst.domain.task_orders import TaskOrders, DD254s
from atst.domain.task_orders import TaskOrders
from atst.models import Permissions
from atst.models.task_order import Status as TaskOrderStatus
@ -16,14 +15,8 @@ from atst.models.task_order import Status as TaskOrderStatus
def view_task_order(task_order_id):
task_order = TaskOrders.get(task_order_id)
to_form_complete = TaskOrders.all_sections_complete(task_order)
dd_254_complete = DD254s.is_complete(task_order.dd_254)
return render_template(
"portfolios/task_orders/show.html",
dd_254_complete=dd_254_complete,
is_cor=Authorization.is_cor(g.current_user, task_order),
is_ko=Authorization.is_ko(g.current_user, task_order),
is_so=Authorization.is_so(g.current_user, task_order),
is_to_signed=TaskOrders.is_signed_by_ko(task_order),
task_order=task_order,
to_form_complete=to_form_complete,
user=g.current_user,

132
atst/routes/task_orders/invitations.py
View File

@ -1,132 +0,0 @@
from flask import g, redirect, render_template, url_for, request as http_request
from . import task_orders_bp
from atst.domain.task_orders import TaskOrders
from atst.utils.flash import formatted_flash as flash
from atst.domain.authz.decorator import user_can_access_decorator as user_can
from atst.models.permissions import Permissions
from atst.database import db
from atst.domain.exceptions import NotFoundError, NoAccessError
from atst.domain.invitations import PortfolioInvitations
from atst.domain.portfolios import Portfolios
from atst.utils.localization import translate
from atst.forms.officers import EditTaskOrderOfficersForm
from atst.services.invitation import (
update_officer_invitations,
OFFICER_INVITATIONS,
Invitation as InvitationService,
)
@task_orders_bp.route("/task_orders/<task_order_id>/invite", methods=["POST"])
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS, message="invite task order officers")
def invite(task_order_id):
task_order = TaskOrders.get(task_order_id)
if TaskOrders.all_sections_complete(task_order):
update_officer_invitations(g.current_user, task_order)
portfolio = task_order.portfolio
flash("task_order_congrats", portfolio=portfolio)
return redirect(
url_for("task_orders.view_task_order", task_order_id=task_order.id)
)
else:
flash("task_order_incomplete")
return redirect(
url_for("task_orders.new", screen=4, task_order_id=task_order.id)
)
@task_orders_bp.route("/task_orders/<task_order_id>/resend_invite", methods=["POST"])
@user_can(
Permissions.EDIT_TASK_ORDER_DETAILS, message="resend task order officer invites"
)
def resend_invite(task_order_id):
invite_type = http_request.args.get("invite_type")
if invite_type not in OFFICER_INVITATIONS:
raise NotFoundError("invite_type")
invite_type_info = OFFICER_INVITATIONS[invite_type]
task_order = TaskOrders.get(task_order_id)
portfolio = Portfolios.get(g.current_user, task_order.portfolio_id)
officer = getattr(task_order, invite_type_info["role"])
if not officer:
raise NotFoundError("officer")
invitation = PortfolioInvitations.lookup_by_resource_and_user(portfolio, officer)
if not invitation:
raise NotFoundError("invitation")
if not invitation.can_resend:
raise NoAccessError("invitation")
PortfolioInvitations.revoke(token=invitation.token)
invite_service = InvitationService(
g.current_user,
invitation.role,
invitation.email,
subject=invite_type_info["subject"],
email_template=invite_type_info["template"],
)
invite_service.invite()
flash(
"invitation_resent",
officer_type=translate(
"common.officer_helpers.underscore_to_friendly.{}".format(
invite_type_info["role"]
)
),
)
return redirect(url_for("task_orders.invitations", task_order_id=task_order_id))
@task_orders_bp.route("/task_orders/<task_order_id>/invitations")
@user_can(
Permissions.EDIT_TASK_ORDER_DETAILS, message="view task order invitations page"
)
def invitations(task_order_id):
task_order = TaskOrders.get(task_order_id)
form = EditTaskOrderOfficersForm(obj=task_order)
if TaskOrders.all_sections_complete(task_order):
return render_template(
"portfolios/task_orders/invitations.html",
task_order=task_order,
form=form,
user=g.current_user,
)
else:
raise NotFoundError("task_order")
@task_orders_bp.route("/task_orders/<task_order_id>/invitations/edit", methods=["POST"])
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS, message="edit task order invitations")
def invitations_edit(task_order_id):
task_order = TaskOrders.get(task_order_id)
form = EditTaskOrderOfficersForm(formdata=http_request.form, obj=task_order)
if form.validate():
form.populate_obj(task_order)
db.session.add(task_order)
db.session.commit()
update_officer_invitations(g.current_user, task_order)
return redirect(url_for("task_orders.invitations", task_order_id=task_order.id))
else:
return (
render_template(
"portfolios/task_orders/invitations.html",
task_order=task_order,
form=form,
),
400,
)

117
atst/routes/task_orders/officer_reviews.py
View File

@ -1,117 +0,0 @@
from flask import g, redirect, render_template, url_for, request as http_request
from . import task_orders_bp
from atst.domain.authz import Authorization
from atst.domain.exceptions import NoAccessError
from atst.domain.task_orders import TaskOrders
from atst.forms.dd_254 import DD254Form
from atst.forms.ko_review import KOReviewForm
from atst.domain.authz.decorator import user_can_access_decorator as user_can
def wrap_check_is_ko_or_cor(user, task_order_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id)
Authorization.check_is_ko_or_cor(user, task_order)
return True
@task_orders_bp.route("/task_orders/<task_order_id>/review")
@user_can(
None,
override=wrap_check_is_ko_or_cor,
message="view contracting officer review form",
)
def ko_review(task_order_id):
task_order = TaskOrders.get(task_order_id)
if TaskOrders.all_sections_complete(task_order):
return render_template(
"/portfolios/task_orders/review.html",
task_order=task_order,
form=KOReviewForm(obj=task_order),
)
else:
raise NoAccessError("task_order")
@task_orders_bp.route("/task_orders/<task_order_id>/review", methods=["POST"])
@user_can(
None, override=wrap_check_is_ko_or_cor, message="submit contracting officer review"
)
def submit_ko_review(task_order_id, form=None):
task_order = TaskOrders.get(task_order_id)
form_data = {**http_request.form, **http_request.files}
form = KOReviewForm(form_data)
if form.validate():
TaskOrders.update(task_order=task_order, **form.data)
if Authorization.is_ko(g.current_user, task_order) and TaskOrders.can_ko_sign(
task_order
):
return redirect(
url_for("task_orders.signature_requested", task_order_id=task_order_id)
)
else:
return redirect(
url_for("task_orders.view_task_order", task_order_id=task_order_id)
)
else:
return render_template(
"/portfolios/task_orders/review.html", task_order=task_order, form=form
)
def so_review_form(task_order):
if task_order.dd_254:
dd_254 = task_order.dd_254
form = DD254Form(obj=dd_254)
form.required_distribution.data = dd_254.required_distribution
return form
else:
so = task_order.officer_dictionary("security_officer")
form_data = {
"certifying_official": "{}, {}".format(
so.get("last_name", ""), so.get("first_name", "")
),
"co_phone": so.get("phone_number", ""),
}
return DD254Form(data=form_data)
def wrap_check_is_so(user, task_order_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id)
Authorization.check_is_so(user, task_order)
return True
@task_orders_bp.route("/task_orders/<task_order_id>/dd254")
@user_can(None, override=wrap_check_is_so, message="view security officer review form")
def so_review(task_order_id):
task_order = TaskOrders.get(task_order_id)
form = so_review_form(task_order)
return render_template(
"portfolios/task_orders/so_review.html", form=form, task_order=task_order
)
@task_orders_bp.route("/task_orders/<task_order_id>/dd254", methods=["POST"])
@user_can(
None, override=wrap_check_is_so, message="submit security officer review form"
)
def submit_so_review(task_order_id):
task_order = TaskOrders.get(task_order_id)
form = DD254Form(http_request.form)
if form.validate():
TaskOrders.add_dd_254(task_order, form.data)
# TODO: will redirect to download, sign, upload page
return redirect(
url_for("task_orders.view_task_order", task_order_id=task_order.id)
)
else:
return render_template(
"portfolios/task_orders/so_review.html", form=form, task_order=task_order
)

10
atst/utils/context_processors.py
View File

@ -1,5 +1,3 @@
from operator import attrgetter
from flask import g
from sqlalchemy.orm.exc import NoResultFound
@ -103,9 +101,11 @@ def portfolio():
task_order for task_order in g.portfolio.task_orders if task_order.is_active
]
funding_end_date = (
sorted(active_task_orders, key=attrgetter("end_date"))[-1].end_date
if active_task_orders
else None
# TODO: fix task order -- reimplement logic to get end date from CLINs
# sorted(active_task_orders, key=attrgetter("end_date"))[-1].end_date
# if active_task_orders
# else None
None
)
funded = len(active_task_orders) > 1
else:

55
script/seed_sample.py
View File

@ -10,24 +10,27 @@ sys.path.append(parent_dir)
from atst.app import make_config, make_app
from atst.database import db
from atst.models.application import Application
from atst.models.environment_role import CSPRole
from atst.domain.application_roles import ApplicationRoles
from atst.domain.applications import Applications
from atst.domain.csp.reports import MockReportingProvider
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.exceptions import AlreadyExistsError, NotFoundError
from atst.domain.permission_sets import PermissionSets, APPLICATION_PERMISSION_SETS
from atst.domain.portfolio_roles import PortfolioRoles
from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.portfolios import Portfolios
from atst.domain.users import Users
from atst.models.application import Application
from atst.models.environment_role import CSPRole
from atst.routes.dev import _DEV_USERS as DEV_USERS
from tests.factories import (
TaskOrderFactory,
random_task_order_number,
random_service_branch,
random_task_order_number,
TaskOrderFactory,
)
fake = Faker()
@ -156,38 +159,16 @@ def add_members_to_portfolio(portfolio):
db.session.commit()
def add_task_orders_to_portfolio(portfolio, to_length=90, clin_01=None, clin_03=None):
active_to_offset = random.randint(10, 31)
# exp TO ends same day as active TO starts
active_start = date.today() - timedelta(days=active_to_offset)
# pending TO starts the same day active TO ends
active_end = active_start + timedelta(to_length)
pending_end = active_end + timedelta(to_length)
exp_start = active_start - timedelta(to_length)
create_task_order(portfolio, start=exp_start, end=active_start)
create_task_order(
portfolio, start=active_start, end=active_end, clin_01=clin_01, clin_03=clin_03
)
create_task_order(portfolio, start=active_end, end=pending_end)
def add_task_orders_to_portfolio(portfolio):
# TODO: after CLINs are implemented, vary the start/end dates of TOs
create_task_order(portfolio)
create_task_order(portfolio)
create_task_order(portfolio)
def create_task_order(portfolio, start, end, clin_01=None, clin_03=None):
default_kwargs = {
"start_date": start,
"end_date": end,
"number": random_task_order_number(),
"portfolio": portfolio,
"clin_02": 0,
"clin_04": 0,
}
if clin_01:
default_kwargs["clin_01"] = clin_01
if clin_03:
default_kwargs["clin_03"] = clin_03
task_order = TaskOrderFactory.build(**default_kwargs)
def create_task_order(portfolio):
# TODO: after CLINs are implemented add them to TO
task_order = TaskOrderFactory.build(portfolio=portfolio)
db.session.add(task_order)
db.session.commit()
@ -255,10 +236,8 @@ def create_demo_portfolio(name, data):
portfolio = Portfolios.create(
portfolio_owner, name=name, defense_component=random_service_branch()
)
clin_01 = data["budget"] * 0.8
clin_03 = data["budget"] * 0.2
add_task_orders_to_portfolio(portfolio, clin_01=clin_01, clin_03=clin_03)
add_task_orders_to_portfolio(portfolio)
add_members_to_portfolio(portfolio)
for mock_application in data["applications"]:

18
templates/portfolios/header.html
View File

@ -27,31 +27,15 @@
</div>
<div class='portfolio-header__budget--amount'>
<span class='portfolio-header__budget--dollars'>
{{ portfolio.task_orders | selectattr('is_active') | sum(attribute='balance') | justDollars }}
</span>
<span class='portfolio-header__budget--cents'>
.{{ portfolio.task_orders | selectattr('is_active') | sum(attribute='balance') | justCents }}
</span>
</div>
</div>
<div class='row'>
<div class='column-left'></div>
<div class='column-right portfolio-funding__header--funded-through {{ "funded" if funding_end_date is not none and funded else "unfunded"}}'>
{% if funding_end_date and funded %}
{{ Icon('ok') }}
Funded through
<local-datetime
timestamp='{{ funding_end_date }}'
format="M/D/YYYY">
</local-datetime>
{% elif funding_end_date and not funded %}
{{ Icon('alert') }}
Funded period ends
<local-datetime
timestamp='{{ funding_end_date }}'
format="M/D/YYYY">
</local-datetime>
{% endif %}
</div>
</div>
</div>

82
templates/portfolios/task_orders/show.html
View File

@ -6,10 +6,6 @@
{% block portfolio_content %}
{% set show_dd_254_button = is_so and to_form_complete %}
{% set show_to_info_button = (is_cor or is_ko) and to_form_complete %}
{% set show_sign_to_button = is_ko and dd_254_complete and not is_to_signed %}
{% macro officer_name(officer) -%}
{%- if not officer -%}
not yet invited
@ -73,31 +69,6 @@
</div>
{% endmacro %}
{% macro InvitationStatus(title, officer, officer_info) %}
{% set class = "invited" if officer else "uninvited" %}
<div class="task-order-invitation-status row">
<div class="task-order-invitation-status__icon col">
<span>{{ Icon("avatar" if officer else "alert", classes=class) }}</span>
</div>
<div class="col">
<div class="task-order-invitation-status__title {{ class }}">
{{ title }}
</div>
<div class="task-order-invitation-details">
{% if officer_info %}
<div class="col">
<div>{{ officer_info.first_name }} {{ officer_info.last_name }}</div>
<div>{{ officer_info.email }}</div>
<div>{{ officer_info.phone_number | usPhone }}</div>
</div>
{% else %}
<span>not yet invited</span>
{% endif %}
</div>
</div>
</div>
{% endmacro %}
<div class="task-order-summary">
{% include "fragments/flash.html" %}
@ -138,48 +109,6 @@
})| safe,
)
}}
{{
Step(
button_text=show_dd_254_button and ("common.edit" | translate),
button_url=show_dd_254_button and url_for("task_orders.so_review", task_order_id=task_order.id),
complete=dd_254_complete,
description="task_orders.view.steps.security" | translate({
"security_officer": officer_name(task_order.security_officer)
}) | safe,
)
}}
{% call Step(
description="task_orders.view.steps.record" | translate({
"contracting_officer": officer_name(task_order.contracting_officer),
"contracting_officer_representative": officer_name(task_order.contracting_officer_representative)
}) | safe,
button_url=show_to_info_button and url_for(
"task_orders.ko_review",
task_order_id=task_order.id,
),
button_text=show_to_info_button and ("common.edit" | translate),
complete=False) %}
<div class='alert alert--warning'>
<div class='alert__content'>
<div class='alert__message'>
{{ "task_orders.view.steps.record_description" | translate | safe }}
</div>
</div>
</div>
{% endcall %}
{{
Step(
button_text=show_sign_to_button and ("common.sign" | translate),
button_url=show_sign_to_button and url_for(
"task_orders.ko_review",
task_order_id=task_order.id,
),
complete=is_to_signed,
description="task_orders.view.steps.sign" | translate({
"contracting_officer": officer_name(task_order.contracting_officer)
}) | safe,
)
}}
</div>
</div>
<div class="task-order-sidebar col">
@ -209,15 +138,9 @@
{{ DocumentLink(
title="Instruction Sheet",
link_url="#") }}
{{ DocumentLink(
title="Cloud Services Estimate",
link_url=task_order.csp_estimate and url_for("task_orders.download_csp_estimate", task_order_id=task_order.id) ) }}
{{ DocumentLink(
title="Market Research",
link_url="#") }}
{{ DocumentLink(
title="DD 254",
link_url="") }}
</div>
</div>
<div class="task-order-invitations panel">
@ -225,15 +148,12 @@
<div class="task-order-invitations__heading row">
<h3>Invitations</h3>
{% if to_form_complete %}
<a href="{{ url_for('task_orders.invitations', task_order_id=task_order.id) }}" class="icon-link">
<a href="#" class="icon-link">
<span>{{ "common.manage" | translate }}</span>
{{ Icon("edit") }}
</a>
{% endif %}
</div>
{{ InvitationStatus('Contracting Officer', task_order.contracting_officer, officer_info=task_order.officer_dictionary('contracting_officer')) }}
{{ InvitationStatus('Contracting Officer Representative', task_order.contracting_officer_representative, officer_info=task_order.officer_dictionary('contracting_officer_representative')) }}
{{ InvitationStatus('IA Security Officer', officer=task_order.security_officer, officer_info=task_order.officer_dictionary('security_officer')) }}
</div>
</div>
</div>

27
tests/domain/test_authz.py
View File

@ -27,33 +27,6 @@ def task_order():
return TaskOrderFactory.create()
def test_is_ko(task_order, invalid_user):
assert not Authorization.is_ko(invalid_user, task_order)
assert Authorization.is_ko(task_order.contracting_officer, task_order)
def test_is_cor(task_order, invalid_user):
assert not Authorization.is_cor(invalid_user, task_order)
assert Authorization.is_cor(
task_order.contracting_officer_representative, task_order
)
def test_is_so(task_order, invalid_user):
assert Authorization.is_so(task_order.security_officer, task_order)
assert not Authorization.is_so(invalid_user, task_order)
def test_check_is_ko_or_cor(task_order, invalid_user):
assert Authorization.check_is_ko_or_cor(
task_order.contracting_officer_representative, task_order
)
assert Authorization.check_is_ko_or_cor(task_order.contracting_officer, task_order)
with pytest.raises(UnauthorizedError):
Authorization.check_is_ko_or_cor(invalid_user, task_order)
def test_has_portfolio_permission():
role_one = PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING)
role_two = PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_REPORTS)

54
tests/domain/test_task_orders.py
View File

@ -1,6 +1,6 @@
import pytest
from atst.domain.task_orders import TaskOrders, TaskOrderError, DD254s
from atst.domain.task_orders import TaskOrders, TaskOrderError
from atst.domain.exceptions import UnauthorizedError
from atst.domain.permission_sets import PermissionSets
from atst.domain.portfolio_roles import PortfolioRoles
@ -11,27 +11,15 @@ from tests.factories import (
UserFactory,
PortfolioRoleFactory,
PortfolioFactory,
DD254Factory,
)
def test_is_signed_by_ko():
user = UserFactory.create()
task_order = TaskOrderFactory.create(contracting_officer=user)
assert not TaskOrders.is_signed_by_ko(task_order)
TaskOrders.update(task_order, signer_dod_id=user.dod_id)
assert TaskOrders.is_signed_by_ko(task_order)
@pytest.mark.skip(reason="Need to reimplement after new TO form is created")
def test_section_completion_status():
dict_keys = [k for k in TaskOrders.SECTIONS.keys()]
section = dict_keys[0]
attrs = TaskOrders.SECTIONS[section].copy()
attrs.remove("portfolio_name")
attrs.remove("defense_component")
task_order = TaskOrderFactory.create(**{k: None for k in attrs})
leftover = attrs.pop()
@ -43,6 +31,7 @@ def test_section_completion_status():
assert TaskOrders.section_completion_status(task_order, section) == "complete"
@pytest.mark.skip(reason="Need to reimplement after new TO form is created")
def test_all_sections_complete():
task_order = TaskOrderFactory.create()
attachment = Attachment(
@ -62,40 +51,3 @@ def test_all_sections_complete():
assert not TaskOrders.all_sections_complete(task_order)
task_order.scope = "str12345"
assert TaskOrders.all_sections_complete(task_order)
def test_add_officer():
task_order = TaskOrderFactory.create()
ko = UserFactory.create()
owner = task_order.portfolio.owner
TaskOrders.add_officer(task_order, "contracting_officer", ko.to_dictionary())
assert task_order.contracting_officer == ko
portfolio_users = [ws_role.user for ws_role in task_order.portfolio.members]
assert ko in portfolio_users
def test_add_officer_with_nonexistent_role():
task_order = TaskOrderFactory.create()
ko = UserFactory.create()
owner = task_order.portfolio.owner
with pytest.raises(TaskOrderError):
TaskOrders.add_officer(task_order, "pilot", ko.to_dictionary())
def test_add_officer_who_is_already_portfolio_member():
task_order = TaskOrderFactory.create()
owner = task_order.portfolio.owner
TaskOrders.add_officer(task_order, "contracting_officer", owner.to_dictionary())
assert task_order.contracting_officer == owner
member = task_order.portfolio.members[0]
assert member.user == owner
def test_dd254_complete():
finished = DD254Factory.create()
unfinished = DD254Factory.create(certifying_official=None)
assert DD254s.is_complete(finished)
assert not DD254s.is_complete(unfinished)

51
tests/factories.py
View File

@ -259,56 +259,7 @@ class TaskOrderFactory(Base):
model = TaskOrder
portfolio = factory.SubFactory(PortfolioFactory)
clin_01 = factory.LazyFunction(lambda *args: random.randrange(100, 100_000))
clin_03 = factory.LazyFunction(lambda *args: random.randrange(100, 100_000))
clin_02 = factory.LazyFunction(lambda *args: random.randrange(100, 100_000))
clin_04 = factory.LazyFunction(lambda *args: random.randrange(100, 100_000))
app_migration = random_choice(data.APP_MIGRATION)
native_apps = random.choice(["yes", "no", "not_sure"])
complexity = [random_choice(data.APPLICATION_COMPLEXITY)]
dev_team = [random_choice(data.DEV_TEAM)]
team_experience = random_choice(data.TEAM_EXPERIENCE)
scope = factory.Faker("sentence")
start_date = factory.LazyFunction(
lambda *args: random_future_date(year_min=1, year_max=1)
)
end_date = factory.LazyFunction(
lambda *args: random_future_date(year_min=2, year_max=5)
)
performance_length = random.randint(1, 24)
csp_estimate = factory.SubFactory(AttachmentFactory)
ko_first_name = factory.Faker("first_name")
ko_last_name = factory.Faker("last_name")
ko_email = factory.Faker("email")
ko_phone_number = factory.LazyFunction(random_phone_number)
ko_dod_id = factory.LazyFunction(random_dod_id)
cor_first_name = factory.Faker("first_name")
cor_last_name = factory.Faker("last_name")
cor_email = factory.Faker("email")
cor_phone_number = factory.LazyFunction(random_phone_number)
cor_dod_id = factory.LazyFunction(random_dod_id)
so_first_name = factory.Faker("first_name")
so_last_name = factory.Faker("last_name")
so_email = factory.Faker("email")
so_phone_number = factory.LazyFunction(random_phone_number)
so_dod_id = factory.LazyFunction(random_dod_id)
class DD254Factory(Base):
class Meta:
model = DD254
certifying_official = factory.Faker("name")
certifying_official_title = factory.Faker("job")
certifying_official_address = factory.Faker("address")
certifying_official_phone = factory.LazyFunction(random_phone_number)
required_distribution = factory.LazyFunction(
lambda: [random_choice(data.REQUIRED_DISTRIBUTIONS)]
)
number = factory.LazyFunction(random_task_order_number)
class NotificationRecipientFactory(Base):

63
tests/forms/test_officers.py
View File

@ -1,63 +0,0 @@
from werkzeug.datastructures import ImmutableMultiDict
from atst.forms.officers import EditTaskOrderOfficersForm
from tests.factories import TaskOrderFactory, UserFactory
class TestEditTaskOrderOfficersForm:
def _assert_officer_info_matches(self, form, task_order, officer):
prefix = form.OFFICER_PREFIXES[officer]
for field in form.OFFICER_INFO_FIELD_NAMES:
assert form[officer][field].data == getattr(
task_order, "{}_{}".format(prefix, field)
)
def test_processing_with_existing_task_order(self):
task_order = TaskOrderFactory.create()
form = EditTaskOrderOfficersForm(obj=task_order)
for officer in form.OFFICER_PREFIXES.keys():
self._assert_officer_info_matches(form, task_order, officer)
def test_processing_form_with_formdata(self):
data = {
"contracting_officer-first_name": "Han",
"contracting_officer-last_name": "Solo",
}
formdata = ImmutableMultiDict(data)
task_order = TaskOrderFactory.create()
form = EditTaskOrderOfficersForm(formdata=formdata, obj=task_order)
for officer in ["contracting_officer_representative", "security_officer"]:
self._assert_officer_info_matches(form, task_order, officer)
prefix = "ko"
officer = "contracting_officer"
for field in form.OFFICER_INFO_FIELD_NAMES:
data_field = "{}-{}".format(officer, field)
if data_field in formdata:
assert form[officer][field].data == formdata[data_field]
else:
assert form[officer][field].data == getattr(
task_order, "{}_{}".format(prefix, field)
)
def test_populate_obj(self):
data = {
"security_officer-first_name": "Luke",
"security_officer-last_name": "Skywalker",
}
formdata = ImmutableMultiDict(data)
task_order = TaskOrderFactory.create()
form = EditTaskOrderOfficersForm(formdata=formdata, obj=task_order)
form.populate_obj(task_order)
assert task_order.so_first_name == data["security_officer-first_name"]
assert task_order.so_last_name == data["security_officer-last_name"]
def test_email_validation(self):
data = {"contracting_officer-email": "not_really_an_email_address"}
formdata = ImmutableMultiDict(data)
form = EditTaskOrderOfficersForm(formdata)
assert not form.validate()
assert form.contracting_officer.email.errors == ["Invalid email address."]

66
tests/models/test_task_order.py
View File

@ -9,79 +9,27 @@ from tests.mocks import PDF_FILENAME
class TestTaskOrderStatus:
@pytest.mark.skip(reason="Reimplement after adding CLINs")
def test_started_status(self):
to = TaskOrder()
assert to.status == Status.STARTED
@pytest.mark.skip(reason="See if still needed after implementing CLINs")
def test_pending_status(self):
to = TaskOrder(
number="42", start_date=random_future_date(), end_date=random_future_date()
)
to = TaskOrder(number="42")
assert to.status == Status.PENDING
@pytest.mark.skip(reason="See if still needed after implementing CLINs")
def test_active_status(self):
to = TaskOrder(
number="42", start_date=random_past_date(), end_date=random_future_date()
)
to = TaskOrder(number="42")
assert to.status == Status.ACTIVE
@pytest.mark.skip(reason="See if still needed after implementing CLINs")
def test_expired_status(self):
to = TaskOrder(
number="42", start_date=random_past_date(), end_date=random_past_date()
)
to = TaskOrder(number="42")
assert to.status == Status.EXPIRED
def test_is_submitted():
to = TaskOrder()
assert not to.is_submitted
to = TaskOrder(
number="42",
start_date=datetime.date.today(),
end_date=datetime.date.today() + datetime.timedelta(days=1),
)
assert to.is_submitted
class TestCSPEstimate:
def test_setting_estimate_with_attachment(self):
to = TaskOrder()
attachment = Attachment(filename="sample.pdf", object_name="sample")
to.csp_estimate = attachment
assert to.csp_attachment_id == attachment.id
def test_setting_estimate_with_file_storage(self):
to = TaskOrder()
with open(PDF_FILENAME, "rb") as fp:
fs = FileStorage(fp, content_type="application/pdf")
to.csp_estimate = fs
assert to.csp_estimate is not None
assert to.csp_estimate.filename == PDF_FILENAME
def test_setting_estimate_with_invalid_object(self):
to = TaskOrder()
with pytest.raises(TypeError):
to.csp_estimate = "invalid"
def test_setting_estimate_with_empty_value(self):
to = TaskOrder()
assert to.csp_estimate is None
to.csp_estimate = ""
assert to.csp_estimate is None
def test_removing_estimate(self):
attachment = Attachment(filename="sample.pdf", object_name="sample")
to = TaskOrder(csp_estimate=attachment)
assert to.csp_estimate is not None
to.csp_estimate = ""
assert to.csp_estimate is None
class TestPDF:
def test_setting_pdf_with_attachment(self):
to = TaskOrder()

7
tests/routes/portfolios/test_index.py
View File

@ -42,12 +42,7 @@ def test_portfolio_reports(client, user_session):
{"name": "application1", "environments": [{"name": "application1 prod"}]}
]
)
task_order = TaskOrderFactory.create(
number="42",
start_date=random_past_date(),
end_date=random_future_date(),
portfolio=portfolio,
)
task_order = TaskOrderFactory.create(number="42", portfolio=portfolio)
user_session(portfolio.owner)
response = client.get(url_for("portfolios.reports", portfolio_id=portfolio.id))
assert response.status_code == 200

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

@ -279,111 +279,3 @@ def test_existing_member_invite_resent_to_email_submitted_in_form(
assert user.email != "example@example.com"
assert send_mail_job.func.__func__.__name__ == "_send_mail"
assert send_mail_job.args[0] == ["example@example.com"]
def test_contracting_officer_accepts_invite(monkeypatch, client, user_session):
portfolio = PortfolioFactory.create()
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.invite", task_order_id=task_order.id))
# contracting officer accepts invitation
user = Users.get_by_dod_id(user_info["dod_id"])
token = user.portfolio_invitations[0].token
monkeypatch.setattr(
"atst.domain.auth.should_redirect_to_user_profile", lambda *args: False
)
user_session(user)
response = client.get(
url_for("portfolios.accept_invitation", portfolio_token=token)
)
# user is redirected to the task order review page
assert response.status_code == 302
to_review_url = url_for(
"task_orders.view_task_order", task_order_id=task_order.id, _external=True
)
assert response.headers["Location"] == to_review_url
def test_cor_accepts_invite(monkeypatch, client, user_session):
portfolio = PortfolioFactory.create()
user_info = UserFactory.dictionary()
task_order = TaskOrderFactory.create(
portfolio=portfolio,
cor_first_name=user_info["first_name"],
cor_last_name=user_info["last_name"],
cor_email=user_info["email"],
cor_phone_number=user_info["phone_number"],
cor_dod_id=user_info["dod_id"],
cor_invite=True,
)
# create contracting officer representative
user_session(portfolio.owner)
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
# contracting officer representative accepts invitation
user = Users.get_by_dod_id(user_info["dod_id"])
token = user.portfolio_invitations[0].token
monkeypatch.setattr(
"atst.domain.auth.should_redirect_to_user_profile", lambda *args: False
)
user_session(user)
response = client.get(
url_for("portfolios.accept_invitation", portfolio_token=token)
)
# user is redirected to the task order review page
assert response.status_code == 302
to_review_url = url_for(
"task_orders.view_task_order", task_order_id=task_order.id, _external=True
)
assert response.headers["Location"] == to_review_url
def test_so_accepts_invite(monkeypatch, client, user_session):
portfolio = PortfolioFactory.create()
user_info = UserFactory.dictionary()
task_order = TaskOrderFactory.create(
portfolio=portfolio,
so_first_name=user_info["first_name"],
so_last_name=user_info["last_name"],
so_email=user_info["email"],
so_phone_number=user_info["phone_number"],
so_dod_id=user_info["dod_id"],
so_invite=True,
)
# create security officer
user_session(portfolio.owner)
client.post(url_for("task_orders.invite", task_order_id=task_order.id))
# security officer accepts invitation
user = Users.get_by_dod_id(user_info["dod_id"])
token = user.portfolio_invitations[0].token
monkeypatch.setattr(
"atst.domain.auth.should_redirect_to_user_profile", lambda *args: False
)
user_session(user)
response = client.get(
url_for("portfolios.accept_invitation", portfolio_token=token)
)
# user is redirected to the task order review page
assert response.status_code == 302
to_review_url = url_for(
"task_orders.view_task_order", task_order_id=task_order.id, _external=True
)
assert response.headers["Location"] == to_review_url

43
tests/routes/task_orders/test_downloads.py
View File

@ -29,46 +29,3 @@ def test_download_summary(client, user_session):
for attr, val in task_order.to_dictionary().items():
assert attr in doc
assert xml_translated(val) in doc
class TestDownloadCSPEstimate:
def setup(self):
self.user = UserFactory.create()
self.portfolio = PortfolioFactory.create(owner=self.user)
self.task_order = TaskOrderFactory.create(
creator=self.user, portfolio=self.portfolio
)
def test_successful_download(self, client, user_session, pdf_upload):
self.task_order.csp_estimate = pdf_upload
user_session(self.user)
response = client.get(
url_for(
"task_orders.download_csp_estimate", task_order_id=self.task_order.id
)
)
assert response.status_code == 200
pdf_upload.seek(0)
expected_contents = pdf_upload.read()
assert expected_contents == response.data
def test_download_without_attachment(self, client, user_session):
self.task_order.csp_attachment_id = None
user_session(self.user)
response = client.get(
url_for(
"task_orders.download_csp_estimate", task_order_id=self.task_order.id
)
)
assert response.status_code == 404
def test_download_with_wrong_user(self, client, user_session):
other_user = UserFactory.create()
user_session(other_user)
response = client.get(
url_for(
"task_orders.download_csp_estimate", task_order_id=self.task_order.id
)
)
assert response.status_code == 404

65
tests/routes/task_orders/test_index.py
View File

@ -45,22 +45,13 @@ class TestPortfolioFunding:
assert context["active_task_orders"] == []
assert context["expired_task_orders"] == []
@pytest.mark.skip(reason="Update later when CLINs are implemented")
def test_funded_portfolio(self, app, user_session, portfolio):
user_session(portfolio.owner)
pending_to = TaskOrderFactory.create(portfolio=portfolio)
active_to1 = TaskOrderFactory.create(
portfolio=portfolio,
start_date=random_past_date(),
end_date=random_future_date(),
number="42",
)
active_to2 = TaskOrderFactory.create(
portfolio=portfolio,
start_date=random_past_date(),
end_date=random_future_date(),
number="43",
)
active_to1 = TaskOrderFactory.create(portfolio=portfolio, number="42")
active_to2 = TaskOrderFactory.create(portfolio=portfolio, number="43")
end_date = (
active_to1.end_date
if active_to1.end_date > active_to2.end_date
@ -77,21 +68,12 @@ class TestPortfolioFunding:
assert context["funding_end_date"] is end_date
assert context["total_balance"] == active_to1.budget + active_to2.budget
@pytest.mark.skip(reason="Update later when CLINs are implemented")
def test_expiring_and_funded_portfolio(self, app, user_session, portfolio):
user_session(portfolio.owner)
expiring_to = TaskOrderFactory.create(
portfolio=portfolio,
start_date=random_past_date(),
end_date=(date.today() + timedelta(days=10)),
number="42",
)
active_to = TaskOrderFactory.create(
portfolio=portfolio,
start_date=random_past_date(),
end_date=random_future_date(year_min=1, year_max=2),
number="43",
)
expiring_to = TaskOrderFactory.create(portfolio=portfolio, number="42")
active_to = TaskOrderFactory.create(portfolio=portfolio, number="43")
with captured_templates(app) as templates:
response = app.test_client().get(
@ -103,15 +85,11 @@ class TestPortfolioFunding:
assert context["funding_end_date"] is active_to.end_date
assert context["funded"] == True
@pytest.mark.skip(reason="Update later when CLINs are implemented")
def test_expiring_and_unfunded_portfolio(self, app, user_session, portfolio):
user_session(portfolio.owner)
expiring_to = TaskOrderFactory.create(
portfolio=portfolio,
start_date=random_past_date(),
end_date=(date.today() + timedelta(days=10)),
number="42",
)
expiring_to = TaskOrderFactory.create(portfolio=portfolio, number="42")
with captured_templates(app) as templates:
response = app.test_client().get(
@ -132,30 +110,3 @@ class TestPortfolioFunding:
url_for("task_orders.view_task_order", task_order_id=other_task_order.id)
)
assert response.status_code == 404
def test_ko_can_view_task_order(client, user_session, portfolio, user):
PortfolioRoleFactory.create(
portfolio=portfolio,
user=user,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=user)
user_session(user)
response = client.get(
url_for("task_orders.view_task_order", task_order_id=task_order.id)
)
assert response.status_code == 200
assert translate("common.manage") in response.data.decode()
TaskOrders.update(task_order, clin_01=None)
response = client.get(
url_for("task_orders.view_task_order", task_order_id=task_order.id)
)
assert response.status_code == 200
assert translate("common.manage") not in response.data.decode()

449
tests/routes/task_orders/test_invitations.py
View File

@ -1,449 +0,0 @@
from datetime import datetime, timedelta
from flask import url_for
import pytest
from atst.domain.task_orders import TaskOrders
from atst.models import InvitationStatus
from atst.models.portfolio_role import Status as PortfolioStatus
from atst.queue import queue
from tests.factories import (
PortfolioFactory,
TaskOrderFactory,
UserFactory,
PortfolioRoleFactory,
PortfolioInvitationFactory,
)
def test_invite(client, user_session):
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
to = TaskOrderFactory.create(portfolio=portfolio)
response = client.post(
url_for("task_orders.invite", task_order_id=to.id), follow_redirects=False
)
redirect = url_for("task_orders.view_task_order", 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
# 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.portfolio_invitations) == 1
def test_does_not_invite_if_task_order_incomplete(client, user_session, queue):
task_order = TaskOrderFactory.create(
scope=None, ko_invite=True, cor_invite=True, so_invite=True
)
portfolio = task_order.portfolio
user_session(portfolio.owner)
response = client.post(url_for("task_orders.invite", task_order_id=task_order.id))
# redirected to review screen
assert response.headers["Location"] == url_for(
"task_orders.new", screen=4, task_order_id=task_order.id, _external=True
)
# only owner is portfolio member
assert len(portfolio.members) == 1
# no email invitations are enqueued
assert len(queue.get_queue()) == 0
@pytest.fixture
def portfolio():
return PortfolioFactory.create()
@pytest.fixture
def user():
return UserFactory.create()
class TestTaskOrderInvitations:
def setup(self):
self.portfolio = PortfolioFactory.create()
self.task_order = TaskOrderFactory.create(portfolio=self.portfolio)
def _post(self, client, updates):
return client.post(
url_for("task_orders.invitations_edit", task_order_id=self.task_order.id),
headers={"Content-Type": "application/x-www-form-urlencoded"},
data=updates,
)
def test_editing_with_partial_data(self, user_session, client):
queue_length = len(queue.get_queue())
user_session(self.portfolio.owner)
response = self._post(
client,
{
"contracting_officer-first_name": "Luke",
"contracting_officer-last_name": "Skywalker",
"security_officer-first_name": "Boba",
"security_officer-last_name": "Fett",
},
)
updated_task_order = TaskOrders.get(self.task_order.id)
assert updated_task_order.ko_first_name == "Luke"
assert updated_task_order.ko_last_name == "Skywalker"
assert updated_task_order.so_first_name == "Boba"
assert updated_task_order.so_last_name == "Fett"
assert len(queue.get_queue()) == queue_length
assert response.status_code == 302
assert (
url_for(
"task_orders.invitations",
task_order_id=self.task_order.id,
_external=True,
)
== response.headers["Location"]
)
def test_editing_with_complete_data(self, user_session, client):
queue_length = len(queue.get_queue())
user_session(self.portfolio.owner)
response = self._post(
client,
{
"contracting_officer-first_name": "Luke",
"contracting_officer-last_name": "Skywalker",
"contracting_officer-dod_id": "0123456789",
"contracting_officer-email": "luke@skywalker.mil",
"contracting_officer-phone_number": "0123456789",
"contracting_officer-invite": "y",
},
)
updated_task_order = TaskOrders.get(self.task_order.id)
assert updated_task_order.ko_invite == True
assert updated_task_order.ko_first_name == "Luke"
assert updated_task_order.ko_last_name == "Skywalker"
assert updated_task_order.ko_email == "luke@skywalker.mil"
assert updated_task_order.ko_phone_number == "0123456789"
assert len(queue.get_queue()) == queue_length + 1
assert response.status_code == 302
assert (
url_for(
"task_orders.invitations",
task_order_id=self.task_order.id,
_external=True,
)
== response.headers["Location"]
)
def test_editing_with_invalid_data(self, user_session, client):
queue_length = len(queue.get_queue())
user_session(self.portfolio.owner)
response = self._post(
client,
{
"contracting_officer-phone_number": "invalid input",
"security_officer-first_name": "Boba",
"security_officer-last_name": "Fett",
},
)
assert "There were some errors" in response.data.decode()
updated_task_order = TaskOrders.get(self.task_order.id)
assert updated_task_order.so_first_name != "Boba"
assert len(queue.get_queue()) == queue_length
assert response.status_code == 400
def test_user_can_only_invite_to_task_order_in_their_portfolio(
self, user_session, client, portfolio
):
other_task_order = TaskOrderFactory.create()
user_session(portfolio.owner)
# user can't see invites
response = client.get(
url_for("task_orders.invitations", task_order_id=other_task_order.id)
)
assert response.status_code == 404
# user can't send invites
time_updated = other_task_order.time_updated
response = client.post(
url_for("task_orders.invitations_edit", task_order_id=other_task_order.id),
data={
"contracting_officer-first_name": "Luke",
"contracting_officer-last_name": "Skywalker",
"contracting_officer-dod_id": "0123456789",
"contracting_officer-email": "luke@skywalker.mil",
"contracting_officer-phone_number": "0123456789",
"contracting_officer-invite": "y",
},
)
assert response.status_code == 404
assert time_updated == other_task_order.time_updated
# user can't resend invites
response = client.post(
url_for(
"task_orders.resend_invite",
task_order_id=other_task_order.id,
invite_type="ko_invite",
)
)
assert response.status_code == 404
assert time_updated == other_task_order.time_updated
def test_does_not_render_resend_invite_if_user_is_mo_and_user_is_cor(
self, client, user_session
):
task_order = TaskOrderFactory.create(
portfolio=self.portfolio,
creator=self.portfolio.owner,
cor_first_name=self.portfolio.owner.first_name,
cor_last_name=self.portfolio.owner.last_name,
cor_email=self.portfolio.owner.email,
cor_phone_number=self.portfolio.owner.phone_number,
cor_dod_id=self.portfolio.owner.dod_id,
cor_invite=True,
)
user_session(self.portfolio.owner)
response = client.get(
url_for("task_orders.invitations", task_order_id=task_order.id)
)
assert "Resend Invitation" not in response.data.decode()
def test_renders_resend_invite_if_user_is_mo_and_user_is_not_cor(
self, client, user_session
):
cor = UserFactory.create()
task_order = TaskOrderFactory.create(
portfolio=self.portfolio,
creator=self.portfolio.owner,
contracting_officer_representative=cor,
cor_invite=True,
)
portfolio_role = PortfolioRoleFactory.create(portfolio=self.portfolio, user=cor)
PortfolioInvitationFactory.create(
inviter=self.portfolio.owner,
role=portfolio_role,
user=cor,
status=InvitationStatus.PENDING,
)
user_session(self.portfolio.owner)
response = client.get(
url_for("task_orders.invitations", task_order_id=task_order.id)
)
assert "Resend Invitation" in response.data.decode()
def test_can_view_task_order_invitations_when_complete(client, user_session, portfolio):
user_session(portfolio.owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)
response = client.get(
url_for("task_orders.invitations", task_order_id=task_order.id)
)
assert response.status_code == 200
def test_cant_view_task_order_invitations_when_not_complete(
client, user_session, portfolio
):
user_session(portfolio.owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, clin_01=None)
response = client.get(
url_for("task_orders.invitations", task_order_id=task_order.id)
)
assert response.status_code == 404
def test_resend_invite_when_invalid_invite_officer(
app, client, user_session, portfolio, user
):
queue_length = len(queue.get_queue())
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=user, ko_invite=True
)
PortfolioRoleFactory.create(
portfolio=portfolio, user=user, status=PortfolioStatus.ACTIVE
)
user_session(user)
response = client.post(
url_for(
"task_orders.resend_invite", task_order_id=task_order.id, _external=True
),
data={"invite_type": "invalid_invite_type"},
)
assert response.status_code == 404
assert len(queue.get_queue()) == queue_length
def test_resend_invite_when_officer_type_missing(
app, client, user_session, portfolio, user
):
queue_length = len(queue.get_queue())
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=None, ko_invite=True
)
PortfolioRoleFactory.create(
portfolio=portfolio, user=user, status=PortfolioStatus.ACTIVE
)
user_session(user)
response = client.post(
url_for(
"task_orders.resend_invite", task_order_id=task_order.id, _external=True
),
data={"invite_type": "contracting_officer_invite"},
)
assert response.status_code == 404
assert len(queue.get_queue()) == queue_length
def test_resend_invite_when_not_pending(app, client, user_session, portfolio, user):
queue_length = len(queue.get_queue())
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=user, ko_invite=True
)
portfolio_role = PortfolioRoleFactory.create(
portfolio=portfolio, user=user, status=PortfolioStatus.ACTIVE
)
original_invitation = PortfolioInvitationFactory.create(
inviter=user,
role=portfolio_role,
email=user.email,
status=InvitationStatus.ACCEPTED,
)
user_session(user)
response = client.post(
url_for(
"task_orders.resend_invite", task_order_id=task_order.id, _external=True
),
data={"invite_type": "ko_invite"},
)
assert original_invitation.status == InvitationStatus.ACCEPTED
assert response.status_code == 404
assert len(queue.get_queue()) == queue_length
def test_resending_revoked_invite(app, client, user_session, portfolio, user):
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=user, ko_invite=True
)
portfolio_role = PortfolioRoleFactory.create(portfolio=portfolio, user=user)
invite = PortfolioInvitationFactory.create(
inviter=user,
role=portfolio_role,
email=user.email,
status=InvitationStatus.REVOKED,
)
user_session(user)
response = client.post(
url_for(
"task_orders.resend_invite",
task_order_id=task_order.id,
invite_type="ko_invite",
_external=True,
)
)
assert invite.is_revoked
assert response.status_code == 404
def test_resending_expired_invite(app, client, user_session, portfolio):
queue_length = len(queue.get_queue())
ko = UserFactory.create()
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=ko, ko_invite=True
)
portfolio_role = PortfolioRoleFactory.create(portfolio=portfolio, user=ko)
invite = PortfolioInvitationFactory.create(
inviter=portfolio.owner,
role=portfolio_role,
email=ko.email,
expiration_time=datetime.now() - timedelta(days=1),
)
user_session(portfolio.owner)
response = client.post(
url_for(
"task_orders.resend_invite",
task_order_id=task_order.id,
invite_type="ko_invite",
_external=True,
)
)
assert invite.is_expired
assert response.status_code == 302
assert len(queue.get_queue()) == queue_length + 1

123
tests/routes/task_orders/test_new.py
View File

@ -109,16 +109,13 @@ def test_to_on_pf_cannot_edit_pf_attributes():
assert second_workflow.pf_attributes_read_only
@pytest.mark.skip(reason="Reimplement after TO form is updated")
def test_create_new_task_order(client, user_session, pdf_upload):
creator = UserFactory.create()
user_session(creator)
task_order_data = TaskOrderFactory.dictionary()
app_info_data = slice_data_for_section(task_order_data, "app_info")
portfolio_name = "Mos Eisley"
defense_component = "Defense Health Agency"
app_info_data["portfolio_name"] = portfolio_name
app_info_data["defense_component"] = defense_component
response = client.post(
url_for("task_orders.update", screen=1),
@ -130,12 +127,9 @@ def test_create_new_task_order(client, user_session, pdf_upload):
created_task_order_id = response.headers["Location"].split("/")[-1]
created_task_order = TaskOrders.get(created_task_order_id)
assert created_task_order.portfolio is not None
assert created_task_order.portfolio.name == portfolio_name
assert created_task_order.portfolio.defense_component == defense_component
funding_data = slice_data_for_section(task_order_data, "funding")
funding_data = serialize_dates(funding_data)
funding_data["csp_estimate"] = pdf_upload
response = client.post(
response.headers["Location"], data=funding_data, follow_redirects=False
)
@ -156,7 +150,6 @@ def test_create_new_task_order_for_portfolio(client, user_session):
task_order_data = TaskOrderFactory.dictionary()
app_info_data = slice_data_for_section(task_order_data, "app_info")
app_info_data["portfolio_name"] = portfolio.name
app_info_data["defense_component"] = portfolio.defense_component
response = client.post(
url_for("task_orders.update", screen=1, portfolio_id=portfolio.id),
@ -168,10 +161,10 @@ def test_create_new_task_order_for_portfolio(client, user_session):
created_task_order_id = response.headers["Location"].split("/")[-1]
created_task_order = TaskOrders.get(created_task_order_id)
assert created_task_order.portfolio_name == portfolio.name
assert created_task_order.defense_component == portfolio.defense_component
assert created_task_order.portfolio == portfolio
@pytest.mark.skip(reason="Update after implementing new TO form")
def test_task_order_form_shows_errors(client, user_session, task_order):
creator = task_order.creator
user_session(creator)
@ -192,25 +185,7 @@ def test_task_order_form_shows_errors(client, user_session, task_order):
assert "Not a valid decimal" in body
def test_task_order_validates_email_address(client, user_session, task_order):
creator = task_order.creator
user_session(creator)
task_order_data = TaskOrderFactory.dictionary()
oversight_data = slice_data_for_section(task_order_data, "oversight")
oversight_data.update({"ko_email": "not an email"})
response = client.post(
url_for("task_orders.update", screen=3, task_order_id=task_order.id),
data=oversight_data,
follow_redirects=False,
)
body = response.data.decode()
assert "There were some errors" in body
assert "Invalid email" in body
@pytest.mark.skip(reason="Update after implementing new TO form")
def test_review_screen_when_all_sections_complete(client, user_session, task_order):
user_session(task_order.creator)
response = client.get(
@ -222,6 +197,7 @@ def test_review_screen_when_all_sections_complete(client, user_session, task_ord
assert response.status_code == 200
@pytest.mark.skip(reason="Update after implementing new TO form")
def test_review_screen_when_not_all_sections_complete(client, user_session, task_order):
TaskOrders.update(task_order, clin_01=None)
user_session(task_order.creator)
@ -240,9 +216,7 @@ def task_order():
portfolio = PortfolioFactory.create(owner=user)
attachment = Attachment(filename="sample_attachment", object_name="sample")
return TaskOrderFactory.create(
creator=user, portfolio=portfolio, csp_estimate=attachment
)
return TaskOrderFactory.create(creator=user, portfolio=portfolio)
def test_show_task_order(task_order):
@ -254,29 +228,6 @@ def test_show_task_order(task_order):
assert another_workflow.task_order == task_order
def test_show_task_order_form_list_data():
complexity = ["oconus", "tactical_edge"]
user = UserFactory.create()
portfolio = PortfolioFactory.create(owner=user)
task_order = TaskOrderFactory.create(
creator=user, portfolio=portfolio, complexity=complexity
)
workflow = ShowTaskOrderWorkflow(user, task_order_id=task_order.id)
assert workflow.form.complexity.data == complexity
def test_show_task_order_form(task_order):
workflow = ShowTaskOrderWorkflow(task_order.creator)
assert not workflow.form.data["app_migration"]
another_workflow = ShowTaskOrderWorkflow(
task_order.creator, task_order_id=task_order.id
)
assert (
another_workflow.form.data["defense_component"] == task_order.defense_component
)
def test_show_task_order_display_screen(task_order):
workflow = ShowTaskOrderWorkflow(task_order.creator, task_order_id=task_order.id)
screens = workflow.display_screens
@ -287,16 +238,7 @@ def test_show_task_order_display_screen(task_order):
assert screens[3]["completion"] == "incomplete"
def test_update_task_order_with_no_task_order():
user = UserFactory.create()
to_data = TaskOrderFactory.dictionary()
workflow = UpdateTaskOrderWorkflow(user, to_data)
assert workflow.task_order is None
workflow.update()
assert workflow.task_order
assert workflow.task_order.scope == to_data["scope"]
@pytest.mark.skip(reason="Update after implementing new TO form")
def test_update_task_order_with_existing_task_order(task_order):
to_data = serialize_dates(TaskOrderFactory.dictionary())
workflow = UpdateTaskOrderWorkflow(
@ -307,45 +249,7 @@ def test_update_task_order_with_existing_task_order(task_order):
assert workflow.task_order.start_date.strftime("%m/%d/%Y") == to_data["start_date"]
def test_update_to_redirects_to_ko_review(client, user_session, task_order):
ko = UserFactory.create()
task_order.contracting_officer = ko
PortfolioRoleFactory.create(
user=ko,
portfolio=task_order.portfolio,
permission_sets=[PermissionSets.get(PermissionSets.EDIT_PORTFOLIO_FUNDING)],
)
user_session(ko)
url = url_for("task_orders.ko_review", task_order_id=task_order.id)
response = client.post(
url_for("task_orders.new", screen=1, task_order_id=task_order.id, next=url)
)
body = response.data.decode()
assert url in body
assert response.status_code == 302
def test_other_text_not_saved_if_other_not_checked(task_order):
to_data = {
**TaskOrderFactory.dictionary(),
"complexity": ["conus"],
"complexity_other": "quite complex",
}
workflow = UpdateTaskOrderWorkflow(
task_order.creator, to_data, task_order_id=task_order.id
)
workflow.update()
assert not workflow.task_order.complexity_other
def test_cor_data_set_to_user_data_if_am_cor_is_checked(task_order):
to_data = {**task_order.to_dictionary(), "am_cor": True}
workflow = UpdateTaskOrderWorkflow(task_order.creator, to_data, 3, task_order.id)
workflow.update()
assert task_order.cor_dod_id == task_order.creator.dod_id
@pytest.mark.skip(reason="Update after implementing new TO form")
def test_review_task_order_form(client, user_session, task_order):
user_session(task_order.creator)
@ -357,18 +261,7 @@ def test_review_task_order_form(client, user_session, task_order):
assert response.status_code == 200
def test_update_task_order_clears_unnecessary_other_responses():
user = UserFactory.create()
to_data = TaskOrderFactory.dictionary()
to_data["complexity"] = ["storage"]
to_data["complexity_other"] = "something else"
to_data["dev_team"] = ["civilians"]
to_data["dev_team_other"] = "something else"
workflow = UpdateTaskOrderWorkflow(user, to_data)
assert workflow.task_order_form_data["complexity_other"] is None
assert workflow.task_order_form_data["dev_team_other"] is None
@pytest.mark.skip(reason="Reimplement after TO form is updated")
def test_mo_redirected_to_build_page(client, user_session, portfolio):
user_session(portfolio.owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)

303
tests/routes/task_orders/test_officer_reviews.py
View File

@ -1,303 +0,0 @@
import pytest
from flask import url_for
from atst.domain.permission_sets import PermissionSets
from atst.domain.task_orders import TaskOrders
from atst.models.portfolio_role import Status as PortfolioStatus
from tests.factories import (
PortfolioFactory,
PortfolioRoleFactory,
TaskOrderFactory,
UserFactory,
DD254Factory,
)
from tests.utils import captured_templates
@pytest.fixture
def portfolio():
return PortfolioFactory.create()
@pytest.fixture
def user():
return UserFactory.create()
def test_ko_can_view_ko_review_page(client, user_session):
portfolio = PortfolioFactory.create()
ko = UserFactory.create()
cor = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=ko,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
PortfolioRoleFactory.create(
portfolio=portfolio,
user=cor,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(
portfolio=portfolio,
contracting_officer=ko,
contracting_officer_representative=cor,
)
request_url = url_for("task_orders.ko_review", task_order_id=task_order.id)
#
# KO returns 200
#
user_session(ko)
response = client.get(request_url)
assert response.status_code == 200
#
# COR returns 200
#
user_session(cor)
response = client.get(request_url)
assert response.status_code == 200
#
# Random user raises UnauthorizedError
#
user_session(UserFactory.create())
response = client.get(request_url)
assert response.status_code == 404
def test_cor_cant_view_review_until_to_completed(client, user_session):
portfolio = PortfolioFactory.create()
user_session(portfolio.owner)
task_order = TaskOrderFactory.create(
portfolio=portfolio, clin_01=None, cor_dod_id=portfolio.owner.dod_id
)
response = client.get(url_for("task_orders.ko_review", task_order_id=task_order.id))
assert response.status_code == 404
def test_submit_completed_ko_review_page_as_cor(
client, user_session, pdf_upload, portfolio, user
):
PortfolioRoleFactory.create(
portfolio=portfolio,
user=user,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer_representative=user
)
form_data = {
"start_date": "02/10/2019",
"end_date": "03/10/2019",
"number": "1938745981",
"loas-0": "0813458013405",
"custom_clauses": "hi im a custom clause",
"pdf": pdf_upload,
}
user_session(user)
response = client.post(
url_for("task_orders.ko_review", task_order_id=task_order.id), data=form_data
)
assert task_order.pdf
assert response.headers["Location"] == url_for(
"task_orders.view_task_order", task_order_id=task_order.id, _external=True
)
def test_submit_completed_ko_review_page_as_ko(
client, user_session, pdf_upload, portfolio
):
ko = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=ko,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=ko)
dd_254 = DD254Factory.create()
TaskOrders.add_dd_254(task_order, dd_254.to_dictionary())
user_session(ko)
loa_list = ["123123123", "456456456", "789789789"]
form_data = {
"start_date": "02/10/2019",
"end_date": "03/10/2019",
"number": "1938745981",
"loas-0": loa_list[0],
"loas-1": loa_list[1],
"loas-2": loa_list[2],
"custom_clauses": "hi im a custom clause",
"pdf": pdf_upload,
}
response = client.post(
url_for("task_orders.ko_review", task_order_id=task_order.id), data=form_data
)
assert task_order.pdf
assert response.headers["Location"] == url_for(
"task_orders.signature_requested", task_order_id=task_order.id, _external=True
)
assert task_order.loas == loa_list
def test_ko_can_only_access_their_to(app, user_session, client, portfolio, pdf_upload):
ko = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=ko,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=ko)
dd_254 = DD254Factory.create()
TaskOrders.add_dd_254(task_order, dd_254.to_dictionary())
other_task_order = TaskOrderFactory.create()
user_session(ko)
# KO can't see TO
response = client.get(
url_for("task_orders.ko_review", task_order_id=other_task_order.id)
)
assert response.status_code == 404
# KO can't submit review for TO
form_data = {
"start_date": "02/10/2019",
"end_date": "03/10/2019",
"number": "1938745981",
"loas-0": "1231231231",
"custom_clauses": "hi im a custom clause",
"pdf": pdf_upload,
}
response = client.post(
url_for("task_orders.submit_ko_review", task_order_id=other_task_order.id),
data=form_data,
)
assert response.status_code == 404
assert not TaskOrders.is_signed_by_ko(other_task_order)
def test_so_review_page(app, client, user_session, portfolio):
so = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=so,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, security_officer=so)
user_session(portfolio.owner)
owner_response = client.get(
url_for("task_orders.so_review", task_order_id=task_order.id)
)
assert owner_response.status_code == 404
with captured_templates(app) as templates:
user_session(so)
so_response = app.test_client().get(
url_for("task_orders.so_review", task_order_id=task_order.id)
)
_, context = templates[0]
form = context["form"]
co_name = form.certifying_official.data
assert so_response.status_code == 200
assert (
task_order.so_first_name in co_name and task_order.so_last_name in co_name
)
def test_submit_so_review(app, client, user_session, portfolio):
so = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=so,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, security_officer=so)
dd_254_data = DD254Factory.dictionary()
user_session(so)
response = client.post(
url_for("task_orders.submit_so_review", task_order_id=task_order.id),
data=dd_254_data,
)
expected_redirect = url_for(
"task_orders.view_task_order", task_order_id=task_order.id, _external=True
)
assert response.status_code == 302
assert response.headers["Location"] == expected_redirect
assert task_order.dd_254
assert task_order.dd_254.certifying_official == dd_254_data["certifying_official"]
def test_so_can_only_access_their_to(app, client, user_session, portfolio):
so = UserFactory.create()
PortfolioRoleFactory.create(
portfolio=portfolio,
user=so,
status=PortfolioStatus.ACTIVE,
permission_sets=[
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO),
PermissionSets.get(PermissionSets.VIEW_PORTFOLIO_FUNDING),
],
)
task_order = TaskOrderFactory.create(portfolio=portfolio, security_officer=so)
dd_254_data = DD254Factory.dictionary()
other_task_order = TaskOrderFactory.create()
user_session(so)
# SO can't view dd254
response = client.get(
url_for("task_orders.so_review", task_order_id=other_task_order.id)
)
assert response.status_code == 404
# SO can't submit dd254
response = client.post(
url_for("task_orders.submit_so_review", task_order_id=other_task_order.id),
data=dd_254_data,
)
assert response.status_code == 404
assert not other_task_order.dd_254

165
tests/routes/task_orders/test_signing.py
View File

@ -1,167 +1,6 @@
from flask import url_for
from atst.domain.task_orders import TaskOrders
from tests.factories import (
UserFactory,
TaskOrderFactory,
PortfolioFactory,
DD254Factory,
)
from tests.factories import UserFactory, TaskOrderFactory, PortfolioFactory
def create_ko_task_order(user_session, contracting_officer):
portfolio = PortfolioFactory.create(owner=contracting_officer)
user_session(contracting_officer)
task_order = TaskOrderFactory.create(
portfolio=portfolio, contracting_officer=contracting_officer
)
TaskOrders.add_officer(
task_order, "contracting_officer", contracting_officer.to_dictionary()
)
dd_254 = DD254Factory.create()
TaskOrders.add_dd_254(task_order, dd_254.to_dictionary())
return task_order
def test_show_signature_requested_not_ko(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
TaskOrders.update(task_order, contracting_officer=None)
response = client.get(
url_for("task_orders.signature_requested", task_order_id=task_order.id)
)
assert response.status_code == 404
def test_show_signature_requested(client, user_session):
contracting_officer = UserFactory.create()
portfolio = PortfolioFactory.create(owner=contracting_officer)
user_session(contracting_officer)
# create unfinished TO
task_order = TaskOrderFactory.create(portfolio=portfolio, clin_01=None)
TaskOrders.add_officer(
task_order, "contracting_officer", contracting_officer.to_dictionary()
)
response = client.get(
url_for("task_orders.signature_requested", task_order_id=task_order.id)
)
assert response.status_code == 404
# Finish TO
TaskOrders.update(task_order, clin_01=100)
response = client.get(
url_for("task_orders.signature_requested", task_order_id=task_order.id)
)
assert response.status_code == 404
# Complete DD 254
dd_254 = DD254Factory.create()
TaskOrders.add_dd_254(task_order, dd_254.to_dictionary())
response = client.get(
url_for("task_orders.signature_requested", task_order_id=task_order.id)
)
assert response.status_code == 200
def test_show_signature_requested_already_signed(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
TaskOrders.update(task_order, signer_dod_id=contracting_officer.dod_id)
response = client.get(
url_for("task_orders.signature_requested", task_order_id=task_order.id)
)
assert response.status_code == 404
def test_signing_task_order_not_ko(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
TaskOrders.update(task_order, contracting_officer=None)
response = client.post(
url_for("task_orders.record_signature", task_order_id=task_order.id), data={}
)
assert response.status_code == 404
def test_singing_an_already_signed_task_order(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
TaskOrders.update(task_order, signer_dod_id=contracting_officer.dod_id)
response = client.post(
url_for("task_orders.record_signature", task_order_id=task_order.id),
data={"signature": "y", "level_of_warrant": "33.33"},
)
assert response.status_code == 404
def test_signing_a_task_order(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
assert task_order.signed_at is None
assert task_order.signer_dod_id is None
response = client.post(
url_for("task_orders.record_signature", task_order_id=task_order.id),
data={"signature": "y", "level_of_warrant": "33.33"},
)
assert (
url_for("task_orders.view_task_order", task_order_id=task_order.id)
in response.headers["Location"]
)
assert task_order.signer_dod_id == contracting_officer.dod_id
assert task_order.signed_at is not None
def test_signing_a_task_order_failure(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
response = client.post(
url_for("task_orders.record_signature", task_order_id=task_order.id),
data={"level_of_warrant": "33.33"},
)
assert response.status_code == 400
def test_signing_a_task_order_unlimited_level_of_warrant(client, user_session):
contracting_officer = UserFactory.create()
task_order = create_ko_task_order(user_session, contracting_officer)
assert task_order.signed_at is None
assert task_order.signer_dod_id is None
response = client.post(
url_for("task_orders.record_signature", task_order_id=task_order.id),
data={
"signature": "y",
"level_of_warrant": "33.33",
"unlimited_level_of_warrant": "y",
},
)
assert (
url_for("task_orders.view_task_order", task_order_id=task_order.id)
in response.headers["Location"]
)
assert task_order.signed_at is not None
assert task_order.signer_dod_id == contracting_officer.dod_id
assert task_order.unlimited_level_of_warrant == True
assert task_order.level_of_warrant == None
# TODO: add tests!

171
tests/test_access.py
View File

@ -274,44 +274,6 @@ def test_portfolios_edit_access(post_url_assert_status):
post_url_assert_status(rando, url, 404)
# task_orders.invitations_edit
def test_task_orders_invitations_edit_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_FUNDING)
owner = user_with()
rando = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)
url = url_for(
"task_orders.invitations_edit",
portfolio_id=portfolio.id,
task_order_id=task_order.id,
)
post_url_assert_status(ccpo, url, 302)
post_url_assert_status(owner, url, 302)
post_url_assert_status(rando, url, 404)
# task_orders.ko_review
def test_task_orders_ko_review_access(get_url_assert_status):
ccpo = UserFactory.create_ccpo()
owner = user_with()
cor = user_with()
ko = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(
portfolio=portfolio,
contracting_officer=ko,
contracting_officer_representative=cor,
)
url = url_for("task_orders.ko_review", task_order_id=task_order.id)
get_url_assert_status(ccpo, url, 404)
get_url_assert_status(owner, url, 404)
get_url_assert_status(ko, url, 200)
get_url_assert_status(cor, url, 200)
# applications.new
def test_applications_new_access(get_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_APPLICATION_MANAGEMENT)
@ -405,29 +367,6 @@ def test_portfolios_resend_invitation_access(post_url_assert_status):
post_url_assert_status(rando, url, 404)
# task_orders.resend_invite
def test_task_orders_resend_invite_access(post_url_assert_status):
ccpo = UserFactory.create_ccpo()
owner = user_with()
rando = user_with()
ko = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=ko)
prr = PortfolioRoleFactory.create(user=ko, portfolio=portfolio)
PortfolioInvitationFactory.create(user=UserFactory.create(), role=prr)
url = url_for(
"task_orders.resend_invite",
task_order_id=task_order.id,
invite_type="ko_invite",
)
post_url_assert_status(ccpo, url, 302)
post_url_assert_status(owner, url, 302)
post_url_assert_status(ko, url, 404)
post_url_assert_status(rando, url, 404)
# portfolios.revoke_invitation
def test_portfolios_revoke_invitation_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_ADMIN)
@ -450,72 +389,6 @@ def test_portfolios_revoke_invitation_access(post_url_assert_status):
post_url_assert_status(user, url, status)
# task_orders.so_review
def test_task_orders_so_review_access(get_url_assert_status):
ccpo = UserFactory.create_ccpo()
owner = user_with()
rando = user_with()
so = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, security_officer=so)
url = url_for("task_orders.so_review", task_order_id=task_order.id)
get_url_assert_status(so, url, 200)
get_url_assert_status(ccpo, url, 404)
get_url_assert_status(owner, url, 404)
get_url_assert_status(rando, url, 404)
# task_orders.submit_ko_review
def test_task_orders_submit_ko_review_access(post_url_assert_status):
ccpo = UserFactory.create_ccpo()
owner = user_with()
cor = user_with()
ko = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(
portfolio=portfolio,
contracting_officer=ko,
contracting_officer_representative=cor,
)
url = url_for("task_orders.submit_ko_review", task_order_id=task_order.id)
post_url_assert_status(ccpo, url, 404)
post_url_assert_status(owner, url, 404)
post_url_assert_status(ko, url, 200)
post_url_assert_status(cor, url, 200)
# task_orders.submit_so_review
def test_task_orders_submit_so_review_access(post_url_assert_status):
ccpo = UserFactory.create_ccpo()
owner = user_with()
rando = user_with()
so = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, security_officer=so)
url = url_for("task_orders.submit_so_review", task_order_id=task_order.id)
post_url_assert_status(so, url, 200)
post_url_assert_status(ccpo, url, 404)
post_url_assert_status(owner, url, 404)
post_url_assert_status(rando, url, 404)
# task_orders.invitations
def test_task_orders_invitations_access(get_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_FUNDING)
owner = user_with()
rando = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)
url = url_for("task_orders.invitations", task_order_id=task_order.id)
get_url_assert_status(ccpo, url, 200)
get_url_assert_status(owner, url, 200)
get_url_assert_status(rando, url, 404)
# applications.update
def test_applications_update_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_APPLICATION_MANAGEMENT)
@ -568,24 +441,6 @@ def test_task_orders_view_task_order_access(get_url_assert_status):
get_url_assert_status(rando, url, 404)
# task_orders.download_csp_estimate
def test_task_orders_download_csp_estimate_access(get_url_assert_status, monkeypatch):
monkeypatch.setattr(
"atst.routes.task_orders.downloads.send_file", lambda a: Response("")
)
ccpo = user_with(PermissionSets.VIEW_PORTFOLIO_FUNDING)
owner = user_with()
rando = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)
url = url_for("task_orders.download_csp_estimate", task_order_id=task_order.id)
get_url_assert_status(owner, url, 200)
get_url_assert_status(ccpo, url, 200)
get_url_assert_status(rando, url, 404)
# task_orders.download_summary
def test_task_orders_download_summary_access(get_url_assert_status):
ccpo = user_with(PermissionSets.VIEW_PORTFOLIO_FUNDING)
@ -621,21 +476,6 @@ def test_task_orders_download_task_order_pdf_access(get_url_assert_status, monke
get_url_assert_status(rando, url, 404)
# task_orders.invite
def test_task_orders_invite_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_FUNDING)
owner = user_with()
rando = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio)
url = url_for("task_orders.invite", task_order_id=task_order.id)
post_url_assert_status(owner, url, 302)
post_url_assert_status(ccpo, url, 302)
post_url_assert_status(rando, url, 404)
# task_orders.new
def test_task_orders_new_access(get_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_FUNDING)
@ -662,46 +502,45 @@ def test_task_orders_new_access(get_url_assert_status):
# task_orders.record_signature
@pytest.mark.skip(reason="Update after TO signature is reimplemented")
def test_task_orders_record_signature_access(post_url_assert_status, monkeypatch):
ccpo = UserFactory.create_ccpo()
owner = user_with()
rando = user_with()
ko = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=ko)
task_order = TaskOrderFactory.create(portfolio=portfolio)
monkeypatch.setattr(
"atst.routes.task_orders.signing.find_unsigned_ko_to", lambda *a: task_order
)
url = url_for("task_orders.record_signature", task_order_id=task_order.id)
post_url_assert_status(ko, url, 400)
post_url_assert_status(owner, url, 404)
post_url_assert_status(ccpo, url, 404)
post_url_assert_status(rando, url, 404)
# task_orders.signature_requested
@pytest.mark.skip(reason="Update after TO signature is reimplemented")
def test_task_orders_signature_requested_access(get_url_assert_status, monkeypatch):
ccpo = UserFactory.create_ccpo()
owner = user_with()
rando = user_with()
ko = user_with()
portfolio = PortfolioFactory.create(owner=owner)
task_order = TaskOrderFactory.create(portfolio=portfolio, contracting_officer=ko)
task_order = TaskOrderFactory.create(portfolio=portfolio)
monkeypatch.setattr(
"atst.routes.task_orders.signing.find_unsigned_ko_to", lambda *a: task_order
)
url = url_for("task_orders.record_signature", task_order_id=task_order.id)
get_url_assert_status(ko, url, 200)
get_url_assert_status(owner, url, 404)
get_url_assert_status(ccpo, url, 404)
get_url_assert_status(rando, url, 404)
# task_orders.update
@pytest.mark.skip(reason="Update after TO form is fixed")
def test_task_orders_update_access(post_url_assert_status):
ccpo = user_with(PermissionSets.EDIT_PORTFOLIO_FUNDING)
owner = user_with()