Refactor requests index route for easier testing
This commit is contained in:
richard-dds
parent
58593fc7d6
commit
319f6c05c2
@ -6,84 +6,96 @@ from atst.domain.requests import Requests
|
||||
from atst.models.permissions import Permissions
|
||||
|
||||
|
||||
def map_request(request, viewing_role):
|
||||
class RequestsIndex(object):
|
||||
def __init__(self, user):
|
||||
self.user = user
|
||||
|
||||
time_created = pendulum.instance(request.time_created)
|
||||
is_new = time_created.add(days=1) > pendulum.now()
|
||||
app_count = request.body.get("details_of_use", {}).get("num_software_systems", 0)
|
||||
annual_usage = request.annual_spend
|
||||
last_submission_timestamp = (
|
||||
request.last_submission_timestamp.format("M/DD/YYYY")
|
||||
if request.last_submission_timestamp
|
||||
else "-"
|
||||
)
|
||||
def execute(self):
|
||||
if (
|
||||
Permissions.REVIEW_AND_APPROVE_JEDI_WORKSPACE_REQUEST
|
||||
in self.user.atat_permissions
|
||||
):
|
||||
return self._ccpo_view(self.user)
|
||||
|
||||
if Requests.is_pending_financial_verification(request):
|
||||
edit_link = url_for("requests.financial_verification", request_id=request.id)
|
||||
elif Requests.is_pending_ccpo_approval(request):
|
||||
edit_link = url_for("requests.view_pending_request", request_id=request.id)
|
||||
else:
|
||||
edit_link = url_for(
|
||||
"requests.requests_form_update", screen=1, request_id=request.id
|
||||
else:
|
||||
return self._non_ccpo_view(self.user)
|
||||
|
||||
def _ccpo_view(self, user):
|
||||
requests = Requests.get_many()
|
||||
mapped_requests = [self.map_request(r, "ccpo") for r in requests]
|
||||
num_action_required = len(
|
||||
[r for r in mapped_requests if r.get("action_required")]
|
||||
)
|
||||
|
||||
return {
|
||||
"workspace_id": request.workspace.id if request.workspace else None,
|
||||
"order_id": request.id,
|
||||
"is_new": is_new,
|
||||
"status": request.status_displayname,
|
||||
"app_count": app_count,
|
||||
"last_submission_timestamp": last_submission_timestamp,
|
||||
"full_name": request.creator.full_name,
|
||||
"annual_usage": annual_usage,
|
||||
"edit_link": edit_link,
|
||||
"action_required": request.action_required_by == viewing_role,
|
||||
}
|
||||
return dict(
|
||||
requests=mapped_requests,
|
||||
pending_financial_verification=False,
|
||||
pending_ccpo_approval=False,
|
||||
extended_view=True,
|
||||
kpi_inprogress=Requests.in_progress_count(),
|
||||
kpi_pending=Requests.pending_ccpo_count(),
|
||||
kpi_completed=Requests.completed_count(),
|
||||
num_action_required=num_action_required,
|
||||
)
|
||||
|
||||
def _non_ccpo_view(self, user):
|
||||
requests = Requests.get_many(creator=user)
|
||||
mapped_requests = [self.map_request(r, "mission_owner") for r in requests]
|
||||
num_action_required = len(
|
||||
[r for r in mapped_requests if r.get("action_required")]
|
||||
)
|
||||
pending_fv = any(
|
||||
Requests.is_pending_financial_verification(r) for r in requests
|
||||
)
|
||||
pending_ccpo = any(Requests.is_pending_ccpo_approval(r) for r in requests)
|
||||
|
||||
return dict(
|
||||
requests=mapped_requests,
|
||||
pending_financial_verification=pending_fv,
|
||||
pending_ccpo_approval=pending_ccpo,
|
||||
num_action_required=num_action_required,
|
||||
extended_view=False,
|
||||
)
|
||||
|
||||
def map_request(self, request, viewing_role):
|
||||
time_created = pendulum.instance(request.time_created)
|
||||
is_new = time_created.add(days=1) > pendulum.now()
|
||||
app_count = request.body.get("details_of_use", {}).get(
|
||||
"num_software_systems", 0
|
||||
)
|
||||
annual_usage = request.annual_spend
|
||||
last_submission_timestamp = (
|
||||
request.last_submission_timestamp.format("M/DD/YYYY")
|
||||
if request.last_submission_timestamp
|
||||
else "-"
|
||||
)
|
||||
|
||||
if Requests.is_pending_financial_verification(request):
|
||||
edit_link = url_for(
|
||||
"requests.financial_verification", request_id=request.id
|
||||
)
|
||||
elif Requests.is_pending_ccpo_approval(request):
|
||||
edit_link = url_for("requests.view_pending_request", request_id=request.id)
|
||||
else:
|
||||
edit_link = url_for(
|
||||
"requests.requests_form_update", screen=1, request_id=request.id
|
||||
)
|
||||
|
||||
return {
|
||||
"workspace_id": request.workspace.id if request.workspace else None,
|
||||
"order_id": request.id,
|
||||
"is_new": is_new,
|
||||
"status": request.status_displayname,
|
||||
"app_count": app_count,
|
||||
"last_submission_timestamp": last_submission_timestamp,
|
||||
"full_name": request.creator.full_name,
|
||||
"annual_usage": annual_usage,
|
||||
"edit_link": edit_link,
|
||||
"action_required": request.action_required_by == viewing_role,
|
||||
}
|
||||
|
||||
|
||||
@requests_bp.route("/requests", methods=["GET"])
|
||||
def requests_index():
|
||||
if (
|
||||
Permissions.REVIEW_AND_APPROVE_JEDI_WORKSPACE_REQUEST
|
||||
in g.current_user.atat_permissions
|
||||
):
|
||||
return _ccpo_view()
|
||||
|
||||
else:
|
||||
return _non_ccpo_view()
|
||||
|
||||
|
||||
def _ccpo_view():
|
||||
requests = Requests.get_many()
|
||||
mapped_requests = [map_request(r, "ccpo") for r in requests]
|
||||
num_action_required = len([r for r in mapped_requests if r.get("action_required")])
|
||||
|
||||
return render_template(
|
||||
"requests.html",
|
||||
requests=mapped_requests,
|
||||
pending_financial_verification=False,
|
||||
pending_ccpo_approval=False,
|
||||
extended_view=True,
|
||||
kpi_inprogress=Requests.in_progress_count(),
|
||||
kpi_pending=Requests.pending_ccpo_count(),
|
||||
kpi_completed=Requests.completed_count(),
|
||||
num_action_required=num_action_required,
|
||||
)
|
||||
|
||||
|
||||
def _non_ccpo_view():
|
||||
requests = Requests.get_many(creator=g.current_user)
|
||||
mapped_requests = [map_request(r, "mission_owner") for r in requests]
|
||||
num_action_required = len([r for r in mapped_requests if r.get("action_required")])
|
||||
|
||||
pending_fv = any(Requests.is_pending_financial_verification(r) for r in requests)
|
||||
pending_ccpo = any(Requests.is_pending_ccpo_approval(r) for r in requests)
|
||||
|
||||
return render_template(
|
||||
"requests.html",
|
||||
requests=mapped_requests,
|
||||
pending_financial_verification=pending_fv,
|
||||
pending_ccpo_approval=pending_ccpo,
|
||||
num_action_required=num_action_required,
|
||||
extended_view=False,
|
||||
)
|
||||
context = RequestsIndex(g.current_user).execute()
|
||||
return render_template("requests.html", **context)
|
||||
|
||||
@ -11,7 +11,6 @@ from atst.models.task_order import TaskOrder
|
||||
from atst.models.user import User
|
||||
from atst.models.role import Role
|
||||
from atst.models.workspace import Workspace
|
||||
from atst.models.request_status_event import RequestStatusEvent
|
||||
from atst.domain.roles import Roles
|
||||
|
||||
|
||||
@ -33,6 +32,10 @@ class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
|
||||
atat_role = factory.SubFactory(RoleFactory)
|
||||
dod_id = factory.LazyFunction(lambda: "".join(random.choices(string.digits, k=10)))
|
||||
|
||||
@classmethod
|
||||
def from_atat_role(cls, atat_role_name, **kwargs):
|
||||
role = Roles.get(atat_role_name)
|
||||
return cls.create(atat_role=role, **kwargs)
|
||||
|
||||
class RequestStatusEventFactory(factory.alchemy.SQLAlchemyModelFactory):
|
||||
class Meta:
|
||||
|
||||
23
tests/routes/test_requests_index.py
Normal file
23
tests/routes/test_requests_index.py
Normal file
@ -0,0 +1,23 @@
|
||||
from atst.routes.requests.index import RequestsIndex
|
||||
from tests.factories import RequestFactory, UserFactory
|
||||
from atst.domain.requests import Requests
|
||||
|
||||
|
||||
def test_action_required_mission_owner():
|
||||
creator = UserFactory.create()
|
||||
requests = RequestFactory.create_batch(5, creator=creator)
|
||||
Requests.submit(requests[0])
|
||||
context = RequestsIndex(creator).execute()
|
||||
|
||||
assert context['requests'][0]['action_required'] == False
|
||||
|
||||
|
||||
def test_action_required_ccpo():
|
||||
creator = UserFactory.create()
|
||||
requests = RequestFactory.create_batch(5, creator=creator)
|
||||
Requests.submit(requests[0])
|
||||
|
||||
ccpo = UserFactory.from_atat_role("ccpo")
|
||||
context = RequestsIndex(ccpo).execute()
|
||||
|
||||
assert context['num_action_required'] == 1
|
||||
Loading…
x
Reference in New Issue
Block a user