Refactor FinVer to make room for draft feature
This commit is contained in:
parent
ec8589fdbd
commit
e72b980d94
57
atst/domain/requests/financial_verification.py
Normal file
57
atst/domain/requests/financial_verification.py
Normal file
@ -0,0 +1,57 @@
|
||||
import re
|
||||
|
||||
from atst.domain.task_orders import TaskOrders
|
||||
from atst.domain.pe_numbers import PENumbers
|
||||
from atst.domain.exceptions import NotFoundError
|
||||
|
||||
|
||||
class PENumberValidator(object):
|
||||
PE_REGEX = re.compile(
|
||||
r"""
|
||||
(0?\d) # program identifier
|
||||
(0?\d) # category
|
||||
(\d) # activity
|
||||
(\d+) # sponsor element
|
||||
(.+) # service
|
||||
""",
|
||||
re.X,
|
||||
)
|
||||
|
||||
def validate(self, request, field):
|
||||
if self._same_as_previous(request, field):
|
||||
return True
|
||||
|
||||
try:
|
||||
PENumbers.get(field.data)
|
||||
except NotFoundError:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def suggest_pe_id(self, pe_id):
|
||||
suggestion = pe_id
|
||||
match = self.PE_REGEX.match(pe_id)
|
||||
if match:
|
||||
(program, category, activity, sponsor, service) = match.groups()
|
||||
if len(program) < 2:
|
||||
program = "0" + program
|
||||
if len(category) < 2:
|
||||
category = "0" + category
|
||||
suggestion = "".join((program, category, activity, sponsor, service))
|
||||
|
||||
if suggestion != pe_id:
|
||||
return suggestion
|
||||
return None
|
||||
|
||||
def _same_as_previous(self, request, field):
|
||||
return request.pe_number == field.data
|
||||
|
||||
|
||||
class TaskOrderNumberValidator(object):
|
||||
def validate(self, field):
|
||||
try:
|
||||
TaskOrders.get(field.data)
|
||||
except NotFoundError:
|
||||
return False
|
||||
|
||||
return True
|
6
atst/forms/exceptions.py
Normal file
6
atst/forms/exceptions.py
Normal file
@ -0,0 +1,6 @@
|
||||
class FormValidationError(Exception):
|
||||
|
||||
message = "Form validation failed."
|
||||
|
||||
def __init__(self, form):
|
||||
self.form = form
|
@ -2,77 +2,20 @@ import re
|
||||
import pendulum
|
||||
from wtforms.fields.html5 import DateField, EmailField
|
||||
from wtforms.fields import StringField, FileField
|
||||
from wtforms.validators import InputRequired, Email, Regexp
|
||||
from wtforms.validators import DataRequired, Email, Regexp
|
||||
from flask_wtf.file import FileAllowed
|
||||
|
||||
from atst.domain.exceptions import NotFoundError
|
||||
from atst.domain.pe_numbers import PENumbers
|
||||
from atst.domain.task_orders import TaskOrders
|
||||
|
||||
from .fields import NewlineListField, SelectField, NumberStringField
|
||||
from .forms import ValidatedForm
|
||||
from .data import FUNDING_TYPES
|
||||
from .validators import DateRange
|
||||
|
||||
|
||||
PE_REGEX = re.compile(
|
||||
r"""
|
||||
(0?\d) # program identifier
|
||||
(0?\d) # category
|
||||
(\d) # activity
|
||||
(\d+) # sponsor element
|
||||
(.+) # service
|
||||
""",
|
||||
re.X,
|
||||
)
|
||||
|
||||
TREASURY_CODE_REGEX = re.compile(r"^0*([1-9]{4}|[1-9]{6})$")
|
||||
|
||||
BA_CODE_REGEX = re.compile(r"[0-9]{2}\w?$")
|
||||
|
||||
|
||||
def suggest_pe_id(pe_id):
|
||||
suggestion = pe_id
|
||||
match = PE_REGEX.match(pe_id)
|
||||
if match:
|
||||
(program, category, activity, sponsor, service) = match.groups()
|
||||
if len(program) < 2:
|
||||
program = "0" + program
|
||||
if len(category) < 2:
|
||||
category = "0" + category
|
||||
suggestion = "".join((program, category, activity, sponsor, service))
|
||||
|
||||
if suggestion != pe_id:
|
||||
return suggestion
|
||||
return None
|
||||
|
||||
|
||||
def validate_pe_id(field, existing_request):
|
||||
try:
|
||||
PENumbers.get(field.data)
|
||||
except NotFoundError:
|
||||
suggestion = suggest_pe_id(field.data)
|
||||
error_str = (
|
||||
"We couldn't find that PE number. {}"
|
||||
"If you have double checked it you can submit anyway. "
|
||||
"Your request will need to go through a manual review."
|
||||
).format('Did you mean "{}"? '.format(suggestion) if suggestion else "")
|
||||
field.errors += (error_str,)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def validate_task_order_number(field):
|
||||
try:
|
||||
TaskOrders.get(field.data)
|
||||
except NotFoundError:
|
||||
field.errors += ("Task Order number not found",)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def number_to_int(num):
|
||||
if num:
|
||||
return int(num)
|
||||
@ -86,12 +29,6 @@ class BaseFinancialForm(ValidatedForm):
|
||||
"""
|
||||
self.uii_ids.process_data(self.uii_ids.data)
|
||||
|
||||
def perform_extra_validation(self, existing_request):
|
||||
valid = True
|
||||
if not existing_request or existing_request.get("pe_id") != self.pe_id.data:
|
||||
valid = validate_pe_id(self.pe_id, existing_request)
|
||||
return valid
|
||||
|
||||
@property
|
||||
def is_missing_task_order_number(self):
|
||||
return False
|
||||
@ -99,7 +36,7 @@ class BaseFinancialForm(ValidatedForm):
|
||||
task_order_number = StringField(
|
||||
"Task Order Number associated with this request",
|
||||
description="Include the original Task Order number (including the 000X at the end). Do not include any modification numbers. Note that there may be a lag between approving a task order and when it becomes available in our system.",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
)
|
||||
|
||||
uii_ids = NewlineListField(
|
||||
@ -110,43 +47,38 @@ class BaseFinancialForm(ValidatedForm):
|
||||
pe_id = StringField(
|
||||
"Program Element Number",
|
||||
description="PE numbers help the Department of Defense identify which offices' budgets are contributing towards this resource use. <br/><em>It should be 7 digits followed by 1-3 letters, and should have a zero as the first and third digits.</em>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
)
|
||||
|
||||
treasury_code = StringField(
|
||||
"Program Treasury Code",
|
||||
description="Program Treasury Code (or Appropriations Code) identifies resource types. <br/> <em>It should be a four digit or six digit number, optionally prefixed by one or more zeros.</em>",
|
||||
validators=[InputRequired(), Regexp(TREASURY_CODE_REGEX)],
|
||||
validators=[DataRequired(), Regexp(TREASURY_CODE_REGEX)],
|
||||
)
|
||||
|
||||
ba_code = StringField(
|
||||
"Program Budget Activity (BA) Code",
|
||||
description="BA Code is used to identify the purposes, projects, or types of activities financed by the appropriation fund. <br/><em>It should be two digits, followed by an optional letter.</em>",
|
||||
validators=[InputRequired(), Regexp(BA_CODE_REGEX)],
|
||||
validators=[DataRequired(), Regexp(BA_CODE_REGEX)],
|
||||
)
|
||||
|
||||
fname_co = StringField("KO First Name", validators=[InputRequired()])
|
||||
lname_co = StringField("KO Last Name", validators=[InputRequired()])
|
||||
fname_co = StringField("KO First Name", validators=[DataRequired()])
|
||||
lname_co = StringField("KO Last Name", validators=[DataRequired()])
|
||||
|
||||
email_co = EmailField("KO Email", validators=[InputRequired(), Email()])
|
||||
email_co = EmailField("KO Email", validators=[DataRequired(), Email()])
|
||||
|
||||
office_co = StringField("KO Office", validators=[InputRequired()])
|
||||
office_co = StringField("KO Office", validators=[DataRequired()])
|
||||
|
||||
fname_cor = StringField("COR First Name", validators=[InputRequired()])
|
||||
fname_cor = StringField("COR First Name", validators=[DataRequired()])
|
||||
|
||||
lname_cor = StringField("COR Last Name", validators=[InputRequired()])
|
||||
lname_cor = StringField("COR Last Name", validators=[DataRequired()])
|
||||
|
||||
email_cor = EmailField("COR Email", validators=[InputRequired(), Email()])
|
||||
email_cor = EmailField("COR Email", validators=[DataRequired(), Email()])
|
||||
|
||||
office_cor = StringField("COR Office", validators=[InputRequired()])
|
||||
office_cor = StringField("COR Office", validators=[DataRequired()])
|
||||
|
||||
|
||||
class FinancialForm(BaseFinancialForm):
|
||||
def perform_extra_validation(self, existing_request):
|
||||
previous_valid = super().perform_extra_validation(existing_request)
|
||||
task_order_valid = validate_task_order_number(self.task_order_number)
|
||||
return previous_valid and task_order_valid
|
||||
|
||||
@property
|
||||
def is_missing_task_order_number(self):
|
||||
return "task_order_number" in self.errors
|
||||
@ -159,13 +91,13 @@ class FinancialForm(BaseFinancialForm):
|
||||
class ExtendedFinancialForm(BaseFinancialForm):
|
||||
def validate(self, *args, **kwargs):
|
||||
if self.funding_type.data == "OTHER":
|
||||
self.funding_type_other.validators.append(InputRequired())
|
||||
self.funding_type_other.validators.append(DataRequired())
|
||||
return super().validate(*args, **kwargs)
|
||||
|
||||
funding_type = SelectField(
|
||||
description="What is the source of funding?",
|
||||
choices=FUNDING_TYPES,
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
render_kw={"required": False},
|
||||
)
|
||||
|
||||
@ -175,7 +107,7 @@ class ExtendedFinancialForm(BaseFinancialForm):
|
||||
"Task Order Expiration Date",
|
||||
description="Please enter the expiration date for the Task Order",
|
||||
validators=[
|
||||
InputRequired(),
|
||||
DataRequired(),
|
||||
DateRange(
|
||||
lower_bound=pendulum.duration(days=0),
|
||||
upper_bound=pendulum.duration(years=100),
|
||||
@ -187,42 +119,42 @@ class ExtendedFinancialForm(BaseFinancialForm):
|
||||
|
||||
clin_0001 = NumberStringField(
|
||||
"<dl><dt>CLIN 0001</dt> - <dd>Unclassified IaaS and PaaS Amount</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
|
||||
clin_0003 = NumberStringField(
|
||||
"<dl><dt>CLIN 0003</dt> - <dd>Unclassified Cloud Support Package</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
|
||||
clin_1001 = NumberStringField(
|
||||
"<dl><dt>CLIN 1001</dt> - <dd>Unclassified IaaS and PaaS Amount <br> OPTION PERIOD 1</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
|
||||
clin_1003 = NumberStringField(
|
||||
"<dl><dt>CLIN 1003</dt> - <dd>Unclassified Cloud Support Package <br> OPTION PERIOD 1</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
|
||||
clin_2001 = NumberStringField(
|
||||
"<dl><dt>CLIN 2001</dt> - <dd>Unclassified IaaS and PaaS Amount <br> OPTION PERIOD 2</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
|
||||
clin_2003 = NumberStringField(
|
||||
"<dl><dt>CLIN 2003</dt> - <dd>Unclassified Cloud Support Package <br> OPTION PERIOD 2</dd></dl>",
|
||||
validators=[InputRequired()],
|
||||
validators=[DataRequired()],
|
||||
description="Review your task order document, the amounts for each CLIN must match exactly here",
|
||||
filters=[number_to_int],
|
||||
)
|
||||
@ -231,6 +163,6 @@ class ExtendedFinancialForm(BaseFinancialForm):
|
||||
"Upload a copy of your Task Order",
|
||||
validators=[
|
||||
FileAllowed(["pdf"], "Only PDF documents can be uploaded."),
|
||||
InputRequired(),
|
||||
DataRequired(),
|
||||
],
|
||||
)
|
||||
|
@ -224,6 +224,10 @@ class Request(Base, mixins.TimestampsMixin, mixins.AuditableMixin):
|
||||
def contracting_officer_email(self):
|
||||
return self.latest_revision.email_co
|
||||
|
||||
@property
|
||||
def pe_number(self):
|
||||
return self.body.get("financial_verification", {}).get("pe_id")
|
||||
|
||||
def __repr__(self):
|
||||
return "<Request(status='{}', name='{}', creator='{}', is_approved='{}', time_created='{}', id='{}')>".format(
|
||||
self.status_displayname,
|
||||
|
@ -4,89 +4,87 @@ from flask import request as http_request
|
||||
from . import requests_bp
|
||||
from atst.domain.requests import Requests
|
||||
from atst.forms.financial import FinancialForm, ExtendedFinancialForm
|
||||
from atst.forms.exceptions import FormValidationError
|
||||
from atst.domain.requests.financial_verification import (
|
||||
PENumberValidator,
|
||||
TaskOrderNumberValidator,
|
||||
)
|
||||
|
||||
|
||||
class FinancialVerification:
|
||||
def __init__(self, request, extended=False, post_data=None):
|
||||
class UpdateFinancialVerification(object):
|
||||
def __init__(
|
||||
self,
|
||||
pe_validator,
|
||||
task_order_validator,
|
||||
user,
|
||||
request,
|
||||
fv_data,
|
||||
is_extended=False,
|
||||
):
|
||||
self.pe_validator = pe_validator
|
||||
self.task_order_validator = task_order_validator
|
||||
self.user = user
|
||||
self.request = request
|
||||
self._extended = extended
|
||||
self._post_data = post_data
|
||||
self._form = None
|
||||
self.reset()
|
||||
self.fv_data = fv_data
|
||||
self.is_extended = is_extended
|
||||
|
||||
def reset(self):
|
||||
self._updateable = False
|
||||
self._valid = False
|
||||
self.workspace = None
|
||||
if self._form:
|
||||
self._form.reset()
|
||||
|
||||
@property
|
||||
def is_extended(self):
|
||||
return self._extended or self.is_pending_changes
|
||||
|
||||
@property
|
||||
def is_pending_changes(self):
|
||||
return self.request.is_pending_financial_verification_changes
|
||||
|
||||
@property
|
||||
def _task_order_data(self):
|
||||
if self.request.task_order:
|
||||
task_order = self.request.task_order
|
||||
data = task_order.to_dictionary()
|
||||
data["task_order_number"] = task_order.number
|
||||
data["funding_type"] = task_order.funding_type.value
|
||||
return data
|
||||
def _get_form(self):
|
||||
if self.is_extended:
|
||||
return ExtendedFinancialForm(data=self.fv_data)
|
||||
else:
|
||||
return {}
|
||||
return FinancialForm(data=self.fv_data)
|
||||
|
||||
@property
|
||||
def _form_data(self):
|
||||
if self._post_data:
|
||||
return self._post_data
|
||||
else:
|
||||
form_data = self.request.body.get("financial_verification", {})
|
||||
form_data.update(self._task_order_data)
|
||||
def execute(self):
|
||||
form = self._get_form()
|
||||
|
||||
return form_data
|
||||
should_update = True
|
||||
should_submit = True
|
||||
updated_request = None
|
||||
submitted = False
|
||||
workspace = None
|
||||
|
||||
@property
|
||||
def form(self):
|
||||
if not self._form:
|
||||
if self.is_extended:
|
||||
self._form = ExtendedFinancialForm(data=self._form_data)
|
||||
else:
|
||||
self._form = FinancialForm(data=self._form_data)
|
||||
if not form.validate():
|
||||
should_update = False
|
||||
|
||||
return self._form
|
||||
if not self.pe_validator.validate(self.request, form.pe_id):
|
||||
suggestion = self.pe_validator.suggest_pe_id(form.pe_id.data)
|
||||
error_str = (
|
||||
"We couldn't find that PE number. {}"
|
||||
"If you have double checked it you can submit anyway. "
|
||||
"Your request will need to go through a manual review."
|
||||
).format('Did you mean "{}"? '.format(suggestion) if suggestion else "")
|
||||
form.pe_id.errors += (error_str,)
|
||||
should_submit = False
|
||||
|
||||
def validate(self):
|
||||
if self.form.validate():
|
||||
self._updateable = True
|
||||
self._valid = self.form.perform_extra_validation(
|
||||
self.request.body.get("financial_verification")
|
||||
if not self.task_order_validator.validate(form.task_order_number):
|
||||
form.task_order_number.errors += ("Task Order number not found",)
|
||||
should_submit = False
|
||||
|
||||
if should_update:
|
||||
updated_request = Requests.update_financial_verification(
|
||||
self.request.id, form.data
|
||||
)
|
||||
else:
|
||||
self._updateable = False
|
||||
self._valid = False
|
||||
form.reset()
|
||||
raise FormValidationError(form)
|
||||
|
||||
return self._valid
|
||||
if should_submit:
|
||||
updated_request = Requests.submit_financial_verification(updated_request)
|
||||
if updated_request.is_financially_verified:
|
||||
workspace = Requests.approve_and_create_workspace(updated_request)
|
||||
submitted = True
|
||||
else:
|
||||
form.reset()
|
||||
raise FormValidationError(form)
|
||||
|
||||
@property
|
||||
def pending(self):
|
||||
return self.request.is_pending_ccpo_approval
|
||||
|
||||
def finalize(self):
|
||||
if self._updateable:
|
||||
self.request = Requests.update_financial_verification(
|
||||
self.request.id, self.form.data
|
||||
)
|
||||
|
||||
if self._valid:
|
||||
self.request = Requests.submit_financial_verification(self.request)
|
||||
|
||||
if self.request.is_financially_verified:
|
||||
self.workspace = Requests.approve_and_create_workspace(self.request)
|
||||
if submitted:
|
||||
return {
|
||||
"state": "submitted",
|
||||
"workspace": workspace,
|
||||
"request": updated_request,
|
||||
}
|
||||
else:
|
||||
return {"state": "pending", "request": updated_request}
|
||||
|
||||
|
||||
@requests_bp.route("/requests/verify/<string:request_id>", methods=["GET"])
|
||||
@ -106,29 +104,32 @@ def financial_verification(request_id):
|
||||
@requests_bp.route("/requests/verify/<string:request_id>", methods=["POST"])
|
||||
def update_financial_verification(request_id):
|
||||
request = Requests.get(g.current_user, request_id)
|
||||
finver = FinancialVerification(
|
||||
request, extended=http_request.args.get("extended"), post_data=http_request.form
|
||||
)
|
||||
fv_data = http_request.form
|
||||
is_extended = http_request.args.get("extended")
|
||||
|
||||
finver.validate()
|
||||
|
||||
finver.finalize()
|
||||
|
||||
if finver.workspace:
|
||||
return redirect(
|
||||
url_for(
|
||||
"workspaces.new_project",
|
||||
workspace_id=finver.workspace.id,
|
||||
newWorkspace=True,
|
||||
)
|
||||
)
|
||||
elif finver.pending:
|
||||
return redirect(url_for("requests.requests_index", modal="pendingCCPOApproval"))
|
||||
else:
|
||||
finver.reset()
|
||||
try:
|
||||
response_context = UpdateFinancialVerification(
|
||||
PENumberValidator(),
|
||||
TaskOrderNumberValidator(),
|
||||
g.current_user,
|
||||
request,
|
||||
fv_data,
|
||||
is_extended=is_extended,
|
||||
).execute()
|
||||
except FormValidationError as e:
|
||||
return render_template(
|
||||
"requests/financial_verification.html",
|
||||
jedi_request=finver.request,
|
||||
f=finver.form,
|
||||
extended=finver.is_extended,
|
||||
jedi_request=request,
|
||||
f=e.form,
|
||||
extended=is_extended,
|
||||
)
|
||||
|
||||
if response_context["state"] == "submitted":
|
||||
workspace = response_context["workspace"]
|
||||
return redirect(
|
||||
url_for(
|
||||
"workspaces.new_project", workspace_id=workspace.id, newWorkspace=True
|
||||
)
|
||||
)
|
||||
elif response_context["state"] == "pending":
|
||||
return redirect(url_for("requests.requests_index", modal="pendingCCPOApproval"))
|
||||
|
@ -1,22 +1,22 @@
|
||||
import pytest
|
||||
from werkzeug.datastructures import ImmutableMultiDict
|
||||
|
||||
from atst.forms.financial import suggest_pe_id, FinancialForm, ExtendedFinancialForm
|
||||
from atst.forms.financial import FinancialForm, ExtendedFinancialForm
|
||||
from atst.eda_client import MockEDAClient
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input_,expected",
|
||||
[
|
||||
("0603502N", None),
|
||||
("0603502NZ", None),
|
||||
("603502N", "0603502N"),
|
||||
("063502N", "0603502N"),
|
||||
("63502N", "0603502N"),
|
||||
],
|
||||
)
|
||||
def test_suggest_pe_id(input_, expected):
|
||||
assert suggest_pe_id(input_) == expected
|
||||
# @pytest.mark.parametrize(
|
||||
# "input_,expected",
|
||||
# [
|
||||
# ("0603502N", None),
|
||||
# ("0603502NZ", None),
|
||||
# ("603502N", "0603502N"),
|
||||
# ("063502N", "0603502N"),
|
||||
# ("63502N", "0603502N"),
|
||||
# ],
|
||||
# )
|
||||
# def test_suggest_pe_id(input_, expected):
|
||||
# assert suggest_pe_id(input_) == expected
|
||||
|
||||
|
||||
def test_funding_type_other_not_required_if_funding_type_is_not_other():
|
||||
@ -82,25 +82,6 @@ def test_ba_code_validation(input_, expected):
|
||||
assert is_valid == expected
|
||||
|
||||
|
||||
def test_task_order_number_validation(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"atst.domain.task_orders.TaskOrders._client", lambda: MockEDAClient()
|
||||
)
|
||||
monkeypatch.setattr("atst.forms.financial.validate_pe_id", lambda *args: True)
|
||||
form_invalid = FinancialForm(data={"task_order_number": "1234"})
|
||||
form_invalid.perform_extra_validation({})
|
||||
|
||||
assert "task_order_number" in form_invalid.errors
|
||||
|
||||
form_valid = FinancialForm(
|
||||
data={"task_order_number": MockEDAClient.MOCK_CONTRACT_NUMBER},
|
||||
eda_client=MockEDAClient(),
|
||||
)
|
||||
form_valid.perform_extra_validation({})
|
||||
|
||||
assert "task_order_number" not in form_valid.errors
|
||||
|
||||
|
||||
def test_can_submit_zero_for_clin():
|
||||
form_first = ExtendedFinancialForm()
|
||||
form_first.validate()
|
||||
|
@ -6,7 +6,7 @@ MOCK_REQUEST = RequestFactory.build(creator=MOCK_USER)
|
||||
DOD_SDN_INFO = {"first_name": "ART", "last_name": "GARFUNKEL", "dod_id": "5892460358"}
|
||||
DOD_SDN = f"CN={DOD_SDN_INFO['last_name']}.{DOD_SDN_INFO['first_name']}.G.{DOD_SDN_INFO['dod_id']},OU=OTHER,OU=PKI,OU=DoD,O=U.S. Government,C=US"
|
||||
|
||||
MOCK_VALID_PE_ID = "8675309U"
|
||||
MOCK_VALID_PE_ID = "080675309U"
|
||||
|
||||
FIXTURE_EMAIL_ADDRESS = "artgarfunkel@uso.mil"
|
||||
|
||||
|
@ -1,12 +1,9 @@
|
||||
import urllib
|
||||
import pytest
|
||||
from flask import url_for
|
||||
|
||||
from atst.eda_client import MockEDAClient
|
||||
from atst.models.request_status_event import RequestStatus
|
||||
from atst.routes.requests.financial_verification import FinancialVerification
|
||||
from atst.routes.requests.financial_verification import UpdateFinancialVerification
|
||||
|
||||
from tests.mocks import MOCK_REQUEST, MOCK_USER
|
||||
from tests.mocks import MOCK_REQUEST, MOCK_USER, MOCK_VALID_PE_ID
|
||||
from tests.factories import (
|
||||
PENumberFactory,
|
||||
RequestFactory,
|
||||
@ -14,206 +11,120 @@ from tests.factories import (
|
||||
RequestStatusEventFactory,
|
||||
RequestReviewFactory,
|
||||
)
|
||||
from atst.forms.exceptions import FormValidationError
|
||||
from atst.domain.requests.financial_verification import (
|
||||
PENumberValidator,
|
||||
TaskOrderNumberValidator,
|
||||
)
|
||||
|
||||
required_data = {
|
||||
"pe_id": "123",
|
||||
"task_order_number": MockEDAClient.MOCK_CONTRACT_NUMBER,
|
||||
"fname_co": "Contracting",
|
||||
"lname_co": "Officer",
|
||||
"email_co": "jane@mail.mil",
|
||||
"office_co": "WHS",
|
||||
"fname_cor": "Officer",
|
||||
"lname_cor": "Representative",
|
||||
"email_cor": "jane@mail.mil",
|
||||
"office_cor": "WHS",
|
||||
"uii_ids": "1234",
|
||||
"treasury_code": "00123456",
|
||||
"ba_code": "02A",
|
||||
}
|
||||
|
||||
|
||||
class TestPENumberInForm:
|
||||
|
||||
required_data = {
|
||||
"pe_id": "123",
|
||||
"task_order_number": MockEDAClient.MOCK_CONTRACT_NUMBER,
|
||||
"fname_co": "Contracting",
|
||||
"lname_co": "Officer",
|
||||
"email_co": "jane@mail.mil",
|
||||
"office_co": "WHS",
|
||||
"fname_cor": "Officer",
|
||||
"lname_cor": "Representative",
|
||||
"email_cor": "jane@mail.mil",
|
||||
"office_cor": "WHS",
|
||||
"uii_ids": "1234",
|
||||
"treasury_code": "00123456",
|
||||
"ba_code": "02A",
|
||||
}
|
||||
|
||||
def _set_monkeypatches(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"atst.forms.financial.FinancialForm.validate", lambda s: True
|
||||
)
|
||||
user = UserFactory.create()
|
||||
monkeypatch.setattr("atst.domain.auth.get_current_user", lambda *args: user)
|
||||
return user
|
||||
|
||||
def submit_data(self, client, user, data, extended=False):
|
||||
request = RequestFactory.create(creator=user)
|
||||
url_kwargs = {"request_id": request.id}
|
||||
if extended:
|
||||
url_kwargs["extended"] = True
|
||||
response = client.post(
|
||||
url_for("requests.financial_verification", **url_kwargs),
|
||||
data=data,
|
||||
follow_redirects=False,
|
||||
)
|
||||
return response
|
||||
|
||||
def test_submit_request_form_with_invalid_pe_id(self, monkeypatch, client):
|
||||
user = self._set_monkeypatches(monkeypatch)
|
||||
|
||||
response = self.submit_data(client, user, self.required_data)
|
||||
|
||||
assert "We couldn't find that PE number" in response.data.decode()
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_submit_request_form_with_unchanged_pe_id(self, monkeypatch, client):
|
||||
user = self._set_monkeypatches(monkeypatch)
|
||||
|
||||
data = dict(self.required_data)
|
||||
data["pe_id"] = "0101110F"
|
||||
|
||||
response = self.submit_data(client, user, data)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "/workspaces" in response.headers.get("Location")
|
||||
|
||||
def test_submit_request_form_with_new_valid_pe_id(self, monkeypatch, client):
|
||||
user = self._set_monkeypatches(monkeypatch)
|
||||
pe = PENumberFactory.create(number="8675309U", description="sample PE number")
|
||||
|
||||
data = dict(self.required_data)
|
||||
data["pe_id"] = pe.number
|
||||
|
||||
response = self.submit_data(client, user, data)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "/workspaces" in response.headers.get("Location")
|
||||
|
||||
def test_submit_request_form_with_missing_pe_id(self, monkeypatch, client):
|
||||
user = self._set_monkeypatches(monkeypatch)
|
||||
|
||||
data = dict(self.required_data)
|
||||
data["pe_id"] = ""
|
||||
|
||||
response = self.submit_data(client, user, data)
|
||||
|
||||
assert "There were some errors" in response.data.decode()
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_submit_financial_form_with_invalid_task_order(
|
||||
self, monkeypatch, user_session, client
|
||||
):
|
||||
user = UserFactory.create()
|
||||
user_session(user)
|
||||
|
||||
data = dict(self.required_data)
|
||||
data["pe_id"] = "0101110F"
|
||||
data["task_order_number"] = "1234"
|
||||
|
||||
response = self.submit_data(client, user, data)
|
||||
|
||||
assert "extended=True" in response.data.decode()
|
||||
|
||||
def test_submit_financial_form_with_valid_task_order(
|
||||
self, monkeypatch, user_session, client
|
||||
):
|
||||
user = UserFactory.create()
|
||||
monkeypatch.setattr(
|
||||
"atst.domain.requests.Requests.get", lambda *args: MOCK_REQUEST
|
||||
)
|
||||
user_session(user)
|
||||
|
||||
data = dict(self.required_data)
|
||||
data["pe_id"] = "0101110F"
|
||||
data["task_order_number"] = MockEDAClient.MOCK_CONTRACT_NUMBER
|
||||
|
||||
response = self.submit_data(client, user, data)
|
||||
|
||||
assert "enter TO information manually" not in response.data.decode()
|
||||
|
||||
def test_submit_extended_financial_form(
|
||||
self, monkeypatch, user_session, client, extended_financial_verification_data
|
||||
):
|
||||
user = UserFactory.create()
|
||||
request = RequestFactory.create(creator=user)
|
||||
monkeypatch.setattr("atst.domain.requests.Requests.get", lambda *args: request)
|
||||
monkeypatch.setattr("atst.forms.financial.validate_pe_id", lambda *args: True)
|
||||
user_session(user)
|
||||
data = {**self.required_data, **extended_financial_verification_data}
|
||||
data["task_order_number"] = "1234567"
|
||||
|
||||
response = self.submit_data(client, user, data, extended=True)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert "/requests" in response.headers.get("Location")
|
||||
|
||||
def test_submit_invalid_extended_financial_form(
|
||||
self, monkeypatch, user_session, client, extended_financial_verification_data
|
||||
):
|
||||
monkeypatch.setattr("atst.forms.financial.validate_pe_id", lambda *args: True)
|
||||
user = UserFactory.create()
|
||||
user_session(user)
|
||||
data = {**self.required_data, **extended_financial_verification_data}
|
||||
data["task_order_number"] = "1234567"
|
||||
del (data["clin_0001"])
|
||||
|
||||
response = self.submit_data(client, user, data, extended=True)
|
||||
|
||||
assert response.status_code == 200
|
||||
class MockPEValidator(object):
|
||||
def validate(self, request, field):
|
||||
return True
|
||||
|
||||
|
||||
def test_displays_ccpo_review_comment(user_session, client):
|
||||
creator = UserFactory.create()
|
||||
ccpo = UserFactory.from_atat_role("ccpo")
|
||||
user_session(creator)
|
||||
request = RequestFactory.create(creator=creator)
|
||||
status = RequestStatusEventFactory.create(
|
||||
revision=request.latest_revision,
|
||||
new_status=RequestStatus.CHANGES_REQUESTED_TO_FINVER,
|
||||
request=request,
|
||||
class MockTaskOrderValidator(object):
|
||||
def validate(self, field):
|
||||
return True
|
||||
|
||||
|
||||
def test_update():
|
||||
request = RequestFactory.create()
|
||||
user = UserFactory.create()
|
||||
data = {**required_data, "pe_id": MOCK_VALID_PE_ID}
|
||||
|
||||
response_context = UpdateFinancialVerification(
|
||||
MockPEValidator(),
|
||||
MockTaskOrderValidator(),
|
||||
user,
|
||||
request,
|
||||
data,
|
||||
is_extended=False,
|
||||
).execute()
|
||||
|
||||
assert response_context.get("workspace")
|
||||
|
||||
|
||||
def test_re_enter_pe_number():
|
||||
request = RequestFactory.create()
|
||||
user = UserFactory.create()
|
||||
data = {**required_data, "pe_id": "0101228M"}
|
||||
update_fv = UpdateFinancialVerification(
|
||||
PENumberValidator(),
|
||||
MockTaskOrderValidator(),
|
||||
user,
|
||||
request,
|
||||
data,
|
||||
is_extended=False,
|
||||
)
|
||||
review_comment = "add all of the correct info, instead of the incorrect info"
|
||||
RequestReviewFactory.create(reviewer=ccpo, comment=review_comment, status=status)
|
||||
response = client.get("/requests/verify/{}".format(request.id))
|
||||
body = response.data.decode()
|
||||
assert review_comment in body
|
||||
|
||||
with pytest.raises(FormValidationError):
|
||||
update_fv.execute()
|
||||
response_context = update_fv.execute()
|
||||
|
||||
assert response_context.get("status", "submitted")
|
||||
|
||||
|
||||
class TestFinancialVerification:
|
||||
def _service_object(self, request=None, extended=False, post_data={}):
|
||||
if not request:
|
||||
self.request = RequestFactory.create()
|
||||
else:
|
||||
self.request = request
|
||||
def test_invalid_task_order_number():
|
||||
request = RequestFactory.create()
|
||||
user = UserFactory.create()
|
||||
data = {**required_data, "task_order_number": "DCA10096D0051"}
|
||||
update_fv = UpdateFinancialVerification(
|
||||
MockPEValidator(),
|
||||
TaskOrderNumberValidator(),
|
||||
user,
|
||||
request,
|
||||
data,
|
||||
is_extended=False,
|
||||
)
|
||||
|
||||
return FinancialVerification(
|
||||
self.request, extended=extended, post_data=post_data
|
||||
)
|
||||
with pytest.raises(FormValidationError):
|
||||
update_fv.execute()
|
||||
|
||||
def test_is_extended(self):
|
||||
finver_one = self._service_object()
|
||||
assert not finver_one.is_extended
|
||||
finver_two = self._service_object(
|
||||
request=RequestFactory.create_with_status(
|
||||
RequestStatus.CHANGES_REQUESTED_TO_FINVER
|
||||
)
|
||||
)
|
||||
assert finver_two.is_extended
|
||||
finver_three = self._service_object(extended=True)
|
||||
assert finver_three.is_extended
|
||||
|
||||
def test_is_pending_changes(self):
|
||||
finver_one = self._service_object()
|
||||
assert not finver_one.is_pending_changes
|
||||
finver_two = self._service_object(
|
||||
request=RequestFactory.create_with_status(
|
||||
RequestStatus.CHANGES_REQUESTED_TO_FINVER
|
||||
)
|
||||
)
|
||||
assert finver_two.is_pending_changes
|
||||
def test_extended_fv_data(extended_financial_verification_data):
|
||||
request = RequestFactory.create()
|
||||
user = UserFactory.create()
|
||||
data = {**required_data, **extended_financial_verification_data}
|
||||
update_fv = UpdateFinancialVerification(
|
||||
MockPEValidator(),
|
||||
TaskOrderNumberValidator(),
|
||||
user,
|
||||
request,
|
||||
data,
|
||||
is_extended=True,
|
||||
)
|
||||
|
||||
def test_pending(self):
|
||||
finver_one = self._service_object()
|
||||
assert not finver_one.pending
|
||||
finver_two = self._service_object(
|
||||
request=RequestFactory.create_with_status(
|
||||
RequestStatus.PENDING_CCPO_APPROVAL
|
||||
)
|
||||
)
|
||||
assert finver_two.pending
|
||||
assert update_fv.execute()
|
||||
|
||||
|
||||
def test_missing_extended_fv_data():
|
||||
request = RequestFactory.create()
|
||||
user = UserFactory.create()
|
||||
update_fv = UpdateFinancialVerification(
|
||||
MockPEValidator(),
|
||||
TaskOrderNumberValidator(),
|
||||
user,
|
||||
request,
|
||||
required_data,
|
||||
is_extended=True,
|
||||
)
|
||||
|
||||
with pytest.raises(FormValidationError):
|
||||
update_fv.execute()
|
||||
|
Loading…
x
Reference in New Issue
Block a user