Merge pull request #802 from dod-ccpo/accept-application-invite

Accept application invite
This commit is contained in:
dandds 2019-05-06 14:30:54 -04:00 committed by GitHub
commit b0600a34db
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 177 additions and 50 deletions

View File

@ -1,5 +1,5 @@
from atst.database import db from atst.database import db
from atst.models.application_role import ApplicationRole from atst.models import ApplicationRole, ApplicationRoleStatus
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
@ -21,3 +21,10 @@ class ApplicationRoles(object):
db.session.commit() db.session.commit()
return application_role return application_role
@classmethod
def enable(cls, role):
role.status = ApplicationRoleStatus.ACTIVE
db.session.add(role)
db.session.commit()

View File

@ -6,9 +6,7 @@ from atst.domain.application_roles import ApplicationRoles
from atst.domain.environments import Environments from atst.domain.environments import Environments
from atst.domain.exceptions import NotFoundError from atst.domain.exceptions import NotFoundError
from atst.domain.users import Users from atst.domain.users import Users
from atst.models.application import Application from atst.models import Application, ApplicationRole, ApplicationRoleStatus
from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole
class Applications(BaseDomainClass): class Applications(BaseDomainClass):
@ -31,10 +29,11 @@ class Applications(BaseDomainClass):
def for_user(self, user, portfolio): def for_user(self, user, portfolio):
return ( return (
db.session.query(Application) db.session.query(Application)
.join(Environment) .join(ApplicationRole)
.join(EnvironmentRole)
.filter(Application.portfolio_id == portfolio.id) .filter(Application.portfolio_id == portfolio.id)
.filter(EnvironmentRole.user_id == user.id) .filter(ApplicationRole.application_id == Application.id)
.filter(ApplicationRole.user_id == user.id)
.filter(ApplicationRole.status == ApplicationRoleStatus.ACTIVE)
.all() .all()
) )

View File

@ -1,7 +1,6 @@
from atst.domain.authz import Authorization from atst.domain.authz import Authorization
from atst.models.permissions import Permissions from atst.models.permissions import Permissions
from atst.domain.applications import Applications from atst.domain.applications import Applications
from atst.domain.environments import Environments
class ScopedResource(object): class ScopedResource(object):
@ -35,31 +34,6 @@ class ScopedPortfolio(ScopedResource):
) )
if can_view_all_applications: if can_view_all_applications:
applications = self.resource.applications return self.resource.applications
else: else:
applications = Applications.for_user(self.user, self.resource) return Applications.for_user(self.user, self.resource)
return [
ScopedApplication(self.user, application) for application in applications
]
class ScopedApplication(ScopedResource):
"""
An object that obeys the same API as a Portfolio, but with the added
functionality that it only returns sub-resources (environments)
that the given user is allowed to see.
"""
@property
def environments(self):
can_view_all_environments = Authorization.has_portfolio_permission(
self.user, self.resource.portfolio, Permissions.VIEW_ENVIRONMENT
)
if can_view_all_environments:
environments = self.resource.environments
else:
environments = Environments.for_user(self.user, self.resource)
return environments

View File

@ -62,7 +62,13 @@ class ApplicationRole(
@property @property
def history(self): def history(self):
return self.get_changes() previous_state = self.get_changes()
change_set = {}
if "status" in previous_state:
from_status = previous_state["status"][0].value
to_status = self.status.value
change_set["status"] = [from_status, to_status]
return change_set
def has_permission_set(self, perm_set_name): def has_permission_set(self, perm_set_name):
return first_or_none( return first_or_none(

View File

@ -6,6 +6,7 @@ from . import index
from . import new from . import new
from . import settings from . import settings
from . import team from . import team
from . import invitations
from atst.domain.environment_roles import EnvironmentRoles from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.exceptions import UnauthorizedError from atst.domain.exceptions import UnauthorizedError
from atst.domain.authz.decorator import user_can_access_decorator as user_can from atst.domain.authz.decorator import user_can_access_decorator as user_can

View File

@ -5,7 +5,21 @@ from atst.domain.authz.decorator import user_can_access_decorator as user_can
from atst.models.permissions import Permissions from atst.models.permissions import Permissions
def has_portfolio_applications(_user, portfolio=None, **_kwargs):
"""
If the portfolio exists and the user has access to applications
within the scoped portfolio, the user has access to the
portfolio landing page.
"""
if portfolio and portfolio.applications:
return True
@applications_bp.route("/portfolios/<portfolio_id>/applications") @applications_bp.route("/portfolios/<portfolio_id>/applications")
@user_can(Permissions.VIEW_APPLICATION, message="view portfolio applications") @user_can(
Permissions.VIEW_APPLICATION,
override=has_portfolio_applications,
message="view portfolio applications",
)
def portfolio_applications(portfolio_id): def portfolio_applications(portfolio_id):
return render_template("portfolios/applications/index.html") return render_template("portfolios/applications/index.html")

View File

@ -0,0 +1,16 @@
from flask import redirect, url_for, g
from . import applications_bp
from atst.domain.invitations import ApplicationInvitations
@applications_bp.route("/applications/invitations/<token>", methods=["GET"])
def accept_invitation(token):
invite = ApplicationInvitations.accept(g.current_user, token)
return redirect(
url_for(
"applications.portfolio_applications",
portfolio_id=invite.application.portfolio_id,
)
)

View File

@ -5,6 +5,6 @@
Join this JEDI Cloud Application Join this JEDI Cloud Application
{{ owner }} has invited you to join a JEDI Cloud Application. Login now to view or use your JEDI Cloud resources. {{ owner }} has invited you to join a JEDI Cloud Application. Login now to view or use your JEDI Cloud resources.
{# url_for("application.accept_invitation", token=token, _external=True) #} {{ url_for("applications.accept_invitation", token=token, _external=True) }}
{% endblock %} {% endblock %}

View File

@ -1,6 +1,8 @@
from atst.domain.application_roles import ApplicationRoles from atst.domain.application_roles import ApplicationRoles
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
from tests.factories import UserFactory, ApplicationFactory from atst.models import ApplicationRoleStatus
from tests.factories import *
def test_create_application_role(): def test_create_application_role():
@ -18,3 +20,16 @@ def test_create_application_role():
) )
assert application_role.application == application assert application_role.application == application
assert application_role.user == user assert application_role.user == user
def test_enabled_application_role():
application = ApplicationFactory.create()
user = UserFactory.create()
app_role = ApplicationRoleFactory.create(
application=application, user=user, status=ApplicationRoleStatus.DISABLED
)
assert app_role.status == ApplicationRoleStatus.DISABLED
ApplicationRoles.enable(app_role)
assert app_role.status == ApplicationRoleStatus.ACTIVE

View File

@ -1,7 +1,7 @@
import pytest import pytest
from uuid import uuid4 from uuid import uuid4
from atst.models import CSPRole from atst.models import CSPRole, ApplicationRoleStatus
from atst.domain.applications import Applications from atst.domain.applications import Applications
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
from atst.domain.exceptions import NotFoundError from atst.domain.exceptions import NotFoundError
@ -128,3 +128,30 @@ def test_create_member():
env_roles = member_role.user.environment_roles env_roles = member_role.user.environment_roles
assert len(env_roles) == 1 assert len(env_roles) == 1
assert env_roles[0].environment == env1 assert env_roles[0].environment == env1
def test_for_user():
user = UserFactory.create()
portfolio = PortfolioFactory.create()
for _x in range(4):
ApplicationFactory.create(portfolio=portfolio)
ApplicationRoleFactory.create(
application=portfolio.applications[0],
user=user,
status=ApplicationRoleStatus.ACTIVE,
)
ApplicationRoleFactory.create(
application=portfolio.applications[1],
user=user,
status=ApplicationRoleStatus.ACTIVE,
)
ApplicationRoleFactory.create(
application=portfolio.applications[2],
user=user,
status=ApplicationRoleStatus.PENDING,
)
assert len(portfolio.applications) == 4
user_applications = Applications.for_user(user, portfolio)
assert len(user_applications) == 2

View File

@ -1,13 +1,6 @@
from flask import url_for, get_flashed_messages from flask import url_for, get_flashed_messages
from tests.factories import ( from tests.factories import *
UserFactory,
PortfolioFactory,
PortfolioRoleFactory,
EnvironmentRoleFactory,
EnvironmentFactory,
ApplicationFactory,
)
from atst.domain.applications import Applications from atst.domain.applications import Applications
from atst.domain.portfolios import Portfolios from atst.domain.portfolios import Portfolios
@ -68,3 +61,31 @@ def test_user_without_permission_has_no_add_application_link(client, user_sessio
url_for("applications.create", portfolio_id=portfolio.id) url_for("applications.create", portfolio_id=portfolio.id)
not in response.data.decode() not in response.data.decode()
) )
def test_portfolio_applications_user_with_application_roles(client, user_session):
user = UserFactory.create()
portfolio = PortfolioFactory.create()
app1 = ApplicationFactory.create(portfolio=portfolio, name="X-Wing")
app2 = ApplicationFactory.create(portfolio=portfolio, name="TIE Fighter")
app3 = ApplicationFactory.create(portfolio=portfolio, name="Millenium Falcon")
ApplicationRoleFactory.create(
application=app1, user=user, status=ApplicationRoleStatus.ACTIVE
)
ApplicationRoleFactory.create(
application=app2, user=user, status=ApplicationRoleStatus.ACTIVE
)
user_session(user)
response = client.get(
url_for("applications.portfolio_applications", portfolio_id=portfolio.id)
)
assert response.status_code == 200
body = response.data.decode()
assert app1.name in body
assert app2.name in body
assert app3.name not in body

View File

@ -0,0 +1,41 @@
from flask import url_for
from tests.factories import *
def test_accept_application_invitation(client, user_session):
user = UserFactory.create()
application = ApplicationFactory.create()
app_role = ApplicationRoleFactory.create(application=application, user=user)
invite = ApplicationInvitationFactory.create(
role=app_role, user=user, inviter=application.portfolio.owner
)
user_session(user)
response = client.get(url_for("applications.accept_invitation", token=invite.token))
assert response.status_code == 302
expected_location = url_for(
"applications.portfolio_applications",
portfolio_id=application.portfolio_id,
_external=True,
)
assert response.location == expected_location
def test_accept_application_invitation_end_to_end(client, user_session):
user = UserFactory.create()
application = ApplicationFactory.create(name="Millenium Falcon")
app_role = ApplicationRoleFactory.create(application=application, user=user)
invite = ApplicationInvitationFactory.create(
role=app_role, user=user, inviter=application.portfolio.owner
)
user_session(user)
response = client.get(
url_for("applications.accept_invitation", token=invite.token),
follow_redirects=True,
)
assert response.status_code == 200
assert application.name in response.data.decode()

View File

@ -8,8 +8,7 @@ import atst
from atst.app import make_app, make_config from atst.app import make_app, make_config
from atst.domain.auth import UNPROTECTED_ROUTES as _NO_LOGIN_REQUIRED from atst.domain.auth import UNPROTECTED_ROUTES as _NO_LOGIN_REQUIRED
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
from atst.models.environment_role import CSPRole from atst.models import CSPRole, PortfolioRoleStatus, ApplicationRoleStatus
from atst.models.portfolio_role import Status as PortfolioRoleStatus
from tests.factories import ( from tests.factories import (
AttachmentFactory, AttachmentFactory,
@ -35,6 +34,7 @@ _NO_ACCESS_CHECK_REQUIRED = _NO_LOGIN_REQUIRED + [
"users.user", # available to all users "users.user", # available to all users
"users.update_user", # available to all users "users.update_user", # available to all users
"portfolios.accept_invitation", # available to all users; access control is built into invitation logic "portfolios.accept_invitation", # available to all users; access control is built into invitation logic
"applications.accept_invitation", # available to all users; access control is built into invitation logic
"atst.catch_all", # available to all users "atst.catch_all", # available to all users
"portfolios.portfolios", # the portfolios list is scoped to the user separately "portfolios.portfolios", # the portfolios list is scoped to the user separately
] ]
@ -360,12 +360,18 @@ def test_portfolios_admin_access(get_url_assert_status):
def test_applications_portfolio_applications_access(get_url_assert_status): def test_applications_portfolio_applications_access(get_url_assert_status):
ccpo = user_with(PermissionSets.VIEW_PORTFOLIO_APPLICATION_MANAGEMENT) ccpo = user_with(PermissionSets.VIEW_PORTFOLIO_APPLICATION_MANAGEMENT)
owner = user_with() owner = user_with()
app_user = user_with()
rando = user_with() rando = user_with()
portfolio = PortfolioFactory.create(owner=owner) portfolio = PortfolioFactory.create(owner=owner)
application = ApplicationFactory.create(portfolio=portfolio)
ApplicationRoleFactory.create(
application=application, user=app_user, status=ApplicationRoleStatus.ACTIVE
)
url = url_for("applications.portfolio_applications", portfolio_id=portfolio.id) url = url_for("applications.portfolio_applications", portfolio_id=portfolio.id)
get_url_assert_status(ccpo, url, 200) get_url_assert_status(ccpo, url, 200)
get_url_assert_status(owner, url, 200) get_url_assert_status(owner, url, 200)
get_url_assert_status(app_user, url, 200)
get_url_assert_status(rando, url, 404) get_url_assert_status(rando, url, 404)