remove unused domain methods and arguments
This commit is contained in:
parent
de7c69bde7
commit
4a0dd2d432
@ -8,7 +8,7 @@ from atst.models.environment_role import EnvironmentRole
|
|||||||
|
|
||||||
class Applications(object):
|
class Applications(object):
|
||||||
@classmethod
|
@classmethod
|
||||||
def create(cls, user, portfolio, name, description, environment_names):
|
def create(cls, portfolio, name, description, environment_names):
|
||||||
application = Application(
|
application = Application(
|
||||||
portfolio=portfolio, name=name, description=description
|
portfolio=portfolio, name=name, description=description
|
||||||
)
|
)
|
||||||
@ -20,7 +20,7 @@ class Applications(object):
|
|||||||
return application
|
return application
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get(cls, user, portfolio, application_id):
|
def get(cls, application_id):
|
||||||
try:
|
try:
|
||||||
application = (
|
application = (
|
||||||
db.session.query(Application).filter_by(id=application_id).one()
|
db.session.query(Application).filter_by(id=application_id).one()
|
||||||
@ -42,7 +42,7 @@ class Applications(object):
|
|||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_all(cls, user, portfolio_role, portfolio):
|
def get_all(cls, portfolio):
|
||||||
try:
|
try:
|
||||||
applications = (
|
applications = (
|
||||||
db.session.query(Application).filter_by(portfolio_id=portfolio.id).all()
|
db.session.query(Application).filter_by(portfolio_id=portfolio.id).all()
|
||||||
@ -53,7 +53,7 @@ class Applications(object):
|
|||||||
return applications
|
return applications
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def update(cls, user, portfolio, application, new_data):
|
def update(cls, application, new_data):
|
||||||
if "name" in new_data:
|
if "name" in new_data:
|
||||||
application.name = new_data["name"]
|
application.name = new_data["name"]
|
||||||
if "description" in new_data:
|
if "description" in new_data:
|
||||||
|
@ -34,11 +34,11 @@ class AuditLog(object):
|
|||||||
return cls._log(resource=resource, action=action, portfolio=portfolio)
|
return cls._log(resource=resource, action=action, portfolio=portfolio)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_all_events(cls, user, pagination_opts=None):
|
def get_all_events(cls, pagination_opts=None):
|
||||||
return AuditEventQuery.get_all(pagination_opts)
|
return AuditEventQuery.get_all(pagination_opts)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_portfolio_events(cls, user, portfolio, pagination_opts=None):
|
def get_portfolio_events(cls, portfolio, pagination_opts=None):
|
||||||
return AuditEventQuery.get_ws_events(portfolio.id, pagination_opts)
|
return AuditEventQuery.get_ws_events(portfolio.id, pagination_opts)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
@ -10,7 +10,7 @@ from atst.domain.task_orders import TaskOrders
|
|||||||
def evaluate_exceptions(user, permission, exceptions, **kwargs):
|
def evaluate_exceptions(user, permission, exceptions, **kwargs):
|
||||||
return (
|
return (
|
||||||
True
|
True
|
||||||
if True in [exc(g.current_user, permission, **kwargs) for exc in exceptions]
|
if True in [exc(user, permission, **kwargs) for exc in exceptions]
|
||||||
else False
|
else False
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -26,7 +26,7 @@ def user_can_access_decorator(permission, message=None, exceptions=None):
|
|||||||
g.current_user, kwargs["portfolio_id"]
|
g.current_user, kwargs["portfolio_id"]
|
||||||
)
|
)
|
||||||
elif "task_order_id" in kwargs:
|
elif "task_order_id" in kwargs:
|
||||||
task_order = TaskOrders.get(g.current_user, kwargs["task_order_id"])
|
task_order = TaskOrders.get(kwargs["task_order_id"])
|
||||||
access_args["portfolio"] = task_order.portfolio
|
access_args["portfolio"] = task_order.portfolio
|
||||||
|
|
||||||
if exceptions and evaluate_exceptions(
|
if exceptions and evaluate_exceptions(
|
||||||
|
@ -58,7 +58,7 @@ class Environments(object):
|
|||||||
return env
|
return env
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def update_environment_roles(cls, user, portfolio, portfolio_role, ids_and_roles):
|
def update_environment_roles(cls, portfolio_role, ids_and_roles):
|
||||||
updated = False
|
updated = False
|
||||||
|
|
||||||
for id_and_role in ids_and_roles:
|
for id_and_role in ids_and_roles:
|
||||||
@ -92,5 +92,5 @@ class Environments(object):
|
|||||||
return updated
|
return updated
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def revoke_access(cls, user, environment, target_user):
|
def revoke_access(cls, environment, target_user):
|
||||||
EnvironmentRoles.delete(environment.id, target_user.id)
|
EnvironmentRoles.delete(environment.id, target_user.id)
|
||||||
|
@ -116,7 +116,7 @@ class Invitations(object):
|
|||||||
return portfolio_role.latest_invitation
|
return portfolio_role.latest_invitation
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def resend(cls, user, portfolio_id, token):
|
def resend(cls, user, token):
|
||||||
previous_invitation = Invitations._get(token)
|
previous_invitation = Invitations._get(token)
|
||||||
Invitations._update_status(previous_invitation, InvitationStatus.REVOKED)
|
Invitations._update_status(previous_invitation, InvitationStatus.REVOKED)
|
||||||
|
|
||||||
|
@ -36,25 +36,7 @@ class Portfolios(object):
|
|||||||
return ScopedPortfolio(user, portfolio)
|
return ScopedPortfolio(user, portfolio)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_for_update_applications(cls, user, portfolio_id):
|
def get_for_update(cls, portfolio_id):
|
||||||
portfolio = PortfoliosQuery.get(portfolio_id)
|
|
||||||
|
|
||||||
return portfolio
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_for_update_information(cls, user, portfolio_id):
|
|
||||||
portfolio = PortfoliosQuery.get(portfolio_id)
|
|
||||||
|
|
||||||
return portfolio
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_for_update_member(cls, user, portfolio_id):
|
|
||||||
portfolio = PortfoliosQuery.get(portfolio_id)
|
|
||||||
|
|
||||||
return portfolio
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_with_members(cls, user, portfolio_id):
|
|
||||||
portfolio = PortfoliosQuery.get(portfolio_id)
|
portfolio = PortfoliosQuery.get(portfolio_id)
|
||||||
|
|
||||||
return portfolio
|
return portfolio
|
||||||
@ -68,7 +50,7 @@ class Portfolios(object):
|
|||||||
return portfolios
|
return portfolios
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def create_member(cls, user, portfolio, data):
|
def create_member(cls, portfolio, data):
|
||||||
new_user = Users.get_or_create_by_dod_id(
|
new_user = Users.get_or_create_by_dod_id(
|
||||||
data["dod_id"],
|
data["dod_id"],
|
||||||
first_name=data["first_name"],
|
first_name=data["first_name"],
|
||||||
@ -87,7 +69,7 @@ class Portfolios(object):
|
|||||||
return portfolio_role
|
return portfolio_role
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def update_member(cls, user, portfolio, member, permission_sets):
|
def update_member(cls, member, permission_sets):
|
||||||
return PortfolioRoles.update(member, permission_sets)
|
return PortfolioRoles.update(member, permission_sets)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@ -118,7 +100,7 @@ class Portfolios(object):
|
|||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def revoke_access(cls, user, portfolio_id, portfolio_role_id):
|
def revoke_access(cls, portfolio_id, portfolio_role_id):
|
||||||
portfolio = PortfoliosQuery.get(portfolio_id)
|
portfolio = PortfoliosQuery.get(portfolio_id)
|
||||||
portfolio_role = PortfolioRoles.get_by_id(portfolio_role_id)
|
portfolio_role = PortfolioRoles.get_by_id(portfolio_role_id)
|
||||||
|
|
||||||
@ -127,7 +109,7 @@ class Portfolios(object):
|
|||||||
|
|
||||||
portfolio_role.status = PortfolioRoleStatus.DISABLED
|
portfolio_role.status = PortfolioRoleStatus.DISABLED
|
||||||
for environment in portfolio.all_environments:
|
for environment in portfolio.all_environments:
|
||||||
Environments.revoke_access(user, environment, portfolio_role.user)
|
Environments.revoke_access(environment, portfolio_role.user)
|
||||||
PortfoliosQuery.add_and_commit(portfolio_role)
|
PortfoliosQuery.add_and_commit(portfolio_role)
|
||||||
|
|
||||||
return portfolio_role
|
return portfolio_role
|
||||||
|
@ -52,7 +52,7 @@ class TaskOrders(object):
|
|||||||
UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"]
|
UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"]
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get(cls, user, task_order_id):
|
def get(cls, task_order_id):
|
||||||
try:
|
try:
|
||||||
task_order = db.session.query(TaskOrder).filter_by(id=task_order_id).one()
|
task_order = db.session.query(TaskOrder).filter_by(id=task_order_id).one()
|
||||||
|
|
||||||
@ -70,7 +70,7 @@ class TaskOrders(object):
|
|||||||
return task_order
|
return task_order
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def update(cls, user, task_order, **kwargs):
|
def update(cls, task_order, **kwargs):
|
||||||
for key, value in kwargs.items():
|
for key, value in kwargs.items():
|
||||||
setattr(task_order, key, value)
|
setattr(task_order, key, value)
|
||||||
|
|
||||||
@ -135,7 +135,7 @@ class TaskOrders(object):
|
|||||||
]
|
]
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def add_officer(cls, user, task_order, officer_type, officer_data):
|
def add_officer(cls, task_order, officer_type, officer_data):
|
||||||
if officer_type in TaskOrders.OFFICERS:
|
if officer_type in TaskOrders.OFFICERS:
|
||||||
portfolio = task_order.portfolio
|
portfolio = task_order.portfolio
|
||||||
|
|
||||||
@ -152,7 +152,6 @@ class TaskOrders(object):
|
|||||||
portfolio_user = existing_member.user
|
portfolio_user = existing_member.user
|
||||||
else:
|
else:
|
||||||
member = Portfolios.create_member(
|
member = Portfolios.create_member(
|
||||||
user,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
{
|
{
|
||||||
**officer_data,
|
**officer_data,
|
||||||
|
@ -148,7 +148,7 @@ def logout():
|
|||||||
@user_can(Permissions.VIEW_AUDIT_LOG, message="view activity log")
|
@user_can(Permissions.VIEW_AUDIT_LOG, message="view activity log")
|
||||||
def activity_history():
|
def activity_history():
|
||||||
pagination_opts = Paginator.get_pagination_opts(request)
|
pagination_opts = Paginator.get_pagination_opts(request)
|
||||||
audit_events = AuditLog.get_all_events(g.current_user, pagination_opts)
|
audit_events = AuditLog.get_all_events(pagination_opts)
|
||||||
return render_template("audit_log/audit_log.html", audit_events=audit_events)
|
return render_template("audit_log/audit_log.html", audit_events=audit_events)
|
||||||
|
|
||||||
|
|
||||||
|
@ -27,7 +27,7 @@ def portfolio_applications(portfolio_id):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new")
|
||||||
@user_can(Permissions.CREATE_APPLICATION)
|
@user_can(Permissions.CREATE_APPLICATION)
|
||||||
def new_application(portfolio_id):
|
def new_application(portfolio_id):
|
||||||
portfolio = Portfolios.get_for_update_applications(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
form = NewApplicationForm()
|
form = NewApplicationForm()
|
||||||
return render_template(
|
return render_template(
|
||||||
"portfolios/applications/new.html", portfolio=portfolio, form=form
|
"portfolios/applications/new.html", portfolio=portfolio, form=form
|
||||||
@ -37,13 +37,12 @@ def new_application(portfolio_id):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new", methods=["POST"])
|
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/new", methods=["POST"])
|
||||||
@user_can(Permissions.CREATE_APPLICATION)
|
@user_can(Permissions.CREATE_APPLICATION)
|
||||||
def create_application(portfolio_id):
|
def create_application(portfolio_id):
|
||||||
portfolio = Portfolios.get_for_update_applications(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
form = NewApplicationForm(http_request.form)
|
form = NewApplicationForm(http_request.form)
|
||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
application_data = form.data
|
application_data = form.data
|
||||||
Applications.create(
|
Applications.create(
|
||||||
g.current_user,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
application_data["name"],
|
application_data["name"],
|
||||||
application_data["description"],
|
application_data["description"],
|
||||||
@ -61,8 +60,8 @@ def create_application(portfolio_id):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/<application_id>/edit")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/<application_id>/edit")
|
||||||
@user_can(Permissions.EDIT_APPLICATION)
|
@user_can(Permissions.EDIT_APPLICATION)
|
||||||
def edit_application(portfolio_id, application_id):
|
def edit_application(portfolio_id, application_id):
|
||||||
portfolio = Portfolios.get_for_update_applications(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
application = Applications.get(g.current_user, portfolio, application_id)
|
application = Applications.get(application_id)
|
||||||
form = ApplicationForm(name=application.name, description=application.description)
|
form = ApplicationForm(name=application.name, description=application.description)
|
||||||
|
|
||||||
return render_template(
|
return render_template(
|
||||||
@ -78,12 +77,12 @@ def edit_application(portfolio_id, application_id):
|
|||||||
)
|
)
|
||||||
@user_can(Permissions.EDIT_APPLICATION)
|
@user_can(Permissions.EDIT_APPLICATION)
|
||||||
def update_application(portfolio_id, application_id):
|
def update_application(portfolio_id, application_id):
|
||||||
portfolio = Portfolios.get_for_update_applications(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
application = Applications.get(g.current_user, portfolio, application_id)
|
application = Applications.get(application_id)
|
||||||
form = ApplicationForm(http_request.form)
|
form = ApplicationForm(http_request.form)
|
||||||
if form.validate():
|
if form.validate():
|
||||||
application_data = form.data
|
application_data = form.data
|
||||||
Applications.update(g.current_user, portfolio, application, application_data)
|
Applications.update(application, application_data)
|
||||||
|
|
||||||
return redirect(
|
return redirect(
|
||||||
url_for("portfolios.portfolio_applications", portfolio_id=portfolio.id)
|
url_for("portfolios.portfolio_applications", portfolio_id=portfolio.id)
|
||||||
|
@ -40,12 +40,10 @@ def serialize_member(member):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/admin")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/admin")
|
||||||
@user_can(Permissions.VIEW_PORTFOLIO_ADMIN)
|
@user_can(Permissions.VIEW_PORTFOLIO_ADMIN)
|
||||||
def portfolio_admin(portfolio_id):
|
def portfolio_admin(portfolio_id):
|
||||||
portfolio = Portfolios.get_for_update_information(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
form = PortfolioForm(data={"name": portfolio.name})
|
form = PortfolioForm(data={"name": portfolio.name})
|
||||||
pagination_opts = Paginator.get_pagination_opts(http_request)
|
pagination_opts = Paginator.get_pagination_opts(http_request)
|
||||||
audit_events = AuditLog.get_portfolio_events(
|
audit_events = AuditLog.get_portfolio_events(portfolio, pagination_opts)
|
||||||
g.current_user, portfolio, pagination_opts
|
|
||||||
)
|
|
||||||
members_data = [serialize_member(member) for member in portfolio.members]
|
members_data = [serialize_member(member) for member in portfolio.members]
|
||||||
return render_template(
|
return render_template(
|
||||||
"portfolios/admin.html",
|
"portfolios/admin.html",
|
||||||
@ -60,7 +58,7 @@ def portfolio_admin(portfolio_id):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/edit", methods=["POST"])
|
@portfolios_bp.route("/portfolios/<portfolio_id>/edit", methods=["POST"])
|
||||||
@user_can(Permissions.EDIT_PORTFOLIO_NAME)
|
@user_can(Permissions.EDIT_PORTFOLIO_NAME)
|
||||||
def edit_portfolio(portfolio_id):
|
def edit_portfolio(portfolio_id):
|
||||||
portfolio = Portfolios.get_for_update_information(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
form = PortfolioForm(http_request.form)
|
form = PortfolioForm(http_request.form)
|
||||||
if form.validate():
|
if form.validate():
|
||||||
Portfolios.update(portfolio, form.data)
|
Portfolios.update(portfolio, form.data)
|
||||||
|
@ -47,7 +47,7 @@ def accept_invitation(token):
|
|||||||
)
|
)
|
||||||
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
||||||
def revoke_invitation(portfolio_id, token):
|
def revoke_invitation(portfolio_id, token):
|
||||||
portfolio = Portfolios.get_for_update_member(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
Invitations.revoke(token)
|
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))
|
||||||
@ -58,7 +58,7 @@ def revoke_invitation(portfolio_id, token):
|
|||||||
)
|
)
|
||||||
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
||||||
def resend_invitation(portfolio_id, token):
|
def resend_invitation(portfolio_id, token):
|
||||||
invite = Invitations.resend(g.current_user, portfolio_id, token)
|
invite = Invitations.resend(g.current_user, token)
|
||||||
send_invite_email(g.current_user.full_name, invite.token, invite.email)
|
send_invite_email(g.current_user.full_name, invite.token, invite.email)
|
||||||
flash("resend_portfolio_invitation", user_name=invite.user_name)
|
flash("resend_portfolio_invitation", user_name=invite.user_name)
|
||||||
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio_id))
|
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio_id))
|
||||||
|
@ -36,7 +36,7 @@ def serialize_portfolio_role(portfolio_role):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/members")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/members")
|
||||||
@user_can(Permissions.VIEW_PORTFOLIO_USERS)
|
@user_can(Permissions.VIEW_PORTFOLIO_USERS)
|
||||||
def portfolio_members(portfolio_id):
|
def portfolio_members(portfolio_id):
|
||||||
portfolio = Portfolios.get_with_members(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
members_list = [serialize_portfolio_role(k) for k in portfolio.members]
|
members_list = [serialize_portfolio_role(k) for k in portfolio.members]
|
||||||
|
|
||||||
return render_template(
|
return render_template(
|
||||||
@ -50,8 +50,8 @@ def portfolio_members(portfolio_id):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/<application_id>/members")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/applications/<application_id>/members")
|
||||||
@user_can(Permissions.VIEW_APPLICATION_MEMBER)
|
@user_can(Permissions.VIEW_APPLICATION_MEMBER)
|
||||||
def application_members(portfolio_id, application_id):
|
def application_members(portfolio_id, application_id):
|
||||||
portfolio = Portfolios.get_with_members(g.current_user, portfolio_id)
|
portfolio = Portfolios.get_for_update(portfolio_id)
|
||||||
application = Applications.get(g.current_user, portfolio, application_id)
|
application = Applications.get(application_id)
|
||||||
# TODO: this should show only members that have env roles in this application
|
# TODO: this should show only members that have env roles in this application
|
||||||
members_list = [serialize_portfolio_role(k) for k in portfolio.members]
|
members_list = [serialize_portfolio_role(k) for k in portfolio.members]
|
||||||
|
|
||||||
@ -81,7 +81,7 @@ def create_member(portfolio_id):
|
|||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
try:
|
try:
|
||||||
member = Portfolios.create_member(g.current_user, portfolio, form.data)
|
member = Portfolios.create_member(portfolio, form.data)
|
||||||
invite_service = InvitationService(
|
invite_service = InvitationService(
|
||||||
g.current_user, member, form.data.get("email")
|
g.current_user, member, form.data.get("email")
|
||||||
)
|
)
|
||||||
@ -107,7 +107,7 @@ def create_member(portfolio_id):
|
|||||||
def view_member(portfolio_id, member_id):
|
def view_member(portfolio_id, member_id):
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
member = PortfolioRoles.get(portfolio_id, member_id)
|
member = PortfolioRoles.get(portfolio_id, member_id)
|
||||||
applications = Applications.get_all(g.current_user, member, portfolio)
|
applications = Applications.get_all(portfolio)
|
||||||
form = member_forms.EditForm(portfolio_role="admin")
|
form = member_forms.EditForm(portfolio_role="admin")
|
||||||
editable = g.current_user == member.user
|
editable = g.current_user == member.user
|
||||||
can_revoke_access = Portfolios.can_revoke_access_for(portfolio, member)
|
can_revoke_access = Portfolios.can_revoke_access_for(portfolio, member)
|
||||||
@ -147,12 +147,8 @@ def update_member(portfolio_id, member_id):
|
|||||||
|
|
||||||
form = member_forms.EditForm(http_request.form)
|
form = member_forms.EditForm(http_request.form)
|
||||||
if form.validate():
|
if form.validate():
|
||||||
member = Portfolios.update_member(
|
member = Portfolios.update_member(member, form.data["permission_sets"])
|
||||||
g.current_user, portfolio, member, form.data["permission_sets"]
|
updated_roles = Environments.update_environment_roles(member, ids_and_roles)
|
||||||
)
|
|
||||||
updated_roles = Environments.update_environment_roles(
|
|
||||||
g.current_user, portfolio, member, ids_and_roles
|
|
||||||
)
|
|
||||||
if updated_roles:
|
if updated_roles:
|
||||||
flash("environment_access_changed")
|
flash("environment_access_changed")
|
||||||
|
|
||||||
@ -173,6 +169,6 @@ def update_member(portfolio_id, member_id):
|
|||||||
)
|
)
|
||||||
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
@user_can(Permissions.EDIT_PORTFOLIO_USERS)
|
||||||
def revoke_access(portfolio_id, member_id):
|
def revoke_access(portfolio_id, member_id):
|
||||||
revoked_role = Portfolios.revoke_access(g.current_user, portfolio_id, member_id)
|
revoked_role = Portfolios.revoke_access(portfolio_id, member_id)
|
||||||
flash("revoked_portfolio_access", member_name=revoked_role.user.full_name)
|
flash("revoked_portfolio_access", member_name=revoked_role.user.full_name)
|
||||||
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio_id))
|
return redirect(url_for("portfolios.portfolio_members", portfolio_id=portfolio_id))
|
||||||
|
@ -72,7 +72,7 @@ def portfolio_funding(portfolio_id):
|
|||||||
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
||||||
def view_task_order(portfolio_id, task_order_id):
|
def view_task_order(portfolio_id, task_order_id):
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
to_form_complete = TaskOrders.all_sections_complete(task_order)
|
to_form_complete = TaskOrders.all_sections_complete(task_order)
|
||||||
dd_254_complete = DD254s.is_complete(task_order.dd_254)
|
dd_254_complete = DD254s.is_complete(task_order.dd_254)
|
||||||
return render_template(
|
return render_template(
|
||||||
@ -90,7 +90,7 @@ def view_task_order(portfolio_id, task_order_id):
|
|||||||
|
|
||||||
|
|
||||||
def wrap_check_is_ko_or_cor(user, _perm, task_order_id=None, **_kwargs):
|
def wrap_check_is_ko_or_cor(user, _perm, task_order_id=None, **_kwargs):
|
||||||
task_order = TaskOrders.get(user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
Authorization.check_is_ko_or_cor(user, task_order)
|
Authorization.check_is_ko_or_cor(user, task_order)
|
||||||
|
|
||||||
return True
|
return True
|
||||||
@ -99,7 +99,7 @@ def wrap_check_is_ko_or_cor(user, _perm, task_order_id=None, **_kwargs):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>/review")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>/review")
|
||||||
@user_can(None, exceptions=[wrap_check_is_ko_or_cor])
|
@user_can(None, exceptions=[wrap_check_is_ko_or_cor])
|
||||||
def ko_review(portfolio_id, task_order_id):
|
def ko_review(portfolio_id, task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
|
|
||||||
if TaskOrders.all_sections_complete(task_order):
|
if TaskOrders.all_sections_complete(task_order):
|
||||||
@ -118,7 +118,7 @@ def ko_review(portfolio_id, task_order_id):
|
|||||||
methods=["POST"],
|
methods=["POST"],
|
||||||
)
|
)
|
||||||
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
||||||
def resend_invite(portfolio_id, task_order_id, form=None):
|
def resend_invite(portfolio_id, task_order_id):
|
||||||
invite_type = http_request.args.get("invite_type")
|
invite_type = http_request.args.get("invite_type")
|
||||||
|
|
||||||
if invite_type not in OFFICER_INVITATIONS:
|
if invite_type not in OFFICER_INVITATIONS:
|
||||||
@ -126,7 +126,7 @@ def resend_invite(portfolio_id, task_order_id, form=None):
|
|||||||
|
|
||||||
invite_type_info = OFFICER_INVITATIONS[invite_type]
|
invite_type_info = OFFICER_INVITATIONS[invite_type]
|
||||||
|
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
|
|
||||||
officer = getattr(task_order, invite_type_info["role"])
|
officer = getattr(task_order, invite_type_info["role"])
|
||||||
@ -177,12 +177,12 @@ def resend_invite(portfolio_id, task_order_id, form=None):
|
|||||||
)
|
)
|
||||||
@user_can(None, exceptions=[wrap_check_is_ko_or_cor])
|
@user_can(None, exceptions=[wrap_check_is_ko_or_cor])
|
||||||
def submit_ko_review(portfolio_id, task_order_id, form=None):
|
def submit_ko_review(portfolio_id, task_order_id, form=None):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
form_data = {**http_request.form, **http_request.files}
|
form_data = {**http_request.form, **http_request.files}
|
||||||
form = KOReviewForm(form_data)
|
form = KOReviewForm(form_data)
|
||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
TaskOrders.update(user=g.current_user, task_order=task_order, **form.data)
|
TaskOrders.update(task_order=task_order, **form.data)
|
||||||
if Authorization.is_ko(g.current_user, task_order) and TaskOrders.can_ko_sign(
|
if Authorization.is_ko(g.current_user, task_order) and TaskOrders.can_ko_sign(
|
||||||
task_order
|
task_order
|
||||||
):
|
):
|
||||||
@ -212,7 +212,7 @@ def submit_ko_review(portfolio_id, task_order_id, form=None):
|
|||||||
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
||||||
def task_order_invitations(portfolio_id, task_order_id):
|
def task_order_invitations(portfolio_id, task_order_id):
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
form = EditTaskOrderOfficersForm(obj=task_order)
|
form = EditTaskOrderOfficersForm(obj=task_order)
|
||||||
|
|
||||||
if TaskOrders.all_sections_complete(task_order):
|
if TaskOrders.all_sections_complete(task_order):
|
||||||
@ -233,7 +233,7 @@ def task_order_invitations(portfolio_id, task_order_id):
|
|||||||
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
||||||
def edit_task_order_invitations(portfolio_id, task_order_id):
|
def edit_task_order_invitations(portfolio_id, task_order_id):
|
||||||
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
portfolio = Portfolios.get(g.current_user, portfolio_id)
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
form = EditTaskOrderOfficersForm(formdata=http_request.form, obj=task_order)
|
form = EditTaskOrderOfficersForm(formdata=http_request.form, obj=task_order)
|
||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
@ -279,7 +279,7 @@ def so_review_form(task_order):
|
|||||||
|
|
||||||
|
|
||||||
def wrap_check_is_so(user, _perm, task_order_id=None, **_kwargs):
|
def wrap_check_is_so(user, _perm, task_order_id=None, **_kwargs):
|
||||||
task_order = TaskOrders.get(user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
Authorization.check_is_so(user, task_order)
|
Authorization.check_is_so(user, task_order)
|
||||||
|
|
||||||
return True
|
return True
|
||||||
@ -288,7 +288,7 @@ def wrap_check_is_so(user, _perm, task_order_id=None, **_kwargs):
|
|||||||
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>/dd254")
|
@portfolios_bp.route("/portfolios/<portfolio_id>/task_order/<task_order_id>/dd254")
|
||||||
@user_can(None, exceptions=[wrap_check_is_so])
|
@user_can(None, exceptions=[wrap_check_is_so])
|
||||||
def so_review(portfolio_id, task_order_id):
|
def so_review(portfolio_id, task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
form = so_review_form(task_order)
|
form = so_review_form(task_order)
|
||||||
|
|
||||||
return render_template(
|
return render_template(
|
||||||
@ -304,7 +304,7 @@ def so_review(portfolio_id, task_order_id):
|
|||||||
)
|
)
|
||||||
@user_can(None, exceptions=[wrap_check_is_so])
|
@user_can(None, exceptions=[wrap_check_is_so])
|
||||||
def submit_so_review(portfolio_id, task_order_id):
|
def submit_so_review(portfolio_id, task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
form = DD254Form(http_request.form)
|
form = DD254Form(http_request.form)
|
||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
|
@ -1,5 +1,5 @@
|
|||||||
from io import BytesIO
|
from io import BytesIO
|
||||||
from flask import g, Response, current_app as app
|
from flask import Response, current_app as app
|
||||||
|
|
||||||
from . import task_orders_bp
|
from . import task_orders_bp
|
||||||
from atst.domain.task_orders import TaskOrders
|
from atst.domain.task_orders import TaskOrders
|
||||||
@ -12,7 +12,7 @@ from atst.models.permissions import Permissions
|
|||||||
@task_orders_bp.route("/task_orders/download_summary/<task_order_id>")
|
@task_orders_bp.route("/task_orders/download_summary/<task_order_id>")
|
||||||
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
||||||
def download_summary(task_order_id):
|
def download_summary(task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
byte_str = BytesIO()
|
byte_str = BytesIO()
|
||||||
Docx.render(byte_str, data=task_order.to_dictionary())
|
Docx.render(byte_str, data=task_order.to_dictionary())
|
||||||
filename = "{}.docx".format(task_order.portfolio_name)
|
filename = "{}.docx".format(task_order.portfolio_name)
|
||||||
@ -36,7 +36,7 @@ def send_file(attachment):
|
|||||||
@task_orders_bp.route("/task_orders/csp_estimate/<task_order_id>")
|
@task_orders_bp.route("/task_orders/csp_estimate/<task_order_id>")
|
||||||
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
||||||
def download_csp_estimate(task_order_id):
|
def download_csp_estimate(task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
if task_order.csp_estimate:
|
if task_order.csp_estimate:
|
||||||
return send_file(task_order.csp_estimate)
|
return send_file(task_order.csp_estimate)
|
||||||
else:
|
else:
|
||||||
@ -46,7 +46,7 @@ def download_csp_estimate(task_order_id):
|
|||||||
@task_orders_bp.route("/task_orders/pdf/<task_order_id>")
|
@task_orders_bp.route("/task_orders/pdf/<task_order_id>")
|
||||||
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
@user_can(Permissions.VIEW_TASK_ORDER_DETAILS)
|
||||||
def download_task_order_pdf(task_order_id):
|
def download_task_order_pdf(task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
if task_order.pdf:
|
if task_order.pdf:
|
||||||
return send_file(task_order.pdf)
|
return send_file(task_order.pdf)
|
||||||
else:
|
else:
|
||||||
|
@ -11,7 +11,7 @@ from atst.models.permissions import Permissions
|
|||||||
@task_orders_bp.route("/task_orders/invite/<task_order_id>", methods=["POST"])
|
@task_orders_bp.route("/task_orders/invite/<task_order_id>", methods=["POST"])
|
||||||
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
@user_can(Permissions.EDIT_TASK_ORDER_DETAILS)
|
||||||
def invite(task_order_id):
|
def invite(task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
if TaskOrders.all_sections_complete(task_order):
|
if TaskOrders.all_sections_complete(task_order):
|
||||||
update_officer_invitations(g.current_user, task_order)
|
update_officer_invitations(g.current_user, task_order)
|
||||||
|
|
||||||
|
@ -61,7 +61,7 @@ class ShowTaskOrderWorkflow:
|
|||||||
@property
|
@property
|
||||||
def task_order(self):
|
def task_order(self):
|
||||||
if not self._task_order and self.task_order_id:
|
if not self._task_order and self.task_order_id:
|
||||||
self._task_order = TaskOrders.get(self.user, self.task_order_id)
|
self._task_order = TaskOrders.get(self.task_order_id)
|
||||||
|
|
||||||
return self._task_order
|
return self._task_order
|
||||||
|
|
||||||
@ -230,7 +230,7 @@ class UpdateTaskOrderWorkflow(ShowTaskOrderWorkflow):
|
|||||||
old_name = self.task_order.portfolio_name
|
old_name = self.task_order.portfolio_name
|
||||||
if not new_name == old_name:
|
if not new_name == old_name:
|
||||||
Portfolios.update(self.task_order.portfolio, {"name": new_name})
|
Portfolios.update(self.task_order.portfolio, {"name": new_name})
|
||||||
TaskOrders.update(self.user, self.task_order, **self.task_order_form_data)
|
TaskOrders.update(self.task_order, **self.task_order_form_data)
|
||||||
else:
|
else:
|
||||||
if self.portfolio_id:
|
if self.portfolio_id:
|
||||||
pf = Portfolios.get(self.user, self.portfolio_id)
|
pf = Portfolios.get(self.user, self.portfolio_id)
|
||||||
@ -241,7 +241,7 @@ class UpdateTaskOrderWorkflow(ShowTaskOrderWorkflow):
|
|||||||
self.form.defense_component.data,
|
self.form.defense_component.data,
|
||||||
)
|
)
|
||||||
self._task_order = TaskOrders.create(portfolio=pf, creator=self.user)
|
self._task_order = TaskOrders.create(portfolio=pf, creator=self.user)
|
||||||
TaskOrders.update(self.user, self.task_order, **self.task_order_form_data)
|
TaskOrders.update(self.task_order, **self.task_order_form_data)
|
||||||
|
|
||||||
return self.task_order
|
return self.task_order
|
||||||
|
|
||||||
@ -251,7 +251,7 @@ def get_started():
|
|||||||
return render_template("task_orders/new/get_started.html") # pragma: no cover
|
return render_template("task_orders/new/get_started.html") # pragma: no cover
|
||||||
|
|
||||||
|
|
||||||
def is_new_task_order(*args, **kwargs):
|
def is_new_task_order(*_args, **kwargs):
|
||||||
return (
|
return (
|
||||||
"screen" in kwargs
|
"screen" in kwargs
|
||||||
and kwargs["screen"] == 1
|
and kwargs["screen"] == 1
|
||||||
|
@ -12,7 +12,7 @@ from atst.domain.authz.decorator import user_can_access_decorator as user_can
|
|||||||
|
|
||||||
|
|
||||||
def find_unsigned_ko_to(task_order_id):
|
def find_unsigned_ko_to(task_order_id):
|
||||||
task_order = TaskOrders.get(g.current_user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
|
|
||||||
if not TaskOrders.can_ko_sign(task_order):
|
if not TaskOrders.can_ko_sign(task_order):
|
||||||
raise NoAccessError("task_order")
|
raise NoAccessError("task_order")
|
||||||
@ -21,7 +21,7 @@ def find_unsigned_ko_to(task_order_id):
|
|||||||
|
|
||||||
|
|
||||||
def wrap_check_is_ko(user, _perm, task_order_id=None, **_kwargs):
|
def wrap_check_is_ko(user, _perm, task_order_id=None, **_kwargs):
|
||||||
task_order = TaskOrders.get(user, task_order_id)
|
task_order = TaskOrders.get(task_order_id)
|
||||||
Authorization.check_is_ko(user, task_order)
|
Authorization.check_is_ko(user, task_order)
|
||||||
|
|
||||||
return True
|
return True
|
||||||
@ -58,7 +58,6 @@ def record_signature(task_order_id):
|
|||||||
|
|
||||||
if form.validate():
|
if form.validate():
|
||||||
TaskOrders.update(
|
TaskOrders.update(
|
||||||
user=g.current_user,
|
|
||||||
task_order=task_order,
|
task_order=task_order,
|
||||||
signer_dod_id=g.current_user.dod_id,
|
signer_dod_id=g.current_user.dod_id,
|
||||||
signed_at=datetime.datetime.now(),
|
signed_at=datetime.datetime.now(),
|
||||||
|
@ -33,7 +33,7 @@ def update_officer_invitations(user, task_order):
|
|||||||
):
|
):
|
||||||
officer_data = task_order.officer_dictionary(invite_opts["role"])
|
officer_data = task_order.officer_dictionary(invite_opts["role"])
|
||||||
officer = TaskOrders.add_officer(
|
officer = TaskOrders.add_officer(
|
||||||
user, task_order, invite_opts["role"], officer_data
|
task_order, invite_opts["role"], officer_data
|
||||||
)
|
)
|
||||||
pf_officer_member = PortfolioRoles.get(task_order.portfolio.id, officer.id)
|
pf_officer_member = PortfolioRoles.get(task_order.portfolio.id, officer.id)
|
||||||
invite_service = Invitation(
|
invite_service = Invitation(
|
||||||
|
@ -68,7 +68,7 @@ def get_users():
|
|||||||
|
|
||||||
def add_members_to_portfolio(portfolio):
|
def add_members_to_portfolio(portfolio):
|
||||||
for portfolio_role in PORTFOLIO_USERS:
|
for portfolio_role in PORTFOLIO_USERS:
|
||||||
ws_role = Portfolios.create_member(portfolio.owner, portfolio, portfolio_role)
|
ws_role = Portfolios.create_member(portfolio, portfolio_role)
|
||||||
db.session.refresh(ws_role)
|
db.session.refresh(ws_role)
|
||||||
PortfolioRoles.enable(ws_role)
|
PortfolioRoles.enable(ws_role)
|
||||||
|
|
||||||
@ -114,7 +114,6 @@ def create_task_order(portfolio, start, end, clin_01=None, clin_03=None):
|
|||||||
def add_applications_to_portfolio(portfolio, applications):
|
def add_applications_to_portfolio(portfolio, applications):
|
||||||
for application in applications:
|
for application in applications:
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio.owner,
|
|
||||||
portfolio=portfolio,
|
portfolio=portfolio,
|
||||||
name=application["name"],
|
name=application["name"],
|
||||||
description=application["description"],
|
description=application["description"],
|
||||||
|
@ -6,7 +6,7 @@ from atst.domain.portfolios import Portfolios
|
|||||||
def test_create_application_with_multiple_environments():
|
def test_create_application_with_multiple_environments():
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
portfolio.owner, portfolio, "My Test Application", "Test", ["dev", "prod"]
|
portfolio, "My Test Application", "Test", ["dev", "prod"]
|
||||||
)
|
)
|
||||||
|
|
||||||
assert application.portfolio == portfolio
|
assert application.portfolio == portfolio
|
||||||
@ -21,7 +21,7 @@ def test_portfolio_owner_can_view_environments():
|
|||||||
owner=owner,
|
owner=owner,
|
||||||
applications=[{"environments": [{"name": "dev"}, {"name": "prod"}]}],
|
applications=[{"environments": [{"name": "dev"}, {"name": "prod"}]}],
|
||||||
)
|
)
|
||||||
application = Applications.get(owner, portfolio, portfolio.applications[0].id)
|
application = Applications.get(portfolio.applications[0].id)
|
||||||
|
|
||||||
assert len(application.environments) == 2
|
assert len(application.environments) == 2
|
||||||
|
|
||||||
@ -38,11 +38,9 @@ def test_can_only_update_name_and_description():
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
application = Applications.get(owner, portfolio, portfolio.applications[0].id)
|
application = Applications.get(portfolio.applications[0].id)
|
||||||
env_name = application.environments[0].name
|
env_name = application.environments[0].name
|
||||||
Applications.update(
|
Applications.update(
|
||||||
owner,
|
|
||||||
portfolio,
|
|
||||||
application,
|
application,
|
||||||
{
|
{
|
||||||
"name": "New Name",
|
"name": "New Name",
|
||||||
|
@ -28,39 +28,43 @@ def test_non_admin_cannot_view_audit_log(developer):
|
|||||||
AuditLog.get_all_events(developer)
|
AuditLog.get_all_events(developer)
|
||||||
|
|
||||||
|
|
||||||
def test_ccpo_can_view_audit_log(ccpo):
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
events = AuditLog.get_all_events(ccpo)
|
def test_ccpo_can_view_audit_log():
|
||||||
|
events = AuditLog.get_all_events()
|
||||||
assert len(events) > 0
|
assert len(events) > 0
|
||||||
|
|
||||||
|
|
||||||
def test_paginate_audit_log(ccpo):
|
def test_paginate_audit_log():
|
||||||
user = UserFactory.create()
|
user = UserFactory.create()
|
||||||
for _ in range(100):
|
for _ in range(100):
|
||||||
AuditLog.log_system_event(user, action="create")
|
AuditLog.log_system_event(user, action="create")
|
||||||
|
|
||||||
events = AuditLog.get_all_events(ccpo, pagination_opts={"per_page": 25, "page": 2})
|
events = AuditLog.get_all_events(pagination_opts={"per_page": 25, "page": 2})
|
||||||
assert len(events) == 25
|
assert len(events) == 25
|
||||||
|
|
||||||
|
|
||||||
def test_ccpo_can_view_ws_audit_log(ccpo):
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
|
def test_ccpo_can_view_ws_audit_log():
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
events = AuditLog.get_portfolio_events(ccpo, portfolio)
|
events = AuditLog.get_portfolio_events(portfolio)
|
||||||
assert len(events) > 0
|
assert len(events) > 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
def test_ws_admin_can_view_ws_audit_log():
|
def test_ws_admin_can_view_ws_audit_log():
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
admin = UserFactory.create()
|
admin = UserFactory.create()
|
||||||
PortfolioRoleFactory.create(
|
PortfolioRoleFactory.create(
|
||||||
portfolio=portfolio, user=admin, status=PortfolioRoleStatus.ACTIVE
|
portfolio=portfolio, user=admin, status=PortfolioRoleStatus.ACTIVE
|
||||||
)
|
)
|
||||||
events = AuditLog.get_portfolio_events(admin, portfolio)
|
events = AuditLog.get_portfolio_events(portfolio)
|
||||||
assert len(events) > 0
|
assert len(events) > 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
def test_ws_owner_can_view_ws_audit_log():
|
def test_ws_owner_can_view_ws_audit_log():
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
events = AuditLog.get_portfolio_events(portfolio.owner, portfolio)
|
events = AuditLog.get_portfolio_events(portfolio)
|
||||||
assert len(events) > 0
|
assert len(events) > 0
|
||||||
|
|
||||||
|
|
||||||
@ -81,7 +85,7 @@ def test_paginate_ws_audit_log():
|
|||||||
)
|
)
|
||||||
|
|
||||||
events = AuditLog.get_portfolio_events(
|
events = AuditLog.get_portfolio_events(
|
||||||
portfolio.owner, portfolio, pagination_opts={"per_page": 25, "page": 2}
|
portfolio, pagination_opts={"per_page": 25, "page": 2}
|
||||||
)
|
)
|
||||||
assert len(events) == 25
|
assert len(events) == 25
|
||||||
|
|
||||||
@ -94,7 +98,7 @@ def test_ws_audit_log_only_includes_current_ws_events():
|
|||||||
application_1 = ApplicationFactory.create(portfolio=portfolio)
|
application_1 = ApplicationFactory.create(portfolio=portfolio)
|
||||||
application_2 = ApplicationFactory.create(portfolio=other_portfolio)
|
application_2 = ApplicationFactory.create(portfolio=other_portfolio)
|
||||||
|
|
||||||
events = AuditLog.get_portfolio_events(portfolio.owner, portfolio)
|
events = AuditLog.get_portfolio_events(portfolio)
|
||||||
for event in events:
|
for event in events:
|
||||||
assert event.portfolio_id == portfolio.id or event.resource_id == portfolio.id
|
assert event.portfolio_id == portfolio.id or event.resource_id == portfolio.id
|
||||||
assert (
|
assert (
|
||||||
|
@ -29,9 +29,7 @@ def test_create_environment_role_creates_cloud_id(session):
|
|||||||
|
|
||||||
portfolio_role = portfolio.members[0]
|
portfolio_role = portfolio.members[0]
|
||||||
assert not portfolio_role.user.cloud_id
|
assert not portfolio_role.user.cloud_id
|
||||||
assert Environments.update_environment_roles(
|
assert Environments.update_environment_roles(portfolio_role, new_role)
|
||||||
owner, portfolio, portfolio_role, new_role
|
|
||||||
)
|
|
||||||
|
|
||||||
assert portfolio_role.user.cloud_id is not None
|
assert portfolio_role.user.cloud_id is not None
|
||||||
|
|
||||||
@ -69,9 +67,7 @@ def test_update_environment_roles():
|
|||||||
]
|
]
|
||||||
|
|
||||||
portfolio_role = portfolio.members[0]
|
portfolio_role = portfolio.members[0]
|
||||||
assert Environments.update_environment_roles(
|
assert Environments.update_environment_roles(portfolio_role, new_ids_and_roles)
|
||||||
owner, portfolio, portfolio_role, new_ids_and_roles
|
|
||||||
)
|
|
||||||
new_dev_env_role = EnvironmentRoles.get(portfolio_role.user.id, dev_env.id)
|
new_dev_env_role = EnvironmentRoles.get(portfolio_role.user.id, dev_env.id)
|
||||||
staging_env_role = EnvironmentRoles.get(portfolio_role.user.id, staging_env.id)
|
staging_env_role = EnvironmentRoles.get(portfolio_role.user.id, staging_env.id)
|
||||||
|
|
||||||
@ -120,9 +116,7 @@ def test_remove_environment_role():
|
|||||||
]
|
]
|
||||||
|
|
||||||
portfolio_role = PortfolioRoles.get(portfolio.id, developer.id)
|
portfolio_role = PortfolioRoles.get(portfolio.id, developer.id)
|
||||||
assert Environments.update_environment_roles(
|
assert Environments.update_environment_roles(portfolio_role, new_environment_roles)
|
||||||
owner, portfolio, portfolio_role, new_environment_roles
|
|
||||||
)
|
|
||||||
|
|
||||||
assert portfolio_role.num_environment_roles == 2
|
assert portfolio_role.num_environment_roles == 2
|
||||||
assert EnvironmentRoles.get(developer.id, now_ba).role == "billing_auditor"
|
assert EnvironmentRoles.get(developer.id, now_ba).role == "billing_auditor"
|
||||||
@ -154,9 +148,7 @@ def test_no_update_to_environment_roles():
|
|||||||
new_ids_and_roles = [{"id": dev_env.id, "role": "devops"}]
|
new_ids_and_roles = [{"id": dev_env.id, "role": "devops"}]
|
||||||
|
|
||||||
portfolio_role = PortfolioRoles.get(portfolio.id, developer.id)
|
portfolio_role = PortfolioRoles.get(portfolio.id, developer.id)
|
||||||
assert not Environments.update_environment_roles(
|
assert not Environments.update_environment_roles(portfolio_role, new_ids_and_roles)
|
||||||
owner, portfolio, portfolio_role, new_ids_and_roles
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_scoped_environments(db):
|
def test_get_scoped_environments(db):
|
||||||
|
@ -130,7 +130,7 @@ def test_resend_invitation():
|
|||||||
user = UserFactory.create()
|
user = UserFactory.create()
|
||||||
ws_role = PortfolioRoleFactory.create(user=user, portfolio=portfolio)
|
ws_role = PortfolioRoleFactory.create(user=user, portfolio=portfolio)
|
||||||
invite = Invitations.create(portfolio.owner, ws_role, user.email)
|
invite = Invitations.create(portfolio.owner, ws_role, user.email)
|
||||||
Invitations.resend(portfolio.owner, portfolio.id, invite.token)
|
Invitations.resend(user, invite.token)
|
||||||
assert ws_role.invitations[0].is_revoked
|
assert ws_role.invitations[0].is_revoked
|
||||||
assert ws_role.invitations[1].is_pending
|
assert ws_role.invitations[1].is_pending
|
||||||
|
|
||||||
|
@ -54,7 +54,7 @@ def test_portfolios_get_ensures_user_is_in_portfolio(portfolio, portfolio_owner)
|
|||||||
|
|
||||||
|
|
||||||
def test_get_for_update_applications_allows_owner(portfolio, portfolio_owner):
|
def test_get_for_update_applications_allows_owner(portfolio, portfolio_owner):
|
||||||
Portfolios.get_for_update_applications(portfolio_owner, portfolio.id)
|
Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skip(reason="redo as a route access test")
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
@ -63,7 +63,7 @@ def test_get_for_update_applications_blocks_developer(portfolio):
|
|||||||
PortfolioRoles.add(developer, portfolio.id)
|
PortfolioRoles.add(developer, portfolio.id)
|
||||||
|
|
||||||
with pytest.raises(UnauthorizedError):
|
with pytest.raises(UnauthorizedError):
|
||||||
Portfolios.get_for_update_applications(developer, portfolio.id)
|
Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
|
|
||||||
def test_can_create_portfolio_role(portfolio, portfolio_owner):
|
def test_can_create_portfolio_role(portfolio, portfolio_owner):
|
||||||
@ -75,7 +75,7 @@ def test_can_create_portfolio_role(portfolio, portfolio_owner):
|
|||||||
"dod_id": "1234567890",
|
"dod_id": "1234567890",
|
||||||
}
|
}
|
||||||
|
|
||||||
new_member = Portfolios.create_member(portfolio_owner, portfolio, user_data)
|
new_member = Portfolios.create_member(portfolio, user_data)
|
||||||
assert new_member.portfolio == portfolio
|
assert new_member.portfolio == portfolio
|
||||||
assert new_member.user.provisional
|
assert new_member.user.provisional
|
||||||
|
|
||||||
@ -90,7 +90,7 @@ def test_can_add_existing_user_to_portfolio(portfolio, portfolio_owner):
|
|||||||
"dod_id": user.dod_id,
|
"dod_id": user.dod_id,
|
||||||
}
|
}
|
||||||
|
|
||||||
new_member = Portfolios.create_member(portfolio_owner, portfolio, user_data)
|
new_member = Portfolios.create_member(portfolio, user_data)
|
||||||
assert new_member.portfolio == portfolio
|
assert new_member.portfolio == portfolio
|
||||||
assert new_member.user.email == user.email
|
assert new_member.user.email == user.email
|
||||||
assert not new_member.user.provisional
|
assert not new_member.user.provisional
|
||||||
@ -109,7 +109,7 @@ def test_need_permission_to_create_portfolio_role(portfolio, portfolio_owner):
|
|||||||
}
|
}
|
||||||
|
|
||||||
with pytest.raises(UnauthorizedError):
|
with pytest.raises(UnauthorizedError):
|
||||||
Portfolios.create_member(random_user, portfolio, user_data)
|
Portfolios.create_member(portfolio, user_data)
|
||||||
|
|
||||||
|
|
||||||
def test_update_portfolio_role_role(portfolio, portfolio_owner):
|
def test_update_portfolio_role_role(portfolio, portfolio_owner):
|
||||||
@ -124,9 +124,7 @@ def test_update_portfolio_role_role(portfolio, portfolio_owner):
|
|||||||
member = PortfolioRoleFactory.create(portfolio=portfolio)
|
member = PortfolioRoleFactory.create(portfolio=portfolio)
|
||||||
permission_sets = [PermissionSets.EDIT_PORTFOLIO_FUNDING]
|
permission_sets = [PermissionSets.EDIT_PORTFOLIO_FUNDING]
|
||||||
|
|
||||||
updated_member = Portfolios.update_member(
|
updated_member = Portfolios.update_member(member, permission_sets=permission_sets)
|
||||||
portfolio_owner, portfolio, member, permission_sets=permission_sets
|
|
||||||
)
|
|
||||||
assert updated_member.portfolio == portfolio
|
assert updated_member.portfolio == portfolio
|
||||||
|
|
||||||
|
|
||||||
@ -140,22 +138,22 @@ def test_need_permission_to_update_portfolio_role_role(portfolio, portfolio_owne
|
|||||||
"portfolio_role": "developer",
|
"portfolio_role": "developer",
|
||||||
"dod_id": "1234567890",
|
"dod_id": "1234567890",
|
||||||
}
|
}
|
||||||
member = Portfolios.create_member(portfolio_owner, portfolio, user_data)
|
member = Portfolios.create_member(portfolio, user_data)
|
||||||
role_name = "developer"
|
role_name = "developer"
|
||||||
|
|
||||||
with pytest.raises(UnauthorizedError):
|
with pytest.raises(UnauthorizedError):
|
||||||
Portfolios.update_member(random_user, portfolio, member, role_name)
|
Portfolios.update_member(member, role_name)
|
||||||
|
|
||||||
|
|
||||||
def test_owner_can_view_portfolio_members(portfolio, portfolio_owner):
|
def test_owner_can_view_portfolio_members(portfolio, portfolio_owner):
|
||||||
portfolio = Portfolios.get_with_members(portfolio_owner, portfolio.id)
|
portfolio = Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
assert portfolio
|
assert portfolio
|
||||||
|
|
||||||
|
|
||||||
def test_ccpo_can_view_portfolio_members(portfolio, portfolio_owner):
|
def test_ccpo_can_view_portfolio_members(portfolio, portfolio_owner):
|
||||||
ccpo = UserFactory.create_ccpo()
|
ccpo = UserFactory.create_ccpo()
|
||||||
assert Portfolios.get_with_members(ccpo, portfolio.id)
|
assert Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skip(reason="redo as a route access test")
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
@ -163,16 +161,12 @@ def test_random_user_cannot_view_portfolio_members(portfolio):
|
|||||||
developer = UserFactory.create()
|
developer = UserFactory.create()
|
||||||
|
|
||||||
with pytest.raises(UnauthorizedError):
|
with pytest.raises(UnauthorizedError):
|
||||||
portfolio = Portfolios.get_with_members(developer, portfolio.id)
|
portfolio = Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
|
|
||||||
def test_scoped_portfolio_for_admin_missing_view_apps_perms(portfolio_owner, portfolio):
|
def test_scoped_portfolio_for_admin_missing_view_apps_perms(portfolio_owner, portfolio):
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio_owner,
|
portfolio, "My Application 2", "My application 2", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"My Application 2",
|
|
||||||
"My application 2",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
restricted_admin = UserFactory.create()
|
restricted_admin = UserFactory.create()
|
||||||
PortfolioRoleFactory.create(
|
PortfolioRoleFactory.create(
|
||||||
@ -191,18 +185,10 @@ def test_scoped_portfolio_only_returns_a_users_applications_and_environments(
|
|||||||
portfolio, portfolio_owner
|
portfolio, portfolio_owner
|
||||||
):
|
):
|
||||||
new_application = Applications.create(
|
new_application = Applications.create(
|
||||||
portfolio_owner,
|
portfolio, "My Application", "My application", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"My Application",
|
|
||||||
"My application",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio_owner,
|
portfolio, "My Application 2", "My application 2", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"My Application 2",
|
|
||||||
"My application 2",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
developer = UserFactory.create()
|
developer = UserFactory.create()
|
||||||
dev_environment = Environments.add_member(
|
dev_environment = Environments.add_member(
|
||||||
@ -222,11 +208,7 @@ def test_scoped_portfolio_returns_all_applications_for_portfolio_admin(
|
|||||||
):
|
):
|
||||||
for _ in range(5):
|
for _ in range(5):
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio_owner,
|
portfolio, "My Application", "My application", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"My Application",
|
|
||||||
"My application",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
|
|
||||||
admin = UserFactory.create()
|
admin = UserFactory.create()
|
||||||
@ -245,11 +227,7 @@ def test_scoped_portfolio_returns_all_applications_for_portfolio_owner(
|
|||||||
):
|
):
|
||||||
for _ in range(5):
|
for _ in range(5):
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio_owner,
|
portfolio, "My Application", "My application", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"My Application",
|
|
||||||
"My application",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
|
|
||||||
scoped_portfolio = Portfolios.get(portfolio_owner, portfolio.id)
|
scoped_portfolio = Portfolios.get(portfolio_owner, portfolio.id)
|
||||||
@ -289,7 +267,7 @@ def test_for_user_returns_all_portfolios_for_ccpo(portfolio, portfolio_owner):
|
|||||||
|
|
||||||
@pytest.mark.skip(reason="redo as a route access test")
|
@pytest.mark.skip(reason="redo as a route access test")
|
||||||
def test_get_for_update_information(portfolio, portfolio_owner):
|
def test_get_for_update_information(portfolio, portfolio_owner):
|
||||||
owner_ws = Portfolios.get_for_update_information(portfolio_owner, portfolio.id)
|
owner_ws = Portfolios.get_for_update(portfolio.id)
|
||||||
assert portfolio == owner_ws
|
assert portfolio == owner_ws
|
||||||
|
|
||||||
admin = UserFactory.create()
|
admin = UserFactory.create()
|
||||||
@ -297,16 +275,16 @@ def test_get_for_update_information(portfolio, portfolio_owner):
|
|||||||
PortfolioRoleFactory.create(
|
PortfolioRoleFactory.create(
|
||||||
user=admin, portfolio=portfolio, permission_sets=perm_sets
|
user=admin, portfolio=portfolio, permission_sets=perm_sets
|
||||||
)
|
)
|
||||||
admin_ws = Portfolios.get_for_update_information(admin, portfolio.id)
|
admin_ws = Portfolios.get_for_update(portfolio.id)
|
||||||
assert portfolio == admin_ws
|
assert portfolio == admin_ws
|
||||||
|
|
||||||
# TODO: implement ccpo roles
|
# TODO: implement ccpo roles
|
||||||
# ccpo = UserFactory.create_ccpo()
|
# ccpo = UserFactory.create_ccpo()
|
||||||
# assert Portfolios.get_for_update_information(ccpo, portfolio.id)
|
# assert Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
developer = UserFactory.create()
|
developer = UserFactory.create()
|
||||||
with pytest.raises(UnauthorizedError):
|
with pytest.raises(UnauthorizedError):
|
||||||
Portfolios.get_for_update_information(developer, portfolio.id)
|
Portfolios.get_for_update(portfolio.id)
|
||||||
|
|
||||||
|
|
||||||
def test_can_create_portfolios_with_matching_names():
|
def test_can_create_portfolios_with_matching_names():
|
||||||
@ -320,7 +298,7 @@ def test_able_to_revoke_portfolio_access_for_active_member():
|
|||||||
portfolio_role = PortfolioRoleFactory.create(
|
portfolio_role = PortfolioRoleFactory.create(
|
||||||
portfolio=portfolio, status=PortfolioRoleStatus.ACTIVE
|
portfolio=portfolio, status=PortfolioRoleStatus.ACTIVE
|
||||||
)
|
)
|
||||||
Portfolios.revoke_access(portfolio.owner, portfolio.id, portfolio_role.id)
|
Portfolios.revoke_access(portfolio.id, portfolio_role.id)
|
||||||
assert Portfolios.for_user(portfolio_role.user) == []
|
assert Portfolios.for_user(portfolio_role.user) == []
|
||||||
|
|
||||||
|
|
||||||
@ -340,7 +318,7 @@ def test_unable_to_revoke_owner_portfolio_access():
|
|||||||
owner_portfolio_role = portfolio.roles[0]
|
owner_portfolio_role = portfolio.roles[0]
|
||||||
|
|
||||||
with pytest.raises(PortfolioError):
|
with pytest.raises(PortfolioError):
|
||||||
Portfolios.revoke_access(portfolio.owner, portfolio.id, owner_portfolio_role.id)
|
Portfolios.revoke_access(portfolio.id, owner_portfolio_role.id)
|
||||||
|
|
||||||
|
|
||||||
def test_disabled_members_dont_show_up(session):
|
def test_disabled_members_dont_show_up(session):
|
||||||
|
@ -21,7 +21,7 @@ def test_is_signed_by_ko():
|
|||||||
|
|
||||||
assert not TaskOrders.is_signed_by_ko(task_order)
|
assert not TaskOrders.is_signed_by_ko(task_order)
|
||||||
|
|
||||||
TaskOrders.update(user, task_order, signer_dod_id=user.dod_id)
|
TaskOrders.update(task_order, signer_dod_id=user.dod_id)
|
||||||
|
|
||||||
assert TaskOrders.is_signed_by_ko(task_order)
|
assert TaskOrders.is_signed_by_ko(task_order)
|
||||||
|
|
||||||
@ -68,7 +68,7 @@ def test_add_officer():
|
|||||||
task_order = TaskOrderFactory.create()
|
task_order = TaskOrderFactory.create()
|
||||||
ko = UserFactory.create()
|
ko = UserFactory.create()
|
||||||
owner = task_order.portfolio.owner
|
owner = task_order.portfolio.owner
|
||||||
TaskOrders.add_officer(owner, task_order, "contracting_officer", ko.to_dictionary())
|
TaskOrders.add_officer(task_order, "contracting_officer", ko.to_dictionary())
|
||||||
|
|
||||||
assert task_order.contracting_officer == ko
|
assert task_order.contracting_officer == ko
|
||||||
portfolio_users = [ws_role.user for ws_role in task_order.portfolio.members]
|
portfolio_users = [ws_role.user for ws_role in task_order.portfolio.members]
|
||||||
@ -80,15 +80,13 @@ def test_add_officer_with_nonexistent_role():
|
|||||||
ko = UserFactory.create()
|
ko = UserFactory.create()
|
||||||
owner = task_order.portfolio.owner
|
owner = task_order.portfolio.owner
|
||||||
with pytest.raises(TaskOrderError):
|
with pytest.raises(TaskOrderError):
|
||||||
TaskOrders.add_officer(owner, task_order, "pilot", ko.to_dictionary())
|
TaskOrders.add_officer(task_order, "pilot", ko.to_dictionary())
|
||||||
|
|
||||||
|
|
||||||
def test_add_officer_who_is_already_portfolio_member():
|
def test_add_officer_who_is_already_portfolio_member():
|
||||||
task_order = TaskOrderFactory.create()
|
task_order = TaskOrderFactory.create()
|
||||||
owner = task_order.portfolio.owner
|
owner = task_order.portfolio.owner
|
||||||
TaskOrders.add_officer(
|
TaskOrders.add_officer(task_order, "contracting_officer", owner.to_dictionary())
|
||||||
owner, task_order, "contracting_officer", owner.to_dictionary()
|
|
||||||
)
|
|
||||||
|
|
||||||
assert task_order.contracting_officer == owner
|
assert task_order.contracting_officer == owner
|
||||||
member = task_order.portfolio.members[0]
|
member = task_order.portfolio.members[0]
|
||||||
@ -122,9 +120,7 @@ def test_task_order_access():
|
|||||||
for prms in PortfolioRoles.DEFAULT_PORTFOLIO_PERMISSION_SETS
|
for prms in PortfolioRoles.DEFAULT_PORTFOLIO_PERMISSION_SETS
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
TaskOrders.add_officer(
|
TaskOrders.add_officer(task_order, "contracting_officer", officer.to_dictionary())
|
||||||
creator, task_order, "contracting_officer", officer.to_dictionary()
|
|
||||||
)
|
|
||||||
|
|
||||||
check_access([creator, officer, member], [rando], "get", [task_order.id])
|
check_access([creator, officer, member], [rando], "get", [task_order.id])
|
||||||
check_access([creator, officer], [member, rando], "create", [portfolio])
|
check_access([creator, officer], [member, rando], "create", [portfolio])
|
||||||
|
@ -9,11 +9,7 @@ def test_add_user_to_environment():
|
|||||||
|
|
||||||
portfolio = PortfolioFactory.create(owner=owner)
|
portfolio = PortfolioFactory.create(owner=owner)
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
owner,
|
portfolio, "my test application", "It's mine.", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"my test application",
|
|
||||||
"It's mine.",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
dev_environment = application.environments[0]
|
dev_environment = application.environments[0]
|
||||||
|
|
||||||
|
@ -120,7 +120,7 @@ def test_has_env_role_history(session):
|
|||||||
user=user, environment=environment, role="developer"
|
user=user, environment=environment, role="developer"
|
||||||
)
|
)
|
||||||
Environments.update_environment_roles(
|
Environments.update_environment_roles(
|
||||||
owner, portfolio, portfolio_role, [{"role": "admin", "id": environment.id}]
|
portfolio_role, [{"role": "admin", "id": environment.id}]
|
||||||
)
|
)
|
||||||
changed_events = (
|
changed_events = (
|
||||||
session.query(AuditEvent)
|
session.query(AuditEvent)
|
||||||
@ -154,7 +154,7 @@ def test_has_no_environment_roles():
|
|||||||
}
|
}
|
||||||
|
|
||||||
portfolio = PortfolioFactory.create(owner=owner)
|
portfolio = PortfolioFactory.create(owner=owner)
|
||||||
portfolio_role = Portfolios.create_member(owner, portfolio, developer_data)
|
portfolio_role = Portfolios.create_member(portfolio, developer_data)
|
||||||
|
|
||||||
assert not portfolio_role.has_environment_roles
|
assert not portfolio_role.has_environment_roles
|
||||||
|
|
||||||
@ -170,13 +170,9 @@ def test_has_environment_roles():
|
|||||||
}
|
}
|
||||||
|
|
||||||
portfolio = PortfolioFactory.create(owner=owner)
|
portfolio = PortfolioFactory.create(owner=owner)
|
||||||
portfolio_role = Portfolios.create_member(owner, portfolio, developer_data)
|
portfolio_role = Portfolios.create_member(portfolio, developer_data)
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
owner,
|
portfolio, "my test application", "It's mine.", ["dev", "staging", "prod"]
|
||||||
portfolio,
|
|
||||||
"my test application",
|
|
||||||
"It's mine.",
|
|
||||||
["dev", "staging", "prod"],
|
|
||||||
)
|
)
|
||||||
Environments.add_member(
|
Environments.add_member(
|
||||||
application.environments[0], portfolio_role.user, "developer"
|
application.environments[0], portfolio_role.user, "developer"
|
||||||
|
@ -130,7 +130,6 @@ def test_creating_application(client, user_session):
|
|||||||
def test_view_edit_application(client, user_session):
|
def test_view_edit_application(client, user_session):
|
||||||
portfolio = PortfolioFactory.create()
|
portfolio = PortfolioFactory.create()
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
portfolio.owner,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
"Snazzy Application",
|
"Snazzy Application",
|
||||||
"A new application for me and my friends",
|
"A new application for me and my friends",
|
||||||
|
@ -168,7 +168,6 @@ def test_update_member_environment_role(client, user_session):
|
|||||||
user = UserFactory.create()
|
user = UserFactory.create()
|
||||||
member = PortfolioRoles.add(user, portfolio.id)
|
member = PortfolioRoles.add(user, portfolio.id)
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
portfolio.owner,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
"Snazzy Application",
|
"Snazzy Application",
|
||||||
"A new application for me and my friends",
|
"A new application for me and my friends",
|
||||||
@ -202,7 +201,6 @@ def test_update_member_environment_role_with_no_data(client, user_session):
|
|||||||
user = UserFactory.create()
|
user = UserFactory.create()
|
||||||
member = PortfolioRoles.add(user, portfolio.id)
|
member = PortfolioRoles.add(user, portfolio.id)
|
||||||
application = Applications.create(
|
application = Applications.create(
|
||||||
portfolio.owner,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
"Snazzy Application",
|
"Snazzy Application",
|
||||||
"A new application for me and my friends",
|
"A new application for me and my friends",
|
||||||
@ -231,7 +229,6 @@ def test_revoke_active_member_access(client, user_session):
|
|||||||
portfolio=portfolio, user=user, status=PortfolioRoleStatus.ACTIVE
|
portfolio=portfolio, user=user, status=PortfolioRoleStatus.ACTIVE
|
||||||
)
|
)
|
||||||
Applications.create(
|
Applications.create(
|
||||||
portfolio.owner,
|
|
||||||
portfolio,
|
portfolio,
|
||||||
"Snazzy Application",
|
"Snazzy Application",
|
||||||
"A new application for me and my friends",
|
"A new application for me and my friends",
|
||||||
|
@ -157,7 +157,7 @@ class TestTaskOrderInvitations:
|
|||||||
"security_officer-last_name": "Fett",
|
"security_officer-last_name": "Fett",
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
updated_task_order = TaskOrders.get(self.portfolio.owner, self.task_order.id)
|
updated_task_order = TaskOrders.get(self.task_order.id)
|
||||||
assert updated_task_order.ko_first_name == "Luke"
|
assert updated_task_order.ko_first_name == "Luke"
|
||||||
assert updated_task_order.ko_last_name == "Skywalker"
|
assert updated_task_order.ko_last_name == "Skywalker"
|
||||||
assert updated_task_order.so_first_name == "Boba"
|
assert updated_task_order.so_first_name == "Boba"
|
||||||
@ -189,7 +189,7 @@ class TestTaskOrderInvitations:
|
|||||||
"contracting_officer-invite": "y",
|
"contracting_officer-invite": "y",
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
updated_task_order = TaskOrders.get(self.portfolio.owner, self.task_order.id)
|
updated_task_order = TaskOrders.get(self.task_order.id)
|
||||||
|
|
||||||
assert updated_task_order.ko_invite == True
|
assert updated_task_order.ko_invite == True
|
||||||
assert updated_task_order.ko_first_name == "Luke"
|
assert updated_task_order.ko_first_name == "Luke"
|
||||||
@ -222,7 +222,7 @@ class TestTaskOrderInvitations:
|
|||||||
|
|
||||||
assert "There were some errors" in response.data.decode()
|
assert "There were some errors" in response.data.decode()
|
||||||
|
|
||||||
updated_task_order = TaskOrders.get(self.portfolio.owner, self.task_order.id)
|
updated_task_order = TaskOrders.get(self.task_order.id)
|
||||||
assert updated_task_order.so_first_name != "Boba"
|
assert updated_task_order.so_first_name != "Boba"
|
||||||
assert len(queue.get_queue()) == queue_length
|
assert len(queue.get_queue()) == queue_length
|
||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
@ -251,7 +251,7 @@ def test_ko_can_view_task_order(client, user_session, portfolio, user):
|
|||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
assert translate("common.manage") in response.data.decode()
|
assert translate("common.manage") in response.data.decode()
|
||||||
|
|
||||||
TaskOrders.update(user, task_order, clin_01=None)
|
TaskOrders.update(task_order, clin_01=None)
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for(
|
url_for(
|
||||||
"portfolios.view_task_order",
|
"portfolios.view_task_order",
|
||||||
@ -706,20 +706,21 @@ def test_resending_revoked_invite(app, client, user_session, portfolio, user):
|
|||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
def test_resending_expired_invite(app, client, user_session, portfolio, user):
|
def test_resending_expired_invite(app, client, user_session, portfolio):
|
||||||
queue_length = len(queue.get_queue())
|
queue_length = len(queue.get_queue())
|
||||||
|
|
||||||
|
ko = UserFactory.create()
|
||||||
task_order = TaskOrderFactory.create(
|
task_order = TaskOrderFactory.create(
|
||||||
portfolio=portfolio, contracting_officer=user, ko_invite=True
|
portfolio=portfolio, contracting_officer=ko, ko_invite=True
|
||||||
)
|
)
|
||||||
portfolio_role = PortfolioRoleFactory.create(portfolio=portfolio, user=user)
|
portfolio_role = PortfolioRoleFactory.create(portfolio=portfolio, user=ko)
|
||||||
invite = InvitationFactory.create(
|
invite = InvitationFactory.create(
|
||||||
inviter=user,
|
inviter=portfolio.owner,
|
||||||
portfolio_role=portfolio_role,
|
portfolio_role=portfolio_role,
|
||||||
email=user.email,
|
email=ko.email,
|
||||||
expiration_time=datetime.now() - timedelta(days=1),
|
expiration_time=datetime.now() - timedelta(days=1),
|
||||||
)
|
)
|
||||||
user_session(user)
|
user_session(portfolio.owner)
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
url_for(
|
url_for(
|
||||||
|
@ -118,7 +118,7 @@ def test_create_new_task_order(client, user_session, pdf_upload):
|
|||||||
assert url_for("task_orders.new", screen=2) in response.headers["Location"]
|
assert url_for("task_orders.new", screen=2) in response.headers["Location"]
|
||||||
|
|
||||||
created_task_order_id = response.headers["Location"].split("/")[-1]
|
created_task_order_id = response.headers["Location"].split("/")[-1]
|
||||||
created_task_order = TaskOrders.get(creator, created_task_order_id)
|
created_task_order = TaskOrders.get(created_task_order_id)
|
||||||
assert created_task_order.portfolio is not None
|
assert created_task_order.portfolio is not None
|
||||||
assert created_task_order.portfolio.name == portfolio_name
|
assert created_task_order.portfolio.name == portfolio_name
|
||||||
assert created_task_order.portfolio.defense_component == defense_component
|
assert created_task_order.portfolio.defense_component == defense_component
|
||||||
@ -156,7 +156,7 @@ def test_create_new_task_order_for_portfolio(client, user_session):
|
|||||||
assert url_for("task_orders.new", screen=2) in response.headers["Location"]
|
assert url_for("task_orders.new", screen=2) in response.headers["Location"]
|
||||||
|
|
||||||
created_task_order_id = response.headers["Location"].split("/")[-1]
|
created_task_order_id = response.headers["Location"].split("/")[-1]
|
||||||
created_task_order = TaskOrders.get(creator, created_task_order_id)
|
created_task_order = TaskOrders.get(created_task_order_id)
|
||||||
assert created_task_order.portfolio_name == portfolio.name
|
assert created_task_order.portfolio_name == portfolio.name
|
||||||
assert created_task_order.defense_component == portfolio.defense_component
|
assert created_task_order.defense_component == portfolio.defense_component
|
||||||
assert created_task_order.portfolio == portfolio
|
assert created_task_order.portfolio == portfolio
|
||||||
@ -213,7 +213,7 @@ def test_review_screen_when_all_sections_complete(client, user_session, task_ord
|
|||||||
|
|
||||||
|
|
||||||
def test_review_screen_when_not_all_sections_complete(client, user_session, task_order):
|
def test_review_screen_when_not_all_sections_complete(client, user_session, task_order):
|
||||||
TaskOrders.update(task_order.creator, task_order, clin_01=None)
|
TaskOrders.update(task_order, clin_01=None)
|
||||||
user_session(task_order.creator)
|
user_session(task_order.creator)
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for("task_orders.new", screen=4, task_order_id=task_order.id)
|
url_for("task_orders.new", screen=4, task_order_id=task_order.id)
|
||||||
|
@ -18,10 +18,7 @@ def create_ko_task_order(user_session, contracting_officer):
|
|||||||
)
|
)
|
||||||
|
|
||||||
TaskOrders.add_officer(
|
TaskOrders.add_officer(
|
||||||
contracting_officer,
|
task_order, "contracting_officer", contracting_officer.to_dictionary()
|
||||||
task_order,
|
|
||||||
"contracting_officer",
|
|
||||||
contracting_officer.to_dictionary(),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
dd_254 = DD254Factory.create()
|
dd_254 = DD254Factory.create()
|
||||||
@ -33,7 +30,7 @@ def create_ko_task_order(user_session, contracting_officer):
|
|||||||
def test_show_signature_requested_not_ko(client, user_session):
|
def test_show_signature_requested_not_ko(client, user_session):
|
||||||
contracting_officer = UserFactory.create()
|
contracting_officer = UserFactory.create()
|
||||||
task_order = create_ko_task_order(user_session, contracting_officer)
|
task_order = create_ko_task_order(user_session, contracting_officer)
|
||||||
TaskOrders.update(contracting_officer, task_order, contracting_officer=None)
|
TaskOrders.update(task_order, contracting_officer=None)
|
||||||
|
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
||||||
@ -50,10 +47,7 @@ def test_show_signature_requested(client, user_session):
|
|||||||
# create unfinished TO
|
# create unfinished TO
|
||||||
task_order = TaskOrderFactory.create(portfolio=portfolio, clin_01=None)
|
task_order = TaskOrderFactory.create(portfolio=portfolio, clin_01=None)
|
||||||
TaskOrders.add_officer(
|
TaskOrders.add_officer(
|
||||||
contracting_officer,
|
task_order, "contracting_officer", contracting_officer.to_dictionary()
|
||||||
task_order,
|
|
||||||
"contracting_officer",
|
|
||||||
contracting_officer.to_dictionary(),
|
|
||||||
)
|
)
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
||||||
@ -61,7 +55,7 @@ def test_show_signature_requested(client, user_session):
|
|||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
# Finish TO
|
# Finish TO
|
||||||
TaskOrders.update(contracting_officer, task_order, clin_01=100)
|
TaskOrders.update(task_order, clin_01=100)
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
||||||
)
|
)
|
||||||
@ -79,9 +73,7 @@ def test_show_signature_requested(client, user_session):
|
|||||||
def test_show_signature_requested_already_signed(client, user_session):
|
def test_show_signature_requested_already_signed(client, user_session):
|
||||||
contracting_officer = UserFactory.create()
|
contracting_officer = UserFactory.create()
|
||||||
task_order = create_ko_task_order(user_session, contracting_officer)
|
task_order = create_ko_task_order(user_session, contracting_officer)
|
||||||
TaskOrders.update(
|
TaskOrders.update(task_order, signer_dod_id=contracting_officer.dod_id)
|
||||||
contracting_officer, task_order, signer_dod_id=contracting_officer.dod_id
|
|
||||||
)
|
|
||||||
|
|
||||||
response = client.get(
|
response = client.get(
|
||||||
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
url_for("task_orders.signature_requested", task_order_id=task_order.id)
|
||||||
@ -93,7 +85,7 @@ def test_show_signature_requested_already_signed(client, user_session):
|
|||||||
def test_signing_task_order_not_ko(client, user_session):
|
def test_signing_task_order_not_ko(client, user_session):
|
||||||
contracting_officer = UserFactory.create()
|
contracting_officer = UserFactory.create()
|
||||||
task_order = create_ko_task_order(user_session, contracting_officer)
|
task_order = create_ko_task_order(user_session, contracting_officer)
|
||||||
TaskOrders.update(contracting_officer, task_order, contracting_officer=None)
|
TaskOrders.update(task_order, contracting_officer=None)
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
url_for("task_orders.record_signature", task_order_id=task_order.id), data={}
|
url_for("task_orders.record_signature", task_order_id=task_order.id), data={}
|
||||||
@ -105,9 +97,7 @@ def test_signing_task_order_not_ko(client, user_session):
|
|||||||
def test_singing_an_already_signed_task_order(client, user_session):
|
def test_singing_an_already_signed_task_order(client, user_session):
|
||||||
contracting_officer = UserFactory.create()
|
contracting_officer = UserFactory.create()
|
||||||
task_order = create_ko_task_order(user_session, contracting_officer)
|
task_order = create_ko_task_order(user_session, contracting_officer)
|
||||||
TaskOrders.update(
|
TaskOrders.update(task_order, signer_dod_id=contracting_officer.dod_id)
|
||||||
contracting_officer, task_order, signer_dod_id=contracting_officer.dod_id
|
|
||||||
)
|
|
||||||
|
|
||||||
response = client.post(
|
response = client.post(
|
||||||
url_for("task_orders.record_signature", task_order_id=task_order.id),
|
url_for("task_orders.record_signature", task_order_id=task_order.id),
|
||||||
|
@ -8,18 +8,18 @@ import atst.domain.authz as authz
|
|||||||
from tests.factories import UserFactory
|
from tests.factories import UserFactory
|
||||||
|
|
||||||
_NO_ACCESS_CHECK_REQUIRED = _NO_LOGIN_REQUIRED + [
|
_NO_ACCESS_CHECK_REQUIRED = _NO_LOGIN_REQUIRED + [
|
||||||
"task_orders.get_started",
|
"task_orders.get_started", # all users can start a new TO
|
||||||
"atst.csp_environment_access",
|
"atst.csp_environment_access", # internal redirect
|
||||||
"atst.jedi_csp_calculator",
|
"atst.jedi_csp_calculator", # internal redirect
|
||||||
"atst.styleguide",
|
"atst.styleguide", # dev reference
|
||||||
"dev.test_email",
|
"dev.test_email", # dev tool
|
||||||
"dev.messages",
|
"dev.messages", # dev tool
|
||||||
"atst.home",
|
"atst.home", # available to all users
|
||||||
"users.user",
|
"users.user", # available to all users
|
||||||
"users.update_user",
|
"users.update_user", # available to all users
|
||||||
"portfolios.accept_invitation",
|
"portfolios.accept_invitation", # available to all users; access control is built into invitation logic
|
||||||
"atst.catch_all",
|
"atst.catch_all", # available to all users
|
||||||
"portfolios.portfolios",
|
"portfolios.portfolios", # the portfolios list is scoped to the user separately
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
@ -38,10 +38,19 @@ def protected_routes(app):
|
|||||||
return _protected_routes
|
return _protected_routes
|
||||||
|
|
||||||
|
|
||||||
_PROTECTED_ROUTES = protected_routes(make_app(make_config()))
|
sample_config = make_config()
|
||||||
|
sample_app = make_app(sample_config)
|
||||||
|
_PROTECTED_ROUTES = protected_routes(sample_app)
|
||||||
|
|
||||||
|
|
||||||
class Null:
|
class Null:
|
||||||
|
"""
|
||||||
|
Very simple null object. Will return itself for all attribute
|
||||||
|
calls:
|
||||||
|
> foo = Null()
|
||||||
|
> foo.bar.baz == foo
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args, **kwargs):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@ -53,10 +62,18 @@ class Null:
|
|||||||
def test_all_protected_routes_have_access_control(
|
def test_all_protected_routes_have_access_control(
|
||||||
rule, route, mocker, client, user_session, monkeypatch
|
rule, route, mocker, client, user_session, monkeypatch
|
||||||
):
|
):
|
||||||
|
"""
|
||||||
|
This tests that all routes, except the ones in
|
||||||
|
_NO_ACCESS_CHECK_REQUIRED, are protected by the access
|
||||||
|
decorator.
|
||||||
|
"""
|
||||||
|
# monkeypatch any object lookups that might happen in the access decorator
|
||||||
monkeypatch.setattr("atst.domain.portfolios.Portfolios.for_user", lambda *a: [])
|
monkeypatch.setattr("atst.domain.portfolios.Portfolios.for_user", lambda *a: [])
|
||||||
monkeypatch.setattr("atst.domain.portfolios.Portfolios.get", lambda *a: None)
|
monkeypatch.setattr("atst.domain.portfolios.Portfolios.get", lambda *a: None)
|
||||||
monkeypatch.setattr("atst.domain.task_orders.TaskOrders.get", lambda *a: Null())
|
monkeypatch.setattr("atst.domain.task_orders.TaskOrders.get", lambda *a: Null())
|
||||||
|
|
||||||
|
# patch the two internal functions the access decorator uses so
|
||||||
|
# that we can check that one or the other was called
|
||||||
mocker.patch("atst.domain.authz.decorator.user_can_access")
|
mocker.patch("atst.domain.authz.decorator.user_can_access")
|
||||||
mocker.patch("atst.domain.authz.decorator.evaluate_exceptions")
|
mocker.patch("atst.domain.authz.decorator.evaluate_exceptions")
|
||||||
|
|
Loading…
x
Reference in New Issue
Block a user