Merge pull request #455 from dod-ccpo/refactor-workspace-routes

Refactor workspace routes
This commit is contained in:
montana-mil 2018-11-28 15:19:13 -05:00 committed by GitHub
commit fd745c261b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 939 additions and 923 deletions

View File

@ -12,7 +12,7 @@ from atst.database import db
from atst.assets import environment as assets_environment from atst.assets import environment as assets_environment
from atst.filters import register_filters from atst.filters import register_filters
from atst.routes import bp from atst.routes import bp
from atst.routes.workspaces import bp as workspace_routes from atst.routes.workspaces import workspaces_bp as workspace_routes
from atst.routes.requests import requests_bp from atst.routes.requests import requests_bp
from atst.routes.dev import bp as dev_routes from atst.routes.dev import bp as dev_routes
from atst.routes.users import bp as user_routes from atst.routes.users import bp as user_routes

View File

@ -1,389 +0,0 @@
import re
from datetime import date, timedelta
from flask import (
Blueprint,
render_template,
request as http_request,
g,
redirect,
url_for,
)
from atst.domain.exceptions import UnauthorizedError, AlreadyExistsError
from atst.domain.projects import Projects
from atst.domain.reports import Reports
from atst.domain.workspaces import Workspaces
from atst.domain.workspace_roles import WorkspaceRoles, MEMBER_STATUSES
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
from atst.forms.project import NewProjectForm, ProjectForm
from atst.forms.new_member import NewMemberForm
from atst.forms.edit_member import EditMemberForm
from atst.forms.workspace import WorkspaceForm
from atst.forms.data import (
ENVIRONMENT_ROLES,
ENV_ROLE_MODAL_DESCRIPTION,
WORKSPACE_ROLE_DEFINITIONS,
)
from atst.domain.authz import Authorization
from atst.models.permissions import Permissions
from atst.domain.invitations import Invitations
from atst.queue import queue
bp = Blueprint("workspaces", __name__)
@bp.context_processor
def workspace():
workspaces = Workspaces.for_user(g.current_user)
workspace = None
if "workspace_id" in http_request.view_args:
try:
workspace = Workspaces.get(
g.current_user, http_request.view_args["workspace_id"]
)
workspaces = [ws for ws in workspaces if not ws.id == workspace.id]
except UnauthorizedError:
pass
def user_can(permission):
if workspace:
return Authorization.has_workspace_permission(
g.current_user, workspace, permission
)
return False
return {
"workspace": workspace,
"workspaces": workspaces,
"permissions": Permissions,
"user_can": user_can,
}
@bp.route("/workspaces")
def workspaces():
workspaces = Workspaces.for_user(g.current_user)
return render_template("workspaces/index.html", page=5, workspaces=workspaces)
@bp.route("/workspaces/<workspace_id>/edit")
def workspace(workspace_id):
workspace = Workspaces.get_for_update_information(g.current_user, workspace_id)
form = WorkspaceForm(data={"name": workspace.name})
return render_template("workspaces/edit.html", form=form, workspace=workspace)
@bp.route("/workspaces/<workspace_id>/projects")
def workspace_projects(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
return render_template("workspaces/projects/index.html", workspace=workspace)
@bp.route("/workspaces/<workspace_id>/edit", methods=["POST"])
def edit_workspace(workspace_id):
workspace = Workspaces.get_for_update_information(g.current_user, workspace_id)
form = WorkspaceForm(http_request.form)
if form.validate():
Workspaces.update(workspace, form.data)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template("workspaces/edit.html", form=form, workspace=workspace)
@bp.route("/workspaces/<workspace_id>")
def show_workspace(workspace_id):
return redirect(url_for("workspaces.workspace_projects", workspace_id=workspace_id))
@bp.route("/workspaces/<workspace_id>/members")
def workspace_members(workspace_id):
workspace = Workspaces.get_with_members(g.current_user, workspace_id)
new_member_name = http_request.args.get("newMemberName")
resent_invitation_to = http_request.args.get("resentInvitationTo")
new_member = next(
filter(lambda m: m.user_name == new_member_name, workspace.members), None
)
members_list = [
{
"name": k.user_name,
"status": k.display_status,
"id": k.user_id,
"role": k.role_displayname,
"num_env": k.num_environment_roles,
"edit_link": url_for(
"workspaces.view_member", workspace_id=workspace.id, member_id=k.user_id
),
}
for k in workspace.members
]
return render_template(
"workspaces/members/index.html",
workspace=workspace,
role_choices=WORKSPACE_ROLE_DEFINITIONS,
status_choices=MEMBER_STATUSES,
members=members_list,
new_member=new_member,
resent_invitation_to=resent_invitation_to,
)
@bp.route("/workspaces/<workspace_id>/reports")
def workspace_reports(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.VIEW_USAGE_DOLLARS,
"view workspace reports",
)
today = date.today()
month = http_request.args.get("month", today.month)
year = http_request.args.get("year", today.year)
current_month = date(int(year), int(month), 15)
prev_month = current_month - timedelta(days=28)
two_months_ago = prev_month - timedelta(days=28)
expiration_date = workspace.request.task_order.expiration_date
if expiration_date:
remaining_difference = expiration_date - today
remaining_days = remaining_difference.days
else:
remaining_days = None
return render_template(
"workspaces/reports/index.html",
cumulative_budget=Reports.cumulative_budget(workspace),
workspace_totals=Reports.workspace_totals(workspace),
monthly_totals=Reports.monthly_totals(workspace),
jedi_request=workspace.request,
task_order=workspace.request.task_order,
current_month=current_month,
prev_month=prev_month,
two_months_ago=two_months_ago,
expiration_date=expiration_date,
remaining_days=remaining_days,
)
@bp.route("/workspaces/<workspace_id>/projects/new")
def new_project(workspace_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
form = NewProjectForm()
return render_template(
"workspaces/projects/new.html", workspace=workspace, form=form
)
@bp.route("/workspaces/<workspace_id>/projects/new", methods=["POST"])
def create_project(workspace_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
form = NewProjectForm(http_request.form)
if form.validate():
project_data = form.data
Projects.create(
g.current_user,
workspace,
project_data["name"],
project_data["description"],
project_data["environment_names"],
)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template(
"workspaces/projects/new.html", workspace=workspace, form=form
)
@bp.route("/workspaces/<workspace_id>/projects/<project_id>/edit")
def edit_project(workspace_id, project_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
project = Projects.get(g.current_user, workspace, project_id)
form = ProjectForm(name=project.name, description=project.description)
return render_template(
"workspaces/projects/edit.html", workspace=workspace, project=project, form=form
)
@bp.route("/workspaces/<workspace_id>/projects/<project_id>/edit", methods=["POST"])
def update_project(workspace_id, project_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
project = Projects.get(g.current_user, workspace, project_id)
form = ProjectForm(http_request.form)
if form.validate():
project_data = form.data
Projects.update(g.current_user, workspace, project, project_data)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template(
"workspaces/projects/edit.html",
workspace=workspace,
project=project,
form=form,
)
@bp.route("/workspaces/<workspace_id>/members/new")
def new_member(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
form = NewMemberForm()
return render_template(
"workspaces/members/new.html", workspace=workspace, form=form
)
def send_invite_email(owner_name, token, new_member_email):
body = render_template("emails/invitation.txt", owner=owner_name, token=token)
queue.send_mail(
[new_member_email],
"{} has invited you to a JEDI Cloud Workspace".format(owner_name),
body,
)
@bp.route("/workspaces/<workspace_id>/members/new", methods=["POST"])
def create_member(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
form = NewMemberForm(http_request.form)
user = g.current_user
if form.validate():
try:
new_member = Workspaces.create_member(user, workspace, form.data)
invite = Invitations.create(user, new_member, form.data["email"])
send_invite_email(g.current_user.full_name, invite.token, invite.email)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace.id,
newMemberName=new_member.user_name,
)
)
except AlreadyExistsError:
return render_template(
"error.html", message="There was an error processing your request."
)
else:
return render_template(
"workspaces/members/new.html", workspace=workspace, form=form
)
@bp.route("/workspaces/<workspace_id>/members/<member_id>/member_edit")
def view_member(workspace_id, member_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.ASSIGN_AND_UNASSIGN_ATAT_ROLE,
"edit this workspace user",
)
member = WorkspaceRoles.get(workspace_id, member_id)
projects = Projects.get_all(g.current_user, member, workspace)
form = EditMemberForm(workspace_role=member.role_name)
editable = g.current_user == member.user
return render_template(
"workspaces/members/edit.html",
workspace=workspace,
member=member,
projects=projects,
form=form,
choices=ENVIRONMENT_ROLES,
env_role_modal_description=ENV_ROLE_MODAL_DESCRIPTION,
EnvironmentRoles=EnvironmentRoles,
editable=editable,
)
@bp.route(
"/workspaces/<workspace_id>/members/<member_id>/member_edit", methods=["POST"]
)
def update_member(workspace_id, member_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.ASSIGN_AND_UNASSIGN_ATAT_ROLE,
"edit this workspace user",
)
member = WorkspaceRoles.get(workspace_id, member_id)
ids_and_roles = []
form_dict = http_request.form.to_dict()
for entry in form_dict:
if re.match("env_", entry):
env_id = entry[4:]
env_role = form_dict[entry] or None
ids_and_roles.append({"id": env_id, "role": env_role})
form = EditMemberForm(http_request.form)
if form.validate():
new_role_name = None
if form.data["workspace_role"] != member.role:
member = Workspaces.update_member(
g.current_user, workspace, member, form.data["workspace_role"]
)
new_role_name = member.role_displayname
Environments.update_environment_roles(
g.current_user, workspace, member, ids_and_roles
)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace.id,
memberName=member.user_name,
updatedRole=new_role_name,
)
)
else:
return render_template(
"workspaces/members/edit.html",
form=form,
workspace=workspace,
member=member,
)
@bp.route("/workspaces/invitations/<token>", methods=["GET"])
def accept_invitation(token):
invite = Invitations.accept(g.current_user, token)
return redirect(
url_for("workspaces.show_workspace", workspace_id=invite.workspace.id)
)
@bp.route("/workspaces/<workspace_id>/invitations/<token>/revoke", methods=["POST"])
def revoke_invitation(workspace_id, token):
workspace = Workspaces.get_for_update_member(g.current_user, workspace_id)
Invitations.revoke(token)
return redirect(url_for("workspaces.workspace_members", workspace_id=workspace.id))
@bp.route("/workspaces/<workspace_id>/invitations/<token>/resend", methods=["POST"])
def resend_invitation(workspace_id, token):
invite = Invitations.resend(g.current_user, workspace_id, token)
send_invite_email(g.current_user.full_name, invite.token, invite.email)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace_id,
resentInvitationTo=invite.user_name,
)
)

View File

@ -0,0 +1,40 @@
from flask import Blueprint, request as http_request, g
workspaces_bp = Blueprint("workspaces", __name__)
from . import index
from . import projects
from . import members
from . import invitations
from atst.domain.exceptions import UnauthorizedError
from atst.domain.workspaces import Workspaces
from atst.domain.authz import Authorization
from atst.models.permissions import Permissions
@workspaces_bp.context_processor
def workspace():
workspaces = Workspaces.for_user(g.current_user)
workspace = None
if "workspace_id" in http_request.view_args:
try:
workspace = Workspaces.get(
g.current_user, http_request.view_args["workspace_id"]
)
workspaces = [ws for ws in workspaces if not ws.id == workspace.id]
except UnauthorizedError:
pass
def user_can(permission):
if workspace:
return Authorization.has_workspace_permission(
g.current_user, workspace, permission
)
return False
return {
"workspace": workspace,
"workspaces": workspaces,
"permissions": Permissions,
"user_can": user_can,
}

View File

@ -0,0 +1,80 @@
from datetime import date, timedelta
from flask import render_template, request as http_request, g, redirect, url_for
from . import workspaces_bp
from atst.domain.reports import Reports
from atst.domain.workspaces import Workspaces
from atst.forms.workspace import WorkspaceForm
from atst.domain.authz import Authorization
from atst.models.permissions import Permissions
@workspaces_bp.route("/workspaces")
def workspaces():
workspaces = Workspaces.for_user(g.current_user)
return render_template("workspaces/index.html", page=5, workspaces=workspaces)
@workspaces_bp.route("/workspaces/<workspace_id>/edit")
def workspace(workspace_id):
workspace = Workspaces.get_for_update_information(g.current_user, workspace_id)
form = WorkspaceForm(data={"name": workspace.name})
return render_template("workspaces/edit.html", form=form, workspace=workspace)
@workspaces_bp.route("/workspaces/<workspace_id>/edit", methods=["POST"])
def edit_workspace(workspace_id):
workspace = Workspaces.get_for_update_information(g.current_user, workspace_id)
form = WorkspaceForm(http_request.form)
if form.validate():
Workspaces.update(workspace, form.data)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template("workspaces/edit.html", form=form, workspace=workspace)
@workspaces_bp.route("/workspaces/<workspace_id>")
def show_workspace(workspace_id):
return redirect(url_for("workspaces.workspace_projects", workspace_id=workspace_id))
@workspaces_bp.route("/workspaces/<workspace_id>/reports")
def workspace_reports(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.VIEW_USAGE_DOLLARS,
"view workspace reports",
)
today = date.today()
month = http_request.args.get("month", today.month)
year = http_request.args.get("year", today.year)
current_month = date(int(year), int(month), 15)
prev_month = current_month - timedelta(days=28)
two_months_ago = prev_month - timedelta(days=28)
expiration_date = workspace.request.task_order.expiration_date
if expiration_date:
remaining_difference = expiration_date - today
remaining_days = remaining_difference.days
else:
remaining_days = None
return render_template(
"workspaces/reports/index.html",
cumulative_budget=Reports.cumulative_budget(workspace),
workspace_totals=Reports.workspace_totals(workspace),
monthly_totals=Reports.monthly_totals(workspace),
jedi_request=workspace.request,
task_order=workspace.request.task_order,
current_month=current_month,
prev_month=prev_month,
two_months_ago=two_months_ago,
expiration_date=expiration_date,
remaining_days=remaining_days,
)

View File

@ -0,0 +1,49 @@
from flask import g, redirect, url_for, render_template
from . import workspaces_bp
from atst.domain.workspaces import Workspaces
from atst.domain.invitations import Invitations
from atst.queue import queue
def send_invite_email(owner_name, token, new_member_email):
body = render_template("emails/invitation.txt", owner=owner_name, token=token)
queue.send_mail(
[new_member_email],
"{} has invited you to a JEDI Cloud Workspace".format(owner_name),
body,
)
@workspaces_bp.route("/workspaces/invitations/<token>", methods=["GET"])
def accept_invitation(token):
invite = Invitations.accept(g.current_user, token)
return redirect(
url_for("workspaces.show_workspace", workspace_id=invite.workspace.id)
)
@workspaces_bp.route(
"/workspaces/<workspace_id>/invitations/<token>/revoke", methods=["POST"]
)
def revoke_invitation(workspace_id, token):
workspace = Workspaces.get_for_update_member(g.current_user, workspace_id)
Invitations.revoke(token)
return redirect(url_for("workspaces.workspace_members", workspace_id=workspace.id))
@workspaces_bp.route(
"/workspaces/<workspace_id>/invitations/<token>/resend", methods=["POST"]
)
def resend_invitation(workspace_id, token):
invite = Invitations.resend(g.current_user, workspace_id, token)
send_invite_email(g.current_user.full_name, invite.token, invite.email)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace_id,
resentInvitationTo=invite.user_name,
)
)

View File

@ -0,0 +1,170 @@
import re
from flask import render_template, request as http_request, g, redirect, url_for
from . import workspaces_bp
from atst.routes.workspaces.invitations import send_invite_email
from atst.domain.exceptions import AlreadyExistsError
from atst.domain.projects import Projects
from atst.domain.workspaces import Workspaces
from atst.domain.workspace_roles import WorkspaceRoles, MEMBER_STATUSES
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
from atst.forms.new_member import NewMemberForm
from atst.forms.edit_member import EditMemberForm
from atst.forms.data import (
ENVIRONMENT_ROLES,
ENV_ROLE_MODAL_DESCRIPTION,
WORKSPACE_ROLE_DEFINITIONS,
)
from atst.domain.authz import Authorization
from atst.models.permissions import Permissions
from atst.domain.invitations import Invitations
@workspaces_bp.route("/workspaces/<workspace_id>/members")
def workspace_members(workspace_id):
workspace = Workspaces.get_with_members(g.current_user, workspace_id)
new_member_name = http_request.args.get("newMemberName")
resent_invitation_to = http_request.args.get("resentInvitationTo")
new_member = next(
filter(lambda m: m.user_name == new_member_name, workspace.members), None
)
members_list = [
{
"name": k.user_name,
"status": k.display_status,
"id": k.user_id,
"role": k.role_displayname,
"num_env": k.num_environment_roles,
"edit_link": url_for(
"workspaces.view_member", workspace_id=workspace.id, member_id=k.user_id
),
}
for k in workspace.members
]
return render_template(
"workspaces/members/index.html",
workspace=workspace,
role_choices=WORKSPACE_ROLE_DEFINITIONS,
status_choices=MEMBER_STATUSES,
members=members_list,
new_member=new_member,
resent_invitation_to=resent_invitation_to,
)
@workspaces_bp.route("/workspaces/<workspace_id>/members/new")
def new_member(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
form = NewMemberForm()
return render_template(
"workspaces/members/new.html", workspace=workspace, form=form
)
@workspaces_bp.route("/workspaces/<workspace_id>/members/new", methods=["POST"])
def create_member(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
form = NewMemberForm(http_request.form)
user = g.current_user
if form.validate():
try:
new_member = Workspaces.create_member(user, workspace, form.data)
invite = Invitations.create(user, new_member, form.data["email"])
send_invite_email(g.current_user.full_name, invite.token, invite.email)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace.id,
newMemberName=new_member.user_name,
)
)
except AlreadyExistsError:
return render_template(
"error.html", message="There was an error processing your request."
)
else:
return render_template(
"workspaces/members/new.html", workspace=workspace, form=form
)
@workspaces_bp.route("/workspaces/<workspace_id>/members/<member_id>/member_edit")
def view_member(workspace_id, member_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.ASSIGN_AND_UNASSIGN_ATAT_ROLE,
"edit this workspace user",
)
member = WorkspaceRoles.get(workspace_id, member_id)
projects = Projects.get_all(g.current_user, member, workspace)
form = EditMemberForm(workspace_role=member.role_name)
editable = g.current_user == member.user
return render_template(
"workspaces/members/edit.html",
workspace=workspace,
member=member,
projects=projects,
form=form,
choices=ENVIRONMENT_ROLES,
env_role_modal_description=ENV_ROLE_MODAL_DESCRIPTION,
EnvironmentRoles=EnvironmentRoles,
editable=editable,
)
@workspaces_bp.route(
"/workspaces/<workspace_id>/members/<member_id>/member_edit", methods=["POST"]
)
def update_member(workspace_id, member_id):
workspace = Workspaces.get(g.current_user, workspace_id)
Authorization.check_workspace_permission(
g.current_user,
workspace,
Permissions.ASSIGN_AND_UNASSIGN_ATAT_ROLE,
"edit this workspace user",
)
member = WorkspaceRoles.get(workspace_id, member_id)
ids_and_roles = []
form_dict = http_request.form.to_dict()
for entry in form_dict:
if re.match("env_", entry):
env_id = entry[4:]
env_role = form_dict[entry] or None
ids_and_roles.append({"id": env_id, "role": env_role})
form = EditMemberForm(http_request.form)
if form.validate():
new_role_name = None
if form.data["workspace_role"] != member.role:
member = Workspaces.update_member(
g.current_user, workspace, member, form.data["workspace_role"]
)
new_role_name = member.role_displayname
Environments.update_environment_roles(
g.current_user, workspace, member, ids_and_roles
)
return redirect(
url_for(
"workspaces.workspace_members",
workspace_id=workspace.id,
memberName=member.user_name,
updatedRole=new_role_name,
)
)
else:
return render_template(
"workspaces/members/edit.html",
form=form,
workspace=workspace,
member=member,
)

View File

@ -0,0 +1,78 @@
from flask import render_template, request as http_request, g, redirect, url_for
from . import workspaces_bp
from atst.domain.projects import Projects
from atst.domain.workspaces import Workspaces
from atst.forms.project import NewProjectForm, ProjectForm
@workspaces_bp.route("/workspaces/<workspace_id>/projects")
def workspace_projects(workspace_id):
workspace = Workspaces.get(g.current_user, workspace_id)
return render_template("workspaces/projects/index.html", workspace=workspace)
@workspaces_bp.route("/workspaces/<workspace_id>/projects/new")
def new_project(workspace_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
form = NewProjectForm()
return render_template(
"workspaces/projects/new.html", workspace=workspace, form=form
)
@workspaces_bp.route("/workspaces/<workspace_id>/projects/new", methods=["POST"])
def create_project(workspace_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
form = NewProjectForm(http_request.form)
if form.validate():
project_data = form.data
Projects.create(
g.current_user,
workspace,
project_data["name"],
project_data["description"],
project_data["environment_names"],
)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template(
"workspaces/projects/new.html", workspace=workspace, form=form
)
@workspaces_bp.route("/workspaces/<workspace_id>/projects/<project_id>/edit")
def edit_project(workspace_id, project_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
project = Projects.get(g.current_user, workspace, project_id)
form = ProjectForm(name=project.name, description=project.description)
return render_template(
"workspaces/projects/edit.html", workspace=workspace, project=project, form=form
)
@workspaces_bp.route(
"/workspaces/<workspace_id>/projects/<project_id>/edit", methods=["POST"]
)
def update_project(workspace_id, project_id):
workspace = Workspaces.get_for_update_projects(g.current_user, workspace_id)
project = Projects.get(g.current_user, workspace, project_id)
form = ProjectForm(http_request.form)
if form.validate():
project_data = form.data
Projects.update(g.current_user, workspace, project, project_data)
return redirect(
url_for("workspaces.workspace_projects", workspace_id=workspace.id)
)
else:
return render_template(
"workspaces/projects/edit.html",
workspace=workspace,
project=project,
form=form,
)

View File

@ -1,533 +0,0 @@
import datetime
from flask import url_for
import pytest
from tests.factories import (
UserFactory,
WorkspaceFactory,
WorkspaceRoleFactory,
InvitationFactory,
)
from atst.domain.workspaces import Workspaces
from atst.domain.workspace_roles import WorkspaceRoles
from atst.domain.projects import Projects
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
from atst.models.workspace_role import WorkspaceRole
from atst.models.workspace_role import Status as WorkspaceRoleStatus
from atst.models.invitation import Status as InvitationStatus
from atst.queue import queue
from atst.domain.users import Users
def test_user_with_permission_has_budget_report_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/reports"'.format(workspace.id).encode() in response.data
)
def test_user_without_permission_has_no_budget_report_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(
user, workspace, "developer", status=WorkspaceRoleStatus.ACTIVE
)
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/reports"'.format(workspace.id).encode()
not in response.data
)
def test_user_with_permission_has_add_project_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/projects/new"'.format(workspace.id).encode()
in response.data
)
def test_user_without_permission_has_no_add_project_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/projects/new"'.format(workspace.id).encode()
not in response.data
)
def test_user_with_permission_has_add_member_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/members".format(workspace.id))
assert (
'href="/workspaces/{}/members/new"'.format(workspace.id).encode()
in response.data
)
def test_user_without_permission_has_no_add_member_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
user_session(user)
response = client.get("/workspaces/{}/members".format(workspace.id))
assert (
'href="/workspaces/{}/members/new"'.format(workspace.id).encode()
not in response.data
)
def test_update_workspace_name(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.post(
url_for("workspaces.edit_workspace", workspace_id=workspace.id),
data={"name": "a cool new name"},
follow_redirects=True,
)
assert response.status_code == 200
assert workspace.name == "a cool new name"
def test_view_edit_project(client, user_session):
workspace = WorkspaceFactory.create()
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1", "env2"},
)
user_session(workspace.owner)
response = client.get(
"/workspaces/{}/projects/{}/edit".format(workspace.id, project.id)
)
assert response.status_code == 200
def test_user_with_permission_can_update_project(client, user_session):
owner = UserFactory.create()
workspace = WorkspaceFactory.create(
owner=owner,
projects=[
{
"name": "Awesome Project",
"description": "It's really awesome!",
"environments": [{"name": "dev"}, {"name": "prod"}],
}
],
)
project = workspace.projects[0]
user_session(owner)
response = client.post(
url_for(
"workspaces.update_project",
workspace_id=workspace.id,
project_id=project.id,
),
data={"name": "Really Cool Project", "description": "A very cool project."},
follow_redirects=True,
)
assert response.status_code == 200
assert project.name == "Really Cool Project"
assert project.description == "A very cool project."
def test_user_without_permission_cannot_update_project(client, user_session):
dev = UserFactory.create()
owner = UserFactory.create()
workspace = WorkspaceFactory.create(
owner=owner,
members=[{"user": dev, "role_name": "developer"}],
projects=[
{
"name": "Great Project",
"description": "Cool stuff happening here!",
"environments": [{"name": "dev"}, {"name": "prod"}],
}
],
)
project = workspace.projects[0]
user_session(dev)
response = client.post(
url_for(
"workspaces.update_project",
workspace_id=workspace.id,
project_id=project.id,
),
data={"name": "New Name", "description": "A new description."},
follow_redirects=True,
)
assert response.status_code == 404
assert project.name == "Great Project"
assert project.description == "Cool stuff happening here!"
def test_create_member(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
queue_length = len(queue.get_queue())
response = client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={
"dod_id": user.dod_id,
"first_name": "Wilbur",
"last_name": "Zuckerman",
"email": "some_pig@zuckermans.com",
"workspace_role": "developer",
},
follow_redirects=True,
)
assert response.status_code == 200
assert user.has_workspaces
assert user.invitations
assert len(queue.get_queue()) == queue_length + 1
def test_view_member_shows_role(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.get(
url_for("workspaces.view_member", workspace_id=workspace.id, member_id=user.id)
)
assert response.status_code == 200
assert "initial-choice='developer'".encode() in response.data
def test_permissions_for_view_member(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(user)
response = client.get(
url_for("workspaces.view_member", workspace_id=workspace.id, member_id=user.id)
)
assert response.status_code == 404
def test_update_member_workspace_role(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={"workspace_role": "security_auditor"},
follow_redirects=True,
)
assert response.status_code == 200
assert member.role_name == "security_auditor"
def test_update_member_workspace_role_with_no_data(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={},
follow_redirects=True,
)
assert response.status_code == 200
assert member.role_name == "developer"
def test_update_member_environment_role(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1", "env2"},
)
env1_id = project.environments[0].id
env2_id = project.environments[1].id
for env in project.environments:
Environments.add_member(env, user, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={
"workspace_role": "developer",
"env_" + str(env1_id): "security_auditor",
"env_" + str(env2_id): "devops",
},
follow_redirects=True,
)
assert response.status_code == 200
assert EnvironmentRoles.get(user.id, env1_id).role == "security_auditor"
assert EnvironmentRoles.get(user.id, env2_id).role == "devops"
def test_update_member_environment_role_with_no_data(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1"},
)
env1_id = project.environments[0].id
for env in project.environments:
Environments.add_member(env, user, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={"env_" + str(env1_id): None, "env_" + str(env1_id): ""},
follow_redirects=True,
)
assert response.status_code == 200
assert EnvironmentRoles.get(user.id, env1_id).role == "developer"
def test_existing_member_accepts_valid_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
workspace=workspace, user=user, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(user_id=user.id, workspace_role_id=ws_role.id)
# the user does not have access to the workspace before accepting the invite
assert len(Workspaces.for_user(user)) == 0
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
# user is redirected to the workspace view
assert response.status_code == 302
assert (
url_for("workspaces.show_workspace", workspace_id=invite.workspace.id)
in response.headers["Location"]
)
# the one-time use invite is no longer usable
assert invite.is_accepted
# the user has access to the workspace
assert len(Workspaces.for_user(user)) == 1
def test_existing_member_invite_sent_to_email_submitted_in_form(
client, user_session, queue
):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member_form_data = {
"dod_id": user.dod_id,
"first_name": user.first_name,
"last_name": user.last_name,
"workspace_role": "developer",
"email": "example@example.com",
}
user_session(workspace.owner)
client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={**member_form_data},
)
assert user.email != "example@example.com"
assert len(queue.get_queue().jobs[0].args[0]) == 1
assert queue.get_queue().jobs[0].args[0][0] == "example@example.com"
def test_new_member_accepts_valid_invite(monkeypatch, client, user_session):
workspace = WorkspaceFactory.create()
user_info = UserFactory.dictionary()
user_session(workspace.owner)
client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={"workspace_role": "developer", **user_info},
)
user = Users.get_by_dod_id(user_info["dod_id"])
token = user.invitations[0].token
monkeypatch.setattr(
"atst.domain.auth.should_redirect_to_user_profile", lambda *args: False
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=token))
# user is redirected to the workspace view
assert response.status_code == 302
assert (
url_for("workspaces.show_workspace", workspace_id=workspace.id)
in response.headers["Location"]
)
# the user has access to the workspace
assert len(Workspaces.for_user(user)) == 1
def test_member_accepts_invalid_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_WRONG_USER,
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_user_who_has_not_accepted_workspace_invite_cannot_view(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
# create user in workspace with invitation
user_session(workspace.owner)
response = client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={"workspace_role": "developer", **user.to_dictionary()},
)
# user tries to view workspace before accepting invitation
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert response.status_code == 404
def test_user_accepts_invite_with_wrong_dod_id(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
different_user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(user_id=user.id, workspace_role_id=ws_role.id)
user_session(different_user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_user_accepts_expired_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_EXPIRED,
expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1),
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_revoke_invitation(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_EXPIRED,
expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1),
)
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.revoke_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
assert response.status_code == 302
assert invite.is_revoked
def test_resend_invitation_sends_email(client, user_session, queue):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id, workspace_role_id=ws_role.id, status=InvitationStatus.PENDING
)
user_session(workspace.owner)
client.post(
url_for(
"workspaces.resend_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
assert len(queue.get_queue()) == 1
def test_existing_member_invite_resent_to_email_submitted_in_form(
client, user_session, queue
):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.PENDING,
email="example@example.com",
)
user_session(workspace.owner)
client.post(
url_for(
"workspaces.resend_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
send_mail_job = queue.get_queue().jobs[0]
assert user.email != "example@example.com"
assert send_mail_job.func.__func__.__name__ == "_send_mail"
assert send_mail_job.args[0] == ["example@example.com"]

View File

@ -0,0 +1,15 @@
from flask import url_for
from tests.factories import WorkspaceFactory
def test_update_workspace_name(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.post(
url_for("workspaces.edit_workspace", workspace_id=workspace.id),
data={"name": "a cool new name"},
follow_redirects=True,
)
assert response.status_code == 200
assert workspace.name == "a cool new name"

View File

@ -0,0 +1,209 @@
import datetime
from flask import url_for
from tests.factories import (
UserFactory,
WorkspaceFactory,
WorkspaceRoleFactory,
InvitationFactory,
)
from atst.domain.workspaces import Workspaces
from atst.models.workspace_role import Status as WorkspaceRoleStatus
from atst.models.invitation import Status as InvitationStatus
from atst.domain.users import Users
def test_existing_member_accepts_valid_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
workspace=workspace, user=user, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(user_id=user.id, workspace_role_id=ws_role.id)
# the user does not have access to the workspace before accepting the invite
assert len(Workspaces.for_user(user)) == 0
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
# user is redirected to the workspace view
assert response.status_code == 302
assert (
url_for("workspaces.show_workspace", workspace_id=invite.workspace.id)
in response.headers["Location"]
)
# the one-time use invite is no longer usable
assert invite.is_accepted
# the user has access to the workspace
assert len(Workspaces.for_user(user)) == 1
def test_new_member_accepts_valid_invite(monkeypatch, client, user_session):
workspace = WorkspaceFactory.create()
user_info = UserFactory.dictionary()
user_session(workspace.owner)
client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={"workspace_role": "developer", **user_info},
)
user = Users.get_by_dod_id(user_info["dod_id"])
token = user.invitations[0].token
monkeypatch.setattr(
"atst.domain.auth.should_redirect_to_user_profile", lambda *args: False
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=token))
# user is redirected to the workspace view
assert response.status_code == 302
assert (
url_for("workspaces.show_workspace", workspace_id=workspace.id)
in response.headers["Location"]
)
# the user has access to the workspace
assert len(Workspaces.for_user(user)) == 1
def test_member_accepts_invalid_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_WRONG_USER,
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_user_who_has_not_accepted_workspace_invite_cannot_view(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
# create user in workspace with invitation
user_session(workspace.owner)
response = client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={"workspace_role": "developer", **user.to_dictionary()},
)
# user tries to view workspace before accepting invitation
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert response.status_code == 404
def test_user_accepts_invite_with_wrong_dod_id(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
different_user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(user_id=user.id, workspace_role_id=ws_role.id)
user_session(different_user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_user_accepts_expired_invite(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_EXPIRED,
expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1),
)
user_session(user)
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
assert response.status_code == 404
def test_revoke_invitation(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.REJECTED_EXPIRED,
expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1),
)
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.revoke_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
assert response.status_code == 302
assert invite.is_revoked
def test_resend_invitation_sends_email(client, user_session, queue):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id, workspace_role_id=ws_role.id, status=InvitationStatus.PENDING
)
user_session(workspace.owner)
client.post(
url_for(
"workspaces.resend_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
assert len(queue.get_queue()) == 1
def test_existing_member_invite_resent_to_email_submitted_in_form(
client, user_session, queue
):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
ws_role = WorkspaceRoleFactory.create(
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
)
invite = InvitationFactory.create(
user_id=user.id,
workspace_role_id=ws_role.id,
status=InvitationStatus.PENDING,
email="example@example.com",
)
user_session(workspace.owner)
client.post(
url_for(
"workspaces.resend_invitation",
workspace_id=workspace.id,
token=invite.token,
)
)
send_mail_job = queue.get_queue().jobs[0]
assert user.email != "example@example.com"
assert send_mail_job.func.__func__.__name__ == "_send_mail"
assert send_mail_job.args[0] == ["example@example.com"]

View File

@ -0,0 +1,170 @@
from flask import url_for
from tests.factories import UserFactory, WorkspaceFactory
from atst.domain.workspaces import Workspaces
from atst.domain.workspace_roles import WorkspaceRoles
from atst.domain.projects import Projects
from atst.domain.environments import Environments
from atst.domain.environment_roles import EnvironmentRoles
from atst.queue import queue
def test_user_with_permission_has_add_member_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/members".format(workspace.id))
assert (
'href="/workspaces/{}/members/new"'.format(workspace.id).encode()
in response.data
)
def test_user_without_permission_has_no_add_member_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
user_session(user)
response = client.get("/workspaces/{}/members".format(workspace.id))
assert (
'href="/workspaces/{}/members/new"'.format(workspace.id).encode()
not in response.data
)
def test_permissions_for_view_member(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(user)
response = client.get(
url_for("workspaces.view_member", workspace_id=workspace.id, member_id=user.id)
)
assert response.status_code == 404
def test_create_member(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
queue_length = len(queue.get_queue())
response = client.post(
url_for("workspaces.create_member", workspace_id=workspace.id),
data={
"dod_id": user.dod_id,
"first_name": "Wilbur",
"last_name": "Zuckerman",
"email": "some_pig@zuckermans.com",
"workspace_role": "developer",
},
follow_redirects=True,
)
assert response.status_code == 200
assert user.has_workspaces
assert user.invitations
assert len(queue.get_queue()) == queue_length + 1
def test_view_member_shows_role(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.get(
url_for("workspaces.view_member", workspace_id=workspace.id, member_id=user.id)
)
assert response.status_code == 200
assert "initial-choice='developer'".encode() in response.data
def test_update_member_workspace_role(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={"workspace_role": "security_auditor"},
follow_redirects=True,
)
assert response.status_code == 200
assert member.role_name == "security_auditor"
def test_update_member_workspace_role_with_no_data(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={},
follow_redirects=True,
)
assert response.status_code == 200
assert member.role_name == "developer"
def test_update_member_environment_role(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1", "env2"},
)
env1_id = project.environments[0].id
env2_id = project.environments[1].id
for env in project.environments:
Environments.add_member(env, user, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={
"workspace_role": "developer",
"env_" + str(env1_id): "security_auditor",
"env_" + str(env2_id): "devops",
},
follow_redirects=True,
)
assert response.status_code == 200
assert EnvironmentRoles.get(user.id, env1_id).role == "security_auditor"
assert EnvironmentRoles.get(user.id, env2_id).role == "devops"
def test_update_member_environment_role_with_no_data(client, user_session):
workspace = WorkspaceFactory.create()
user = UserFactory.create()
member = WorkspaceRoles.add(user, workspace.id, "developer")
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1"},
)
env1_id = project.environments[0].id
for env in project.environments:
Environments.add_member(env, user, "developer")
user_session(workspace.owner)
response = client.post(
url_for(
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
),
data={"env_" + str(env1_id): None, "env_" + str(env1_id): ""},
follow_redirects=True,
)
assert response.status_code == 200
assert EnvironmentRoles.get(user.id, env1_id).role == "developer"

View File

@ -0,0 +1,127 @@
from flask import url_for
from tests.factories import UserFactory, WorkspaceFactory
from atst.domain.projects import Projects
from atst.domain.workspaces import Workspaces
from atst.models.workspace_role import Status as WorkspaceRoleStatus
def test_user_with_permission_has_budget_report_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/reports"'.format(workspace.id).encode() in response.data
)
def test_user_without_permission_has_no_budget_report_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(
user, workspace, "developer", status=WorkspaceRoleStatus.ACTIVE
)
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/reports"'.format(workspace.id).encode()
not in response.data
)
def test_user_with_permission_has_add_project_link(client, user_session):
workspace = WorkspaceFactory.create()
user_session(workspace.owner)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/projects/new"'.format(workspace.id).encode()
in response.data
)
def test_user_without_permission_has_no_add_project_link(client, user_session):
user = UserFactory.create()
workspace = WorkspaceFactory.create()
Workspaces._create_workspace_role(user, workspace, "developer")
user_session(user)
response = client.get("/workspaces/{}/projects".format(workspace.id))
assert (
'href="/workspaces/{}/projects/new"'.format(workspace.id).encode()
not in response.data
)
def test_view_edit_project(client, user_session):
workspace = WorkspaceFactory.create()
project = Projects.create(
workspace.owner,
workspace,
"Snazzy Project",
"A new project for me and my friends",
{"env1", "env2"},
)
user_session(workspace.owner)
response = client.get(
"/workspaces/{}/projects/{}/edit".format(workspace.id, project.id)
)
assert response.status_code == 200
def test_user_with_permission_can_update_project(client, user_session):
owner = UserFactory.create()
workspace = WorkspaceFactory.create(
owner=owner,
projects=[
{
"name": "Awesome Project",
"description": "It's really awesome!",
"environments": [{"name": "dev"}, {"name": "prod"}],
}
],
)
project = workspace.projects[0]
user_session(owner)
response = client.post(
url_for(
"workspaces.update_project",
workspace_id=workspace.id,
project_id=project.id,
),
data={"name": "Really Cool Project", "description": "A very cool project."},
follow_redirects=True,
)
assert response.status_code == 200
assert project.name == "Really Cool Project"
assert project.description == "A very cool project."
def test_user_without_permission_cannot_update_project(client, user_session):
dev = UserFactory.create()
owner = UserFactory.create()
workspace = WorkspaceFactory.create(
owner=owner,
members=[{"user": dev, "role_name": "developer"}],
projects=[
{
"name": "Great Project",
"description": "Cool stuff happening here!",
"environments": [{"name": "dev"}, {"name": "prod"}],
}
],
)
project = workspace.projects[0]
user_session(dev)
response = client.post(
url_for(
"workspaces.update_project",
workspace_id=workspace.id,
project_id=project.id,
),
data={"name": "New Name", "description": "A new description."},
follow_redirects=True,
)
assert response.status_code == 404
assert project.name == "Great Project"
assert project.description == "Cool stuff happening here!"