diff --git a/atst/routes/requests/requests_form.py b/atst/routes/requests/requests_form.py index 63a14224..391d49ed 100644 --- a/atst/routes/requests/requests_form.py +++ b/atst/routes/requests/requests_form.py @@ -3,6 +3,7 @@ from flask import g, redirect, render_template, url_for, request as http_request from . import requests_bp from atst.domain.requests import Requests from atst.routes.requests.jedi_request_flow import JEDIRequestFlow +from atst.models.permissions import Permissions @requests_bp.route("/requests/new/", methods=["GET"]) @@ -25,6 +26,9 @@ def requests_form_new(screen): ) @requests_bp.route("/requests/new//", methods=["GET"]) def requests_form_update(screen=1, request_id=None): + if request_id and not _can_view_request(request_id): + return redirect(url_for("atst.unauthorized")) + request = Requests.get(request_id) if request_id is not None else None jedi_flow = JEDIRequestFlow(screen, request, request_id=request_id) @@ -79,10 +83,12 @@ def requests_update(screen=1, request_id=None): request_id=jedi_flow.request_id, ) return redirect(where) + else: return render_template( "requests/screen-%d.html" % int(screen), **rerender_args ) + else: return render_template("requests/screen-%d.html" % int(screen), **rerender_args) @@ -94,5 +100,15 @@ def requests_submit(request_id=None): if request.status == "approved": return redirect("/requests?modal=True") + else: return redirect("/requests") + + +# TODO: generalize this, along with other authorizations, into a policy-pattern +# for authorization in the application +def _can_view_request(request_id): + return ( + Permissions.REVIEW_AND_APPROVE_JEDI_WORKSPACE_REQUEST in g.current_user.atat_permissions + or Requests.is_creator(request_id, g.current_user.id) + ) diff --git a/tests/factories.py b/tests/factories.py index d731c42f..12c41fd4 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -9,6 +9,8 @@ from atst.models.pe_number import PENumber from atst.models.task_order import TaskOrder from atst.models.user import User from atst.models.role import Role +from atst.models.request_status_event import RequestStatusEvent +from atst.domain.roles import Roles class RequestStatusFactory(factory.alchemy.SQLAlchemyModelFactory): @@ -24,6 +26,7 @@ class RequestFactory(factory.alchemy.SQLAlchemyModelFactory): status_events = factory.RelatedFactory( RequestStatusFactory, "request", new_status=RequestStatus.STARTED ) + body = {} class PENumberFactory(factory.alchemy.SQLAlchemyModelFactory): @@ -53,3 +56,11 @@ class UserFactory(factory.alchemy.SQLAlchemyModelFactory): last_name = factory.Faker("last_name") atat_role = factory.SubFactory(RoleFactory) dod_id = factory.LazyFunction(lambda: "".join(random.choices(string.digits, k=10))) + + +class RequestStatusEventFactory(factory.alchemy.SQLAlchemyModelFactory): + + class Meta: + model = RequestStatusEvent + + id = factory.Sequence(lambda x: uuid4()) diff --git a/tests/routes/test_request_new.py b/tests/routes/test_request_new.py index 0927f1be..a7030e6f 100644 --- a/tests/routes/test_request_new.py +++ b/tests/routes/test_request_new.py @@ -2,7 +2,8 @@ import re import pytest import urllib from tests.mocks import MOCK_USER, MOCK_REQUEST -from tests.factories import RequestFactory +from tests.factories import RequestFactory, UserFactory, RequestStatusEventFactory +from atst.domain.roles import Roles ERROR_CLASS = "alert--error" @@ -27,3 +28,37 @@ def test_submit_valid_request_form(monkeypatch, client, user_session): data="meaning=42", ) assert "/requests/new/2" in response.headers.get("Location") + + +def test_owner_can_view_request(client, user_session): + user = UserFactory.create() + user_session(user) + request = RequestFactory.create(creator=user.id) + status = RequestStatusEventFactory.create(request_id=request.id) + + response = client.get("/requests/new/1/{}".format(request.id), follow_redirects=True) + + assert response.status_code == 200 + + +def test_non_owner_cannot_view_request(client, user_session): + user = UserFactory.create() + user_session(user) + request = RequestFactory.create() + status = RequestStatusEventFactory.create(request_id=request.id) + + response = client.get("/requests/new/1/{}".format(request.id), follow_redirects=True) + + assert response.status_code == 401 + + +def test_ccpo_can_view_request(client, user_session): + ccpo = Roles.get("ccpo") + user = UserFactory.create(atat_role=ccpo) + user_session(user) + request = RequestFactory.create() + status = RequestStatusEventFactory.create(request_id=request.id) + + response = client.get("/requests/new/1/{}".format(request.id), follow_redirects=True) + + assert response.status_code == 200