diff --git a/atst/app.py b/atst/app.py index b309806a..88002c4f 100644 --- a/atst/app.py +++ b/atst/app.py @@ -12,7 +12,7 @@ from atst.database import db from atst.assets import environment as assets_environment from atst.filters import register_filters 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.dev import bp as dev_routes from atst.routes.users import bp as user_routes diff --git a/atst/routes/workspaces.py b/atst/routes/workspaces.py deleted file mode 100644 index c0a0cab6..00000000 --- a/atst/routes/workspaces.py +++ /dev/null @@ -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//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//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//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/") -def show_workspace(workspace_id): - return redirect(url_for("workspaces.workspace_projects", workspace_id=workspace_id)) - - -@bp.route("/workspaces//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//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//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//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//projects//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//projects//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//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//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//members//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//members//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/", 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//invitations//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//invitations//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, - ) - ) diff --git a/atst/routes/workspaces/__init__.py b/atst/routes/workspaces/__init__.py new file mode 100644 index 00000000..16e1f746 --- /dev/null +++ b/atst/routes/workspaces/__init__.py @@ -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, + } diff --git a/atst/routes/workspaces/index.py b/atst/routes/workspaces/index.py new file mode 100644 index 00000000..a92ead98 --- /dev/null +++ b/atst/routes/workspaces/index.py @@ -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//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//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/") +def show_workspace(workspace_id): + return redirect(url_for("workspaces.workspace_projects", workspace_id=workspace_id)) + + +@workspaces_bp.route("/workspaces//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, + ) diff --git a/atst/routes/workspaces/invitations.py b/atst/routes/workspaces/invitations.py new file mode 100644 index 00000000..8e609f88 --- /dev/null +++ b/atst/routes/workspaces/invitations.py @@ -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/", 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//invitations//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//invitations//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, + ) + ) diff --git a/atst/routes/workspaces/members.py b/atst/routes/workspaces/members.py new file mode 100644 index 00000000..6507329f --- /dev/null +++ b/atst/routes/workspaces/members.py @@ -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//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//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//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//members//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//members//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, + ) diff --git a/atst/routes/workspaces/projects.py b/atst/routes/workspaces/projects.py new file mode 100644 index 00000000..cc6f10f8 --- /dev/null +++ b/atst/routes/workspaces/projects.py @@ -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//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//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//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//projects//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//projects//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, + ) diff --git a/tests/routes/test_workspaces.py b/tests/routes/test_workspaces.py deleted file mode 100644 index c15ee2c5..00000000 --- a/tests/routes/test_workspaces.py +++ /dev/null @@ -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"] diff --git a/tests/routes/workspaces/test_index.py b/tests/routes/workspaces/test_index.py new file mode 100644 index 00000000..62c1a5cd --- /dev/null +++ b/tests/routes/workspaces/test_index.py @@ -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" diff --git a/tests/routes/workspaces/test_invitations.py b/tests/routes/workspaces/test_invitations.py new file mode 100644 index 00000000..c981cb72 --- /dev/null +++ b/tests/routes/workspaces/test_invitations.py @@ -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"] diff --git a/tests/routes/workspaces/test_members.py b/tests/routes/workspaces/test_members.py new file mode 100644 index 00000000..1816ecad --- /dev/null +++ b/tests/routes/workspaces/test_members.py @@ -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" diff --git a/tests/routes/workspaces/test_projects.py b/tests/routes/workspaces/test_projects.py new file mode 100644 index 00000000..44ee822d --- /dev/null +++ b/tests/routes/workspaces/test_projects.py @@ -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!"