Updates to workspace role permissions, invitations
This commit is contained in:
parent
4255dbe292
commit
848bbf9c12
@ -0,0 +1,42 @@
|
||||
"""change invitation relationship to workspace role
|
||||
|
||||
Revision ID: d1ea7f3ee4be
|
||||
Revises: 5284ac1ac77c
|
||||
Create Date: 2018-10-30 14:09:42.277467
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = 'd1ea7f3ee4be'
|
||||
down_revision = '5284ac1ac77c'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.add_column('invitations', sa.Column('workspace_role_id', postgresql.UUID(as_uuid=True), nullable=True))
|
||||
op.create_index(op.f('ix_invitations_workspace_role_id'), 'invitations', ['workspace_role_id'], unique=False)
|
||||
op.drop_index('ix_invitations_token', table_name='invitations')
|
||||
op.create_index(op.f('ix_invitations_token'), 'invitations', ['token'], unique=False)
|
||||
op.drop_index('ix_invitations_workspace_id', table_name='invitations')
|
||||
op.drop_constraint('invitations_workspace_id_fkey', 'invitations', type_='foreignkey')
|
||||
op.create_foreign_key(None, 'invitations', 'workspace_roles', ['workspace_role_id'], ['id'])
|
||||
op.drop_column('invitations', 'workspace_id')
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade():
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.add_column('invitations', sa.Column('workspace_id', postgresql.UUID(), autoincrement=False, nullable=True))
|
||||
op.drop_constraint(None, 'invitations', type_='foreignkey')
|
||||
op.create_foreign_key('invitations_workspace_id_fkey', 'invitations', 'workspaces', ['workspace_id'], ['id'])
|
||||
op.create_index('ix_invitations_workspace_id', 'invitations', ['workspace_id'], unique=False)
|
||||
op.drop_index(op.f('ix_invitations_token'), table_name='invitations')
|
||||
op.create_index('ix_invitations_token', 'invitations', ['token'], unique=True)
|
||||
op.drop_index(op.f('ix_invitations_workspace_role_id'), table_name='invitations')
|
||||
op.drop_column('invitations', 'workspace_role_id')
|
||||
# ### end Alembic commands ###
|
@ -0,0 +1,28 @@
|
||||
"""add status to workspace_roles
|
||||
|
||||
Revision ID: e0fc3cd315c1
|
||||
Revises: d1ea7f3ee4be
|
||||
Create Date: 2018-10-30 14:36:51.047876
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = 'e0fc3cd315c1'
|
||||
down_revision = 'd1ea7f3ee4be'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.add_column('workspace_roles', sa.Column('status', sa.Enum('ACTIVE', 'DISABLED', 'PENDING', name='status', native_enum=False), nullable=True))
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade():
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.drop_column('workspace_roles', 'status')
|
||||
# ### end Alembic commands ###
|
@ -3,6 +3,7 @@ from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from atst.database import db
|
||||
from atst.models.invitation import Invitation, Status as InvitationStatus
|
||||
from atst.domain.workspace_users import WorkspaceUsers
|
||||
|
||||
from .exceptions import NotFoundError
|
||||
|
||||
@ -30,9 +31,9 @@ class Invitations(object):
|
||||
return invite
|
||||
|
||||
@classmethod
|
||||
def create(cls, workspace, inviter, user):
|
||||
def create(cls, workspace_role, inviter, user):
|
||||
invite = Invitation(
|
||||
workspace=workspace,
|
||||
workspace_role=workspace_role,
|
||||
inviter=inviter,
|
||||
user=user,
|
||||
status=InvitationStatus.PENDING,
|
||||
@ -43,20 +44,6 @@ class Invitations(object):
|
||||
|
||||
return invite
|
||||
|
||||
@classmethod
|
||||
def create_for_owner(cls, workspace, user):
|
||||
invite = Invitation(
|
||||
workspace=workspace,
|
||||
inviter=user,
|
||||
user=user,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
expiration_time=Invitations.current_expiration_time(),
|
||||
)
|
||||
db.session.add(invite)
|
||||
db.session.commit()
|
||||
|
||||
return invite
|
||||
|
||||
@classmethod
|
||||
def accept(cls, token):
|
||||
invite = Invitations._get(token)
|
||||
@ -72,6 +59,8 @@ class Invitations(object):
|
||||
if invite.is_revoked or invite.is_rejected:
|
||||
raise InvitationError(invite)
|
||||
|
||||
WorkspaceUsers.enable(invite.workspace_role)
|
||||
|
||||
return invite
|
||||
|
||||
@classmethod
|
||||
|
@ -1,10 +1,9 @@
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from atst.database import db
|
||||
from atst.models.workspace_role import WorkspaceRole
|
||||
from atst.models.workspace_role import WorkspaceRole, Status as WorkspaceRoleStatus
|
||||
from atst.models.workspace_user import WorkspaceUser
|
||||
from atst.models.user import User
|
||||
from atst.models.invitation import Invitation, Status as InvitationStatus
|
||||
|
||||
from .roles import Roles
|
||||
from .users import Users
|
||||
@ -38,9 +37,7 @@ class WorkspaceUsers(object):
|
||||
db.session.query(WorkspaceRole)
|
||||
.join(User)
|
||||
.filter(User.id == user_id, WorkspaceRole.workspace_id == workspace_id)
|
||||
.join(Invitation, WorkspaceRole.workspace_id == Invitation.workspace_id)
|
||||
.filter(Invitation.user_id == WorkspaceRole.user_id)
|
||||
.filter(Invitation.status == InvitationStatus.ACCEPTED)
|
||||
.filter(WorkspaceRole.status == WorkspaceRoleStatus.ACTIVE)
|
||||
.one()
|
||||
)
|
||||
except NoResultFound:
|
||||
@ -150,3 +147,10 @@ class WorkspaceUsers(object):
|
||||
db.session.commit()
|
||||
|
||||
return workspace_users
|
||||
|
||||
@classmethod
|
||||
def enable(cls, workspace_role):
|
||||
workspace_role.status = WorkspaceRoleStatus.ACTIVE
|
||||
|
||||
db.session.add(workspace_role)
|
||||
db.session.commit()
|
||||
|
@ -4,8 +4,7 @@ from atst.database import db
|
||||
from atst.domain.common import Query
|
||||
from atst.domain.exceptions import NotFoundError
|
||||
from atst.models.workspace import Workspace
|
||||
from atst.models.workspace_role import WorkspaceRole
|
||||
from atst.models.invitation import Invitation, Status as InvitationStatus
|
||||
from atst.models.workspace_role import WorkspaceRole, Status as WorkspaceRoleStatus
|
||||
|
||||
|
||||
class WorkspacesQuery(Query):
|
||||
@ -25,10 +24,8 @@ class WorkspacesQuery(Query):
|
||||
return (
|
||||
db.session.query(Workspace)
|
||||
.join(WorkspaceRole)
|
||||
.join(Invitation)
|
||||
.filter(WorkspaceRole.user == user)
|
||||
.filter(Invitation.user == user)
|
||||
.filter(Invitation.status == InvitationStatus.ACCEPTED)
|
||||
.filter(WorkspaceRole.status == WorkspaceRoleStatus.ACTIVE)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
@ -3,7 +3,7 @@ from atst.domain.authz import Authorization
|
||||
from atst.models.permissions import Permissions
|
||||
from atst.domain.users import Users
|
||||
from atst.domain.workspace_users import WorkspaceUsers
|
||||
from atst.domain.invitations import Invitations
|
||||
from atst.models.workspace_role import Status as WorkspaceRoleStatus
|
||||
|
||||
from .query import WorkspacesQuery
|
||||
from .scopes import ScopedWorkspace
|
||||
@ -14,8 +14,9 @@ class Workspaces(object):
|
||||
def create(cls, request, name=None):
|
||||
name = name or request.displayname
|
||||
workspace = WorkspacesQuery.create(request=request, name=name)
|
||||
Workspaces._create_workspace_role(request.creator, workspace, "owner")
|
||||
Invitations.create_for_owner(workspace, request.creator)
|
||||
Workspaces._create_workspace_role(
|
||||
request.creator, workspace, "owner", status=WorkspaceRoleStatus.ACTIVE
|
||||
)
|
||||
WorkspacesQuery.add_and_commit(workspace)
|
||||
return workspace
|
||||
|
||||
@ -109,9 +110,13 @@ class Workspaces(object):
|
||||
return WorkspaceUsers.update_role(member, workspace.id, role_name)
|
||||
|
||||
@classmethod
|
||||
def _create_workspace_role(cls, user, workspace, role_name):
|
||||
def _create_workspace_role(
|
||||
cls, user, workspace, role_name, status=WorkspaceRoleStatus.PENDING
|
||||
):
|
||||
role = Roles.get(role_name)
|
||||
workspace_role = WorkspacesQuery.create_workspace_role(user, role, workspace)
|
||||
workspace_role = WorkspacesQuery.create_workspace_role(
|
||||
user, role, workspace, status=status
|
||||
)
|
||||
WorkspacesQuery.add_and_commit(workspace_role)
|
||||
return workspace_role
|
||||
|
||||
|
@ -25,8 +25,10 @@ class Invitation(Base, TimestampsMixin):
|
||||
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), index=True)
|
||||
user = relationship("User", backref="invitations", foreign_keys=[user_id])
|
||||
|
||||
workspace_id = Column(UUID(as_uuid=True), ForeignKey("workspaces.id"), index=True)
|
||||
workspace = relationship("Workspace", backref="invitations")
|
||||
workspace_role_id = Column(
|
||||
UUID(as_uuid=True), ForeignKey("workspace_roles.id"), index=True
|
||||
)
|
||||
workspace_role = relationship("WorkspaceRole", backref="invitations")
|
||||
|
||||
inviter_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), index=True)
|
||||
inviter = relationship("User", backref="sent_invites", foreign_keys=[inviter_id])
|
||||
@ -61,3 +63,8 @@ class Invitation(Base, TimestampsMixin):
|
||||
@property
|
||||
def is_expired(self):
|
||||
return datetime.datetime.now(self.expiration_time.tzinfo) > self.expiration_time
|
||||
|
||||
@property
|
||||
def workspace(self):
|
||||
if self.workspace_role:
|
||||
return self.workspace_role.workspace
|
||||
|
@ -1,4 +1,5 @@
|
||||
from sqlalchemy import Index, ForeignKey, Column
|
||||
from enum import Enum
|
||||
from sqlalchemy import Index, ForeignKey, Column, Enum as SQLAEnum
|
||||
from sqlalchemy.dialects.postgresql import UUID
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
@ -6,6 +7,12 @@ from atst.models import Base, mixins
|
||||
from .types import Id
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
ACTIVE = "active"
|
||||
DISABLED = "disabled"
|
||||
PENDING = "pending"
|
||||
|
||||
|
||||
class WorkspaceRole(Base, mixins.TimestampsMixin, mixins.AuditableMixin):
|
||||
__tablename__ = "workspace_roles"
|
||||
|
||||
@ -22,6 +29,8 @@ class WorkspaceRole(Base, mixins.TimestampsMixin, mixins.AuditableMixin):
|
||||
UUID(as_uuid=True), ForeignKey("users.id"), index=True, nullable=False
|
||||
)
|
||||
|
||||
status = Column(SQLAEnum(Status, native_enum=False, default=Status.PENDING))
|
||||
|
||||
def __repr__(self):
|
||||
return "<WorkspaceRole(role='{}', workspace='{}', user_id='{}', id='{}')>".format(
|
||||
self.role.name, self.workspace.name, self.user_id, self.id
|
||||
|
@ -237,7 +237,9 @@ def create_member(workspace_id):
|
||||
if form.validate():
|
||||
try:
|
||||
new_member = Workspaces.create_member(g.current_user, workspace, form.data)
|
||||
invite = Invitations.create(workspace, g.current_user, new_member.user)
|
||||
invite = Invitations.create(
|
||||
new_member.workspace_role, g.current_user, new_member.user
|
||||
)
|
||||
send_invite_email(
|
||||
g.current_user.full_name, invite.token, new_member.user.email
|
||||
)
|
||||
|
@ -5,15 +5,21 @@ import re
|
||||
from atst.domain.invitations import Invitations, InvitationError
|
||||
from atst.models.invitation import Status
|
||||
|
||||
from tests.factories import WorkspaceFactory, UserFactory, InvitationFactory
|
||||
from tests.factories import (
|
||||
WorkspaceFactory,
|
||||
WorkspaceRoleFactory,
|
||||
UserFactory,
|
||||
InvitationFactory,
|
||||
)
|
||||
|
||||
|
||||
def test_create_invitation():
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
invite = Invitations.create(workspace, workspace.owner, user)
|
||||
ws_role = WorkspaceRoleFactory.create(user=user, workspace=workspace)
|
||||
invite = Invitations.create(ws_role, workspace.owner, user)
|
||||
assert invite.user == user
|
||||
assert invite.workspace == workspace
|
||||
assert invite.workspace_role == ws_role
|
||||
assert invite.inviter == workspace.owner
|
||||
assert invite.status == Status.PENDING
|
||||
assert re.match(r"^[\w\-_]+$", invite.token)
|
||||
@ -22,22 +28,19 @@ def test_create_invitation():
|
||||
def test_accept_invitation():
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
invite = Invitations.create(workspace, workspace.owner, user)
|
||||
ws_role = WorkspaceRoleFactory.create(user=user, workspace=workspace)
|
||||
invite = Invitations.create(ws_role, workspace.owner, user)
|
||||
assert invite.is_pending
|
||||
accepted_invite = Invitations.accept(invite.token)
|
||||
assert accepted_invite.is_accepted
|
||||
|
||||
|
||||
def test_accept_expired_invitation():
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
increment = Invitations.EXPIRATION_LIMIT_MINUTES + 1
|
||||
expiration_time = datetime.datetime.now() - datetime.timedelta(minutes=increment)
|
||||
invite = InvitationFactory.create(
|
||||
workspace_id=workspace.id,
|
||||
user_id=user.id,
|
||||
expiration_time=expiration_time,
|
||||
status=Status.PENDING,
|
||||
user_id=user.id, expiration_time=expiration_time, status=Status.PENDING
|
||||
)
|
||||
with pytest.raises(InvitationError):
|
||||
Invitations.accept(invite.token)
|
||||
@ -46,20 +49,14 @@ def test_accept_expired_invitation():
|
||||
|
||||
|
||||
def test_accept_rejected_invite():
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
invite = InvitationFactory.create(
|
||||
workspace_id=workspace.id, user_id=user.id, status=Status.REJECTED
|
||||
)
|
||||
invite = InvitationFactory.create(user_id=user.id, status=Status.REJECTED)
|
||||
with pytest.raises(InvitationError):
|
||||
Invitations.accept(invite.token)
|
||||
|
||||
|
||||
def test_accept_revoked_invite():
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
invite = InvitationFactory.create(
|
||||
workspace_id=workspace.id, user_id=user.id, status=Status.REVOKED
|
||||
)
|
||||
invite = InvitationFactory.create(user_id=user.id, status=Status.REVOKED)
|
||||
with pytest.raises(InvitationError):
|
||||
Invitations.accept(invite.token)
|
||||
|
@ -1,8 +1,14 @@
|
||||
from atst.domain.workspace_users import WorkspaceUsers
|
||||
from atst.domain.users import Users
|
||||
from atst.models.invitation import Status as InvitationStatus
|
||||
from atst.models.workspace_role import Status as WorkspaceRoleStatus
|
||||
from atst.domain.roles import Roles
|
||||
|
||||
from tests.factories import WorkspaceFactory, UserFactory, InvitationFactory
|
||||
from tests.factories import (
|
||||
WorkspaceFactory,
|
||||
UserFactory,
|
||||
InvitationFactory,
|
||||
WorkspaceRoleFactory,
|
||||
)
|
||||
|
||||
|
||||
def test_can_create_new_workspace_user():
|
||||
@ -42,17 +48,17 @@ def test_workspace_user_permissions():
|
||||
workspace_one = WorkspaceFactory.create()
|
||||
workspace_two = WorkspaceFactory.create()
|
||||
new_user = UserFactory.create()
|
||||
WorkspaceUsers.add_many(
|
||||
workspace_one.id, [{"id": new_user.id, "workspace_role": "developer"}]
|
||||
)
|
||||
WorkspaceUsers.add_many(
|
||||
workspace_two.id, [{"id": new_user.id, "workspace_role": "developer"}]
|
||||
)
|
||||
InvitationFactory.create(
|
||||
WorkspaceRoleFactory.create(
|
||||
workspace=workspace_one,
|
||||
user=new_user,
|
||||
inviter=workspace_one.owner,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
role=Roles.get("developer"),
|
||||
status=WorkspaceRoleStatus.ACTIVE,
|
||||
)
|
||||
WorkspaceRoleFactory.create(
|
||||
workspace=workspace_two,
|
||||
user=new_user,
|
||||
role=Roles.get("developer"),
|
||||
status=WorkspaceRoleStatus.PENDING,
|
||||
)
|
||||
|
||||
assert WorkspaceUsers.workspace_user_permissions(workspace_one, new_user)
|
||||
|
@ -6,9 +6,14 @@ from atst.domain.workspaces import Workspaces
|
||||
from atst.domain.workspace_users import WorkspaceUsers
|
||||
from atst.domain.projects import Projects
|
||||
from atst.domain.environments import Environments
|
||||
from atst.models.invitation import Status as InvitationStatus
|
||||
from atst.models.workspace_role import Status as WorkspaceRoleStatus
|
||||
|
||||
from tests.factories import RequestFactory, UserFactory, InvitationFactory
|
||||
from tests.factories import (
|
||||
RequestFactory,
|
||||
UserFactory,
|
||||
InvitationFactory,
|
||||
WorkspaceRoleFactory,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
@ -217,14 +222,9 @@ def test_scoped_workspace_returns_all_projects_for_workspace_admin(
|
||||
["dev", "staging", "prod"],
|
||||
)
|
||||
|
||||
admin = Workspaces.add_member(
|
||||
workspace, UserFactory.from_atat_role("default"), "admin"
|
||||
).user
|
||||
InvitationFactory.create(
|
||||
user=admin,
|
||||
inviter=workspace.owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
admin = UserFactory.from_atat_role("default")
|
||||
Workspaces._create_workspace_role(
|
||||
admin, workspace, "admin", status=WorkspaceRoleStatus.ACTIVE
|
||||
)
|
||||
scoped_workspace = Workspaces.get(admin, workspace.id)
|
||||
|
||||
@ -250,23 +250,19 @@ def test_scoped_workspace_returns_all_projects_for_workspace_owner(
|
||||
assert len(scoped_workspace.projects[0].environments) == 3
|
||||
|
||||
|
||||
def test_for_user_returns_assigned_workspaces_for_user(workspace, workspace_owner):
|
||||
def test_for_user_returns_active_workspaces_for_user(workspace, workspace_owner):
|
||||
bob = UserFactory.from_atat_role("default")
|
||||
Workspaces.add_member(workspace, bob, "developer")
|
||||
Workspaces.create(RequestFactory.create())
|
||||
InvitationFactory.create(
|
||||
user=bob,
|
||||
inviter=workspace.owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
WorkspaceRoleFactory.create(
|
||||
user=bob, workspace=workspace, status=WorkspaceRoleStatus.ACTIVE
|
||||
)
|
||||
Workspaces.create(RequestFactory.create())
|
||||
|
||||
bobs_workspaces = Workspaces.for_user(bob)
|
||||
|
||||
assert len(bobs_workspaces) == 1
|
||||
|
||||
|
||||
def test_for_user_does_not_return_unaccepted_workspaces(workspace, workspace_owner):
|
||||
def test_for_user_does_not_return_inactive_workspaces(workspace, workspace_owner):
|
||||
bob = UserFactory.from_atat_role("default")
|
||||
Workspaces.add_member(workspace, bob, "developer")
|
||||
Workspaces.create(RequestFactory.create())
|
||||
@ -286,22 +282,12 @@ def test_for_user_returns_all_workspaces_for_ccpo(workspace, workspace_owner):
|
||||
def test_get_for_update_information():
|
||||
workspace_owner = UserFactory.create()
|
||||
workspace = Workspaces.create(RequestFactory.create(creator=workspace_owner))
|
||||
InvitationFactory.create(
|
||||
user=workspace_owner,
|
||||
inviter=workspace_owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
)
|
||||
owner_ws = Workspaces.get_for_update_information(workspace_owner, workspace.id)
|
||||
assert workspace == owner_ws
|
||||
|
||||
admin = UserFactory.create()
|
||||
Workspaces.add_member(workspace, admin, "admin")
|
||||
InvitationFactory.create(
|
||||
user=admin,
|
||||
inviter=workspace_owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
Workspaces._create_workspace_role(
|
||||
admin, workspace, "admin", status=WorkspaceRoleStatus.ACTIVE
|
||||
)
|
||||
admin_ws = Workspaces.get_for_update_information(admin, workspace.id)
|
||||
assert workspace == admin_ws
|
||||
|
@ -18,7 +18,7 @@ from atst.models.user import User
|
||||
from atst.models.role import Role
|
||||
from atst.models.workspace import Workspace
|
||||
from atst.domain.roles import Roles
|
||||
from atst.models.workspace_role import WorkspaceRole
|
||||
from atst.models.workspace_role import WorkspaceRole, Status as WorkspaceRoleStatus
|
||||
from atst.models.environment_role import EnvironmentRole
|
||||
from atst.models.invitation import Invitation, Status as InvitationStatus
|
||||
from atst.domain.workspaces import Workspaces
|
||||
@ -257,38 +257,25 @@ class WorkspaceFactory(Base):
|
||||
|
||||
workspace.request.creator = owner
|
||||
WorkspaceRoleFactory.create(
|
||||
workspace=workspace, role=Roles.get("owner"), user=owner
|
||||
)
|
||||
InvitationFactory.create(
|
||||
user=owner,
|
||||
inviter=owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
role=Roles.get("owner"),
|
||||
user=owner,
|
||||
status=WorkspaceRoleStatus.ACTIVE,
|
||||
)
|
||||
|
||||
for member in members:
|
||||
user = member.get("user", UserFactory.create())
|
||||
role_name = member["role_name"]
|
||||
WorkspaceRoleFactory.create(
|
||||
workspace=workspace, role=Roles.get(role_name), user=user
|
||||
workspace=workspace,
|
||||
role=Roles.get(role_name),
|
||||
user=user,
|
||||
status=WorkspaceRoleStatus.ACTIVE,
|
||||
)
|
||||
|
||||
workspace.projects = projects
|
||||
return workspace
|
||||
|
||||
@classmethod
|
||||
def create_user_and_workspace_with_role(cls, role="owner"):
|
||||
user = UserFactory.create()
|
||||
workspace = WorkspaceFactory.create()
|
||||
Workspaces._create_workspace_role(user, workspace, role)
|
||||
InvitationFactory.create(
|
||||
user=user,
|
||||
inviter=workspace.owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
)
|
||||
return user, workspace
|
||||
|
||||
|
||||
class ProjectFactory(Base):
|
||||
class Meta:
|
||||
|
@ -1,25 +1,10 @@
|
||||
from tests.factories import (
|
||||
UserFactory,
|
||||
WorkspaceFactory,
|
||||
RequestFactory,
|
||||
InvitationFactory,
|
||||
)
|
||||
from tests.factories import UserFactory, WorkspaceFactory, RequestFactory
|
||||
from atst.domain.workspaces import Workspaces
|
||||
from atst.models.invitation import Status as InvitationStatus
|
||||
|
||||
|
||||
def test_user_with_workspaces_has_workspaces_nav(client, user_session):
|
||||
user = UserFactory.create()
|
||||
workspace = WorkspaceFactory.create()
|
||||
Workspaces._create_workspace_role(user, workspace, "developer")
|
||||
InvitationFactory.create(
|
||||
user=user,
|
||||
inviter=workspace.owner,
|
||||
workspace=workspace,
|
||||
status=InvitationStatus.ACCEPTED,
|
||||
)
|
||||
|
||||
user_session(user)
|
||||
user_session(workspace.owner)
|
||||
response = client.get("/home", follow_redirects=True)
|
||||
assert b'href="/workspaces"' in response.data
|
||||
|
||||
|
@ -1,21 +1,25 @@
|
||||
from flask import url_for
|
||||
|
||||
from tests.factories import UserFactory, WorkspaceFactory, InvitationFactory
|
||||
from tests.factories import (
|
||||
UserFactory,
|
||||
WorkspaceFactory,
|
||||
WorkspaceRoleFactory,
|
||||
InvitationFactory,
|
||||
)
|
||||
from atst.domain.workspaces import Workspaces
|
||||
from atst.domain.workspace_users import WorkspaceUsers
|
||||
from atst.domain.projects import Projects
|
||||
from atst.domain.environments import Environments
|
||||
from atst.domain.environment_roles import EnvironmentRoles
|
||||
from atst.domain.invitations import Invitations
|
||||
from atst.models.workspace_user import WorkspaceUser
|
||||
from atst.models.workspace_role import Status as WorkspaceRoleStatus
|
||||
from atst.models.invitation import Status as InvitationStatus
|
||||
from atst.queue import queue
|
||||
|
||||
|
||||
def test_user_with_permission_has_budget_report_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("owner")
|
||||
|
||||
user_session(user)
|
||||
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
|
||||
@ -23,8 +27,11 @@ def test_user_with_permission_has_budget_report_link(client, user_session):
|
||||
|
||||
|
||||
def test_user_without_permission_has_no_budget_report_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("developer")
|
||||
|
||||
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 (
|
||||
@ -34,9 +41,8 @@ def test_user_without_permission_has_no_budget_report_link(client, user_session)
|
||||
|
||||
|
||||
def test_user_with_permission_has_add_project_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("owner")
|
||||
|
||||
user_session(user)
|
||||
workspace = WorkspaceFactory.create()
|
||||
user_session(workspace.owner)
|
||||
response = client.get("/workspaces/{}/projects".format(workspace.id))
|
||||
assert (
|
||||
'href="/workspaces/{}/projects/new"'.format(workspace.id).encode()
|
||||
@ -45,8 +51,9 @@ def test_user_with_permission_has_add_project_link(client, user_session):
|
||||
|
||||
|
||||
def test_user_without_permission_has_no_add_project_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("developer")
|
||||
|
||||
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 (
|
||||
@ -56,9 +63,8 @@ def test_user_without_permission_has_no_add_project_link(client, user_session):
|
||||
|
||||
|
||||
def test_user_with_permission_has_add_member_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("owner")
|
||||
|
||||
user_session(user)
|
||||
workspace = WorkspaceFactory.create()
|
||||
user_session(workspace.owner)
|
||||
response = client.get("/workspaces/{}/members".format(workspace.id))
|
||||
assert (
|
||||
'href="/workspaces/{}/members/new"'.format(workspace.id).encode()
|
||||
@ -67,8 +73,9 @@ def test_user_with_permission_has_add_member_link(client, user_session):
|
||||
|
||||
|
||||
def test_user_without_permission_has_no_add_member_link(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("developer")
|
||||
|
||||
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 (
|
||||
@ -78,9 +85,8 @@ def test_user_without_permission_has_no_add_member_link(client, user_session):
|
||||
|
||||
|
||||
def test_update_workspace_name(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
user_session(user)
|
||||
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"},
|
||||
@ -91,16 +97,15 @@ def test_update_workspace_name(client, user_session):
|
||||
|
||||
|
||||
def test_view_edit_project(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
project = Projects.create(
|
||||
owner,
|
||||
workspace.owner,
|
||||
workspace,
|
||||
"Snazzy Project",
|
||||
"A new project for me and my friends",
|
||||
{"env1", "env2"},
|
||||
)
|
||||
user_session(owner)
|
||||
user_session(workspace.owner)
|
||||
response = client.get(
|
||||
"/workspaces/{}/projects/{}/edit".format(workspace.id, project.id)
|
||||
)
|
||||
@ -168,12 +173,11 @@ def test_user_without_permission_cannot_update_project(client, user_session):
|
||||
|
||||
|
||||
def test_create_member(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
user_session(owner)
|
||||
user = UserFactory.create()
|
||||
workspace = WorkspaceFactory.create()
|
||||
user_session(workspace.owner)
|
||||
queue_length = len(queue.get_queue())
|
||||
|
||||
user = UserFactory.create()
|
||||
response = client.post(
|
||||
url_for("workspaces.create_member", workspace_id=workspace.id),
|
||||
data={
|
||||
@ -193,8 +197,10 @@ def test_create_member(client, user_session):
|
||||
|
||||
|
||||
def test_permissions_for_view_member(client, user_session):
|
||||
user, workspace = WorkspaceFactory.create_user_and_workspace_with_role("developer")
|
||||
|
||||
user = UserFactory.create()
|
||||
workspace = WorkspaceFactory.create()
|
||||
Workspaces._create_workspace_role(user, workspace, "developer")
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
user_session(user)
|
||||
response = client.post(
|
||||
url_for("workspaces.view_member", workspace_id=workspace.id, member_id=user.id),
|
||||
@ -204,11 +210,10 @@ def test_permissions_for_view_member(client, user_session):
|
||||
|
||||
|
||||
def test_update_member_workspace_role(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
user_session(owner)
|
||||
user_session(workspace.owner)
|
||||
response = client.post(
|
||||
url_for(
|
||||
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
|
||||
@ -221,11 +226,10 @@ def test_update_member_workspace_role(client, user_session):
|
||||
|
||||
|
||||
def test_update_member_workspace_role_with_no_data(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
user_session(owner)
|
||||
user_session(workspace.owner)
|
||||
response = client.post(
|
||||
url_for(
|
||||
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
|
||||
@ -238,12 +242,11 @@ def test_update_member_workspace_role_with_no_data(client, user_session):
|
||||
|
||||
|
||||
def test_update_member_environment_role(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
project = Projects.create(
|
||||
owner,
|
||||
workspace.owner,
|
||||
workspace,
|
||||
"Snazzy Project",
|
||||
"A new project for me and my friends",
|
||||
@ -253,7 +256,7 @@ def test_update_member_environment_role(client, user_session):
|
||||
env2_id = project.environments[1].id
|
||||
for env in project.environments:
|
||||
Environments.add_member(env, user, "developer")
|
||||
user_session(owner)
|
||||
user_session(workspace.owner)
|
||||
response = client.post(
|
||||
url_for(
|
||||
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
|
||||
@ -271,12 +274,11 @@ def test_update_member_environment_role(client, user_session):
|
||||
|
||||
|
||||
def test_update_member_environment_role_with_no_data(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
project = Projects.create(
|
||||
owner,
|
||||
workspace.owner,
|
||||
workspace,
|
||||
"Snazzy Project",
|
||||
"A new project for me and my friends",
|
||||
@ -285,7 +287,7 @@ def test_update_member_environment_role_with_no_data(client, user_session):
|
||||
env1_id = project.environments[0].id
|
||||
for env in project.environments:
|
||||
Environments.add_member(env, user, "developer")
|
||||
user_session(owner)
|
||||
user_session(workspace.owner)
|
||||
response = client.post(
|
||||
url_for(
|
||||
"workspaces.update_member", workspace_id=workspace.id, member_id=user.id
|
||||
@ -298,11 +300,12 @@ def test_update_member_environment_role_with_no_data(client, user_session):
|
||||
|
||||
|
||||
def test_new_member_accepts_valid_invite(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
invite = InvitationFactory.create(user_id=member.user.id, workspace_id=workspace.id)
|
||||
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
|
||||
@ -323,14 +326,13 @@ def test_new_member_accepts_valid_invite(client, user_session):
|
||||
|
||||
|
||||
def test_new_member_accept_invalid_invite(client, user_session):
|
||||
owner, workspace = WorkspaceFactory.create_user_and_workspace_with_role("admin")
|
||||
|
||||
workspace = WorkspaceFactory.create()
|
||||
user = UserFactory.create()
|
||||
member = WorkspaceUsers.add(user, workspace.id, "developer")
|
||||
ws_role = WorkspaceRoleFactory.create(
|
||||
user=user, workspace=workspace, status=WorkspaceRoleStatus.PENDING
|
||||
)
|
||||
invite = InvitationFactory.create(
|
||||
user_id=member.user.id,
|
||||
workspace_id=workspace.id,
|
||||
status=InvitationStatus.REJECTED,
|
||||
user_id=user.id, workspace_role_id=ws_role.id, status=InvitationStatus.REJECTED
|
||||
)
|
||||
user_session(user)
|
||||
response = client.get(url_for("workspaces.accept_invitation", token=invite.token))
|
||||
@ -341,7 +343,6 @@ def test_new_member_accept_invalid_invite(client, user_session):
|
||||
def test_user_who_has_not_accepted_workspace_invite_cannot_view(client, user_session):
|
||||
user = UserFactory.create()
|
||||
workspace = WorkspaceFactory.create()
|
||||
Invitations.create_for_owner(workspace, workspace.owner)
|
||||
|
||||
# create user in workspace with invitation
|
||||
user_session(workspace.owner)
|
||||
|
@ -211,17 +211,3 @@ def test_redirected_on_login(client, monkeypatch):
|
||||
target_route = url_for("requests.requests_form_new", screen=1)
|
||||
response = _login(client, next=target_route)
|
||||
assert target_route in response.headers.get("Location")
|
||||
|
||||
|
||||
def test_invited_user_finalized_on_login(monkeypatch, client):
|
||||
user = UserFactory.create(provisional=True)
|
||||
monkeypatch.setattr(
|
||||
"atst.domain.authnid.AuthenticationContext.authenticate", lambda *args: True
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"atst.domain.authnid.AuthenticationContext.get_user", lambda *args: user
|
||||
)
|
||||
|
||||
resp = _login(client)
|
||||
|
||||
assert not user.provisional
|
||||
|
Loading…
x
Reference in New Issue
Block a user