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

Merge pull request #766 from dod-ccpo/resource-scope-#164896879

Browse Source 
Resource scope #164896879
This commit is contained in:
dandds 2019-04-17 11:02:41 -04:00 committed by GitHub
commit d722f8f375
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 159 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
atst/domain/__init__.py
View File

@ -0,0 +1,25 @@
from sqlalchemy.orm.exc import NoResultFound
from atst.database import db
from atst.domain.exceptions import NotFoundError
class BaseDomainClass(object):
model = None
resource_name = None
@classmethod
def get(cls, resource_id, **kwargs):
base_query = db.session.query(cls.model).filter(cls.model.id == resource_id)
if getattr(cls.model, "deleted", False):
base_query = base_query.filter(cls.model.deleted == False)
for col, val in kwargs.items():
base_query = base_query.filter(getattr(cls.model, col) == val)
try:
resource = base_query.one()
return resource
except NoResultFound:
raise NotFoundError(cls.resource_name)

19
atst/domain/applications.py
View File

@ -1,6 +1,7 @@
from sqlalchemy.orm.exc import NoResultFound
from atst.database import db
from . import BaseDomainClass
from atst.domain.environments import Environments
from atst.domain.exceptions import NotFoundError
from atst.models.application import Application
@ -8,7 +9,10 @@ from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole
class Applications(object):
class Applications(BaseDomainClass):
model = Application
resource_name = "application"
@classmethod
def create(cls, portfolio, name, description, environment_names):
application = Application(
@ -21,19 +25,6 @@ class Applications(object):
db.session.commit()
return application
@classmethod
def get(cls, application_id):
try:
application = (
db.session.query(Application)
.filter_by(id=application_id, deleted=False)
.one()
)
except NoResultFound:
raise NotFoundError("application")
return application
@classmethod
def for_user(self, user, portfolio):
return (

21
atst/domain/environment_roles.py
View File

@ -1,7 +1,9 @@
from flask import current_app as app
from sqlalchemy.orm.exc import NoResultFound
from atst.models.environment_role import EnvironmentRole
from atst.database import db
from atst.domain.exceptions import NotFoundError
from atst.models import EnvironmentRole, Environment, Application
class EnvironmentRoles(object):
@ -13,6 +15,23 @@ class EnvironmentRoles(object):
app.csp.cloud.create_role(env_role)
return env_role
@classmethod
def get_for_portfolio(cls, user_id, environment_id, portfolio_id):
try:
return (
db.session.query(EnvironmentRole)
.join(Environment, EnvironmentRole.environment_id == Environment.id)
.join(Application, Environment.application_id == Application.id)
.filter(
EnvironmentRole.user_id == user_id,
EnvironmentRole.environment_id == environment_id,
Application.portfolio_id == portfolio_id,
)
.one()
)
except NoResultFound:
raise NotFoundError("environment_role")
@classmethod
def get(cls, user_id, environment_id):
existing_env_role = (

17
atst/domain/task_orders.py
View File

@ -1,19 +1,21 @@
from sqlalchemy.orm.exc import NoResultFound
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
from .exceptions import NotFoundError
class TaskOrderError(Exception):
pass
class TaskOrders(object):
class TaskOrders(BaseDomainClass):
model = TaskOrder
resource_name = "task_order"
SECTIONS = {
"app_info": [
"portfolio_name",
@ -51,15 +53,6 @@ class TaskOrders(object):
UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"]
@classmethod
def get(cls, task_order_id):
try:
task_order = db.session.query(TaskOrder).filter_by(id=task_order_id).one()
return task_order
except NoResultFound:
raise NotFoundError("task_order")
@classmethod
def create(cls, creator, portfolio):
task_order = TaskOrder(portfolio=portfolio, creator=creator)

34
atst/routes/portfolios/applications.py
View File

@ -21,18 +21,14 @@ from atst.utils.flash import formatted_flash as flash
@portfolios_bp.route("/portfolios/<portfolio_id>/applications")
@user_can(Permissions.VIEW_APPLICATION, message="view portfolio applications")
def portfolio_applications(portfolio_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
return render_template("portfolios/applications/index.html", portfolio=portfolio)
return render_template("portfolios/applications/index.html")
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new")
@user_can(Permissions.CREATE_APPLICATION, message="view create new application form")
def new_application(portfolio_id):
portfolio = Portfolios.get_for_update(portfolio_id)
form = NewApplicationForm()
return render_template(
"portfolios/applications/new.html", portfolio=portfolio, form=form
)
return render_template("portfolios/applications/new.html", form=form)
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new", methods=["POST"])
@ -50,12 +46,10 @@ def create_application(portfolio_id):
application_data["environment_names"],
)
return redirect(
url_for("portfolios.portfolio_applications", portfolio_id=portfolio.id)
url_for("portfolios.portfolio_applications", portfolio_id=portfolio_id)
)
else:
return render_template(
"portfolios/applications/new.html", portfolio=portfolio, form=form
)
return render_template("portfolios/applications/new.html", form=form)
def get_environments_obj_for_app(application):
@ -70,13 +64,11 @@ def get_environments_obj_for_app(application):
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/<application_id>/edit")
@user_can(Permissions.EDIT_APPLICATION, message="view application edit form")
def edit_application(portfolio_id, application_id):
portfolio = Portfolios.get_for_update(portfolio_id)
application = Applications.get(application_id)
application = Applications.get(application_id, portfolio_id=portfolio_id)
form = ApplicationForm(name=application.name, description=application.description)
return render_template(
"portfolios/applications/edit.html",
portfolio=portfolio,
application=application,
form=form,
environments_obj=get_environments_obj_for_app(application=application),
@ -88,20 +80,18 @@ def edit_application(portfolio_id, application_id):
)
@user_can(Permissions.EDIT_APPLICATION, message="update application")
def update_application(portfolio_id, application_id):
portfolio = Portfolios.get_for_update(portfolio_id)
application = Applications.get(application_id)
application = Applications.get(application_id, portfolio_id=portfolio_id)
form = ApplicationForm(http_request.form)
if form.validate():
application_data = form.data
Applications.update(application, application_data)
return redirect(
url_for("portfolios.portfolio_applications", portfolio_id=portfolio.id)
url_for("portfolios.portfolio_applications", portfolio_id=portfolio_id)
)
else:
return render_template(
"portfolios/applications/edit.html",
portfolio=portfolio,
application=application,
form=form,
environments_obj=get_environments_obj_for_app(application=application),
@ -111,7 +101,9 @@ def update_application(portfolio_id, application_id):
def wrap_environment_role_lookup(
user, portfolio_id=None, environment_id=None, **kwargs
):
env_role = EnvironmentRoles.get(user.id, environment_id)
env_role = EnvironmentRoles.get_for_portfolio(
user.id, environment_id, portfolio_id=portfolio_id
)
if not env_role:
raise UnauthorizedError(user, "access environment {}".format(environment_id))
@ -121,7 +113,9 @@ def wrap_environment_role_lookup(
@portfolios_bp.route("/portfolios/<portfolio_id>/environments/<environment_id>/access")
@user_can(None, override=wrap_environment_role_lookup, message="access environment")
def access_environment(portfolio_id, environment_id):
env_role = EnvironmentRoles.get(g.current_user.id, environment_id)
env_role = EnvironmentRoles.get_for_portfolio(
g.current_user.id, environment_id, portfolio_id=portfolio_id
)
token = app.csp.cloud.get_access_token(env_role)
return redirect(url_for("atst.csp_environment_access", token=token))
@ -132,7 +126,7 @@ def access_environment(portfolio_id, environment_id):
)
@user_can(Permissions.DELETE_APPLICATION, message="delete application")
def delete_application(portfolio_id, application_id):
application = Applications.get(application_id)
application = Applications.get(application_id, portfolio_id=portfolio_id)
Applications.delete(application)
flash("application_deleted", application_name=application.name)

2
atst/routes/portfolios/index.py
View File

@ -119,7 +119,7 @@ def edit_portfolio_members(portfolio_id):
return redirect(
url_for(
"portfolios.portfolio_admin",
portfolio_id=portfolio.id,
portfolio_id=portfolio_id,
fragment="portfolio-members",
_anchor="portfolio-members",
)

4
atst/routes/portfolios/invitations.py
View File

@ -1,7 +1,6 @@
from flask import g, redirect, url_for, render_template
from . import portfolios_bp
from atst.domain.portfolios import Portfolios
from atst.domain.invitations import Invitations
from atst.queue import queue
from atst.utils.flash import formatted_flash as flash
@ -47,10 +46,9 @@ def accept_invitation(token):
)
@user_can(Permissions.EDIT_PORTFOLIO_USERS, message="revoke invitation")
def revoke_invitation(portfolio_id, token):
portfolio = Portfolios.get_for_update(portfolio_id)
Invitations.revoke(token)
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio.id))
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio_id))
@portfolios_bp.route(

26
atst/routes/portfolios/members.py
View File

@ -41,7 +41,6 @@ def portfolio_members(portfolio_id):
return render_template(
"portfolios/members/index.html",
portfolio=portfolio,
status_choices=MEMBER_STATUS_CHOICES,
members=members_list,
)
@ -51,13 +50,12 @@ def portfolio_members(portfolio_id):
@user_can(Permissions.VIEW_APPLICATION_MEMBER, message="view application members")
def application_members(portfolio_id, application_id):
portfolio = Portfolios.get_for_update(portfolio_id)
application = Applications.get(application_id)
application = Applications.get(application_id, portfolio_id=portfolio_id)
# TODO: this should show only members that have env roles in this application
members_list = [serialize_portfolio_role(k) for k in portfolio.members]
return render_template(
"portfolios/applications/members.html",
portfolio=portfolio,
application=application,
members=members_list,
)
@ -68,11 +66,8 @@ def application_members(portfolio_id, application_id):
Permissions.CREATE_PORTFOLIO_USERS, message="view create new portfolio member form"
)
def new_member(portfolio_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
form = member_forms.NewForm()
return render_template(
"portfolios/members/new.html", portfolio=portfolio, form=form
)
return render_template("portfolios/members/new.html", form=form)
@portfolios_bp.route("/portfolios/<portfolio_id>/members/new", methods=["POST"])
@ -94,7 +89,7 @@ def create_member(portfolio_id):
return redirect(
url_for(
"portfolios.portfolio_admin",
portfolio_id=portfolio.id,
portfolio_id=portfolio_id,
fragment="portfolio-members",
_anchor="portfolio-members",
)
@ -104,9 +99,7 @@ def create_member(portfolio_id):
"error.html", message="There was an error processing your request."
)
else:
return render_template(
"portfolios/members/new.html", portfolio=portfolio, form=form
)
return render_template("portfolios/members/new.html", form=form)
@portfolios_bp.route("/portfolios/<portfolio_id>/members/<member_id>/member_edit")
@ -124,7 +117,6 @@ def view_member(portfolio_id, member_id):
return render_template(
"portfolios/members/edit.html",
portfolio=portfolio,
member=member,
applications=applications,
form=form,
@ -143,7 +135,6 @@ def view_member(portfolio_id, member_id):
)
@user_can(Permissions.EDIT_PORTFOLIO_USERS, message="update portfolio member")
def update_member(portfolio_id, member_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
member = PortfolioRoles.get(portfolio_id, member_id)
ids_and_roles = []
@ -162,15 +153,10 @@ def update_member(portfolio_id, member_id):
flash("environment_access_changed")
return redirect(
url_for("portfolios.portfolio_members", portfolio_id=portfolio.id)
url_for("portfolios.portfolio_members", portfolio_id=portfolio_id)
)
else:
return render_template(
"portfolios/members/edit.html",
form=form,
portfolio=portfolio,
member=member,
)
return render_template("portfolios/members/edit.html", form=form, member=member)
@portfolios_bp.route(

54
atst/routes/portfolios/task_orders.py
View File

@ -57,7 +57,6 @@ def portfolio_funding(portfolio_id):
return render_template(
"portfolios/task_orders/index.html",
portfolio=portfolio,
pending_task_orders=(
task_orders_by_status.get(TaskOrderStatus.STARTED, [])
+ task_orders_by_status.get(TaskOrderStatus.PENDING, [])
@ -71,8 +70,7 @@ def portfolio_funding(portfolio_id):
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>")
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS, message="view task order details")
def view_task_order(portfolio_id, task_order_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
to_form_complete = TaskOrders.all_sections_complete(task_order)
dd_254_complete = DD254s.is_complete(task_order.dd_254)
return render_template(
@ -82,15 +80,14 @@ def view_task_order(portfolio_id, task_order_id):
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),
portfolio=portfolio,
task_order=task_order,
to_form_complete=to_form_complete,
user=g.current_user,
)
def wrap_check_is_ko_or_cor(user, task_order_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id)
def wrap_check_is_ko_or_cor(user, task_order_id=None, portfolio_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
Authorization.check_is_ko_or_cor(user, task_order)
return True
@ -103,13 +100,11 @@ def wrap_check_is_ko_or_cor(user, task_order_id=None, **_kwargs):
message="view contracting officer review form",
)
def ko_review(portfolio_id, task_order_id):
task_order = TaskOrders.get(task_order_id)
portfolio = Portfolios.get(g.current_user, portfolio_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
if TaskOrders.all_sections_complete(task_order):
return render_template(
"/portfolios/task_orders/review.html",
portfolio=portfolio,
task_order=task_order,
form=KOReviewForm(obj=task_order),
)
@ -132,7 +127,7 @@ def resend_invite(portfolio_id, task_order_id):
invite_type_info = OFFICER_INVITATIONS[invite_type]
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
portfolio = Portfolios.get(g.current_user, portfolio_id)
officer = getattr(task_order, invite_type_info["role"])
@ -172,8 +167,8 @@ def resend_invite(portfolio_id, task_order_id):
return redirect(
url_for(
"portfolios.task_order_invitations",
portfolio_id=portfolio.id,
task_order_id=task_order.id,
portfolio_id=portfolio_id,
task_order_id=task_order_id,
)
)
@ -185,7 +180,7 @@ def resend_invite(portfolio_id, task_order_id):
None, override=wrap_check_is_ko_or_cor, message="submit contracting officer review"
)
def submit_ko_review(portfolio_id, task_order_id, form=None):
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
form_data = {**http_request.form, **http_request.files}
form = KOReviewForm(form_data)
@ -207,10 +202,7 @@ def submit_ko_review(portfolio_id, task_order_id, form=None):
)
else:
return render_template(
"/portfolios/task_orders/review.html",
portfolio=Portfolios.get(g.current_user, portfolio_id),
task_order=task_order,
form=form,
"/portfolios/task_orders/review.html", task_order=task_order, form=form
)
@ -221,14 +213,12 @@ def submit_ko_review(portfolio_id, task_order_id, form=None):
Permissions.EDIT_TASK_ORDER_DETAILS, message="view task order invitations page"
)
def task_order_invitations(portfolio_id, task_order_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
form = EditTaskOrderOfficersForm(obj=task_order)
if TaskOrders.all_sections_complete(task_order):
return render_template(
"portfolios/task_orders/invitations.html",
portfolio=portfolio,
task_order=task_order,
form=form,
user=g.current_user,
@ -243,8 +233,7 @@ def task_order_invitations(portfolio_id, task_order_id):
)
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS, message="edit task order invitations")
def edit_task_order_invitations(portfolio_id, task_order_id):
portfolio = Portfolios.get(g.current_user, portfolio_id)
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
form = EditTaskOrderOfficersForm(formdata=http_request.form, obj=task_order)
if form.validate():
@ -256,7 +245,7 @@ def edit_task_order_invitations(portfolio_id, task_order_id):
return redirect(
url_for(
"portfolios.task_order_invitations",
portfolio_id=portfolio.id,
portfolio_id=portfolio_id,
task_order_id=task_order.id,
)
)
@ -264,7 +253,6 @@ def edit_task_order_invitations(portfolio_id, task_order_id):
return (
render_template(
"portfolios/task_orders/invitations.html",
portfolio=portfolio,
task_order=task_order,
form=form,
),
@ -289,8 +277,8 @@ def so_review_form(task_order):
return DD254Form(data=form_data)
def wrap_check_is_so(user, task_order_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id)
def wrap_check_is_so(user, task_order_id=None, portfolio_id=None, **_kwargs):
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
Authorization.check_is_so(user, task_order)
return True
@ -299,14 +287,11 @@ def wrap_check_is_so(user, task_order_id=None, **_kwargs):
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>/dd254")
@user_can(None, override=wrap_check_is_so, message="view security officer review form")
def so_review(portfolio_id, task_order_id):
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
form = so_review_form(task_order)
return render_template(
"portfolios/task_orders/so_review.html",
form=form,
portfolio=task_order.portfolio,
task_order=task_order,
"portfolios/task_orders/so_review.html", form=form, task_order=task_order
)
@ -317,7 +302,7 @@ def so_review(portfolio_id, task_order_id):
None, override=wrap_check_is_so, message="submit security officer review form"
)
def submit_so_review(portfolio_id, task_order_id):
task_order = TaskOrders.get(task_order_id)
task_order = TaskOrders.get(task_order_id, portfolio_id=portfolio_id)
form = DD254Form(http_request.form)
if form.validate():
@ -332,8 +317,5 @@ def submit_so_review(portfolio_id, task_order_id):
)
else:
return render_template(
"portfolios/task_orders/so_review.html",
form=form,
portfolio=task_order.portfolio,
task_order=task_order,
"portfolios/task_orders/so_review.html", form=form, task_order=task_order
)

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

@ -61,7 +61,12 @@ class ShowTaskOrderWorkflow:
@property
def task_order(self):
if not self._task_order and self.task_order_id:
self._task_order = TaskOrders.get(self.task_order_id)
if self.portfolio_id:
self._task_order = TaskOrders.get(
self.task_order_id, portfolio_id=self.portfolio_id
)
else:
self._task_order = TaskOrders.get(self.task_order_id)
return self._task_order
@ -260,6 +265,7 @@ def is_new_task_order(*_args, **kwargs):
)
# TODO: /task_orders/new/<int:screen>/<task_order_id> should not exist
@task_orders_bp.route("/task_orders/new/<int:screen>")
@task_orders_bp.route("/task_orders/new/<int:screen>/<task_order_id>")
@task_orders_bp.route("/portfolios/<portfolio_id>/task_orders/new/<int:screen>")
@ -309,6 +315,7 @@ def new(screen, task_order_id=None, portfolio_id=None):
return render_template(workflow.template, **template_args)
# TODO: /task_orders/new/<int:screen>/<task_order_id> should not exist
@task_orders_bp.route("/task_orders/new/<int:screen>", methods=["POST"])
@task_orders_bp.route("/task_orders/new/<int:screen>/<task_order_id>", methods=["POST"])
@task_orders_bp.route(

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

@ -27,6 +27,7 @@ def wrap_check_is_ko(user, task_order_id=None, **_kwargs):
return True
# TODO: this route needs to be moved under the portfolio blueprint
@task_orders_bp.route("/task_orders/<task_order_id>/digital_signature", methods=["GET"])
@user_can(
None, override=wrap_check_is_ko, message="view contracting officer signature page"
@ -42,6 +43,7 @@ def signature_requested(task_order_id):
)
# TODO: this route needs to be moved under the portfolio blueprint
@task_orders_bp.route(
"/task_orders/<task_order_id>/digital_signature", methods=["POST"]
)

11
tests/domain/test_applications.py
View File

@ -1,4 +1,5 @@
import pytest
from uuid import uuid4
from atst.domain.applications import Applications
from atst.domain.exceptions import NotFoundError
@ -70,6 +71,16 @@ def test_get_excludes_deleted():
Applications.get(app.id)
def test_get_application():
app = ApplicationFactory.create()
assert Applications.get(app.id) == app
assert Applications.get(app.id, portfolio_id=app.portfolio_id) == app
with pytest.raises(NotFoundError):
# make the uuid a string like you'd get from a route
rando_id = str(uuid4())
Applications.get(app.id, portfolio_id=rando_id)
def test_delete_application(session):
app = ApplicationFactory.create()
app_role = ApplicationRoleFactory.create(user=UserFactory.create(), application=app)

27
tests/domain/test_environment_roles.py Normal file
View File

@ -0,0 +1,27 @@
import pytest
from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.exceptions import NotFoundError
from tests.factories import *
def test_get_for_portfolio():
user = UserFactory.create()
portfolio = PortfolioFactory.create()
application = ApplicationFactory.create(portfolio=portfolio)
environment = EnvironmentFactory.create(application=application)
env_role = EnvironmentRoleFactory.create(
environment=environment, user=user, role="basic access"
)
assert (
EnvironmentRoles.get_for_portfolio(
user.id, environment.id, portfolio_id=portfolio.id
)
== env_role
)
with pytest.raises(NotFoundError):
EnvironmentRoles.get_for_portfolio(
user.id, environment.id, portfolio_id=application.id
)

16
tests/routes/portfolios/test_applications.py
View File

@ -329,3 +329,19 @@ def test_delete_application(client, user_session):
# app and envs are soft deleted
assert len(port.applications) == 0
assert len(application.environments) == 0
def test_edit_application_scope(client, user_session):
owner = UserFactory.create()
port1 = PortfolioFactory.create(owner=owner, applications=[{"name": "first app"}])
port2 = PortfolioFactory.create(owner=owner, applications=[{"name": "second app"}])
user_session(owner)
response = client.get(
url_for(
"portfolios.edit_application",
portfolio_id=port2.id,
application_id=port1.applications[0].id,
)
)
assert response.status_code == 404