add WorkspaceUser tests from authz
This commit is contained in:
@@ -1,17 +1,23 @@
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
from sqlalchemy.dialects.postgresql import insert
|
||||
|
||||
from atst.models import User, WorkspaceRole, Role
|
||||
from atst.models.workspace_role import WorkspaceRole
|
||||
from atst.models.workspace_user import WorkspaceUser
|
||||
from atst.models.user import User
|
||||
from .roles import Roles
|
||||
from .users import Users
|
||||
from .exceptions import NotFoundError
|
||||
|
||||
class WorkspaceUsers(object):
|
||||
|
||||
def __init__(self, db_session):
|
||||
self.db_session = db_session
|
||||
self.roles_repo = Roles(db_session)
|
||||
self.users_repo = Users(db_session)
|
||||
|
||||
def get(self, workspace_id, user_id):
|
||||
try:
|
||||
user = User.query.filter_by(id=user_id).one()
|
||||
user = self.users_repo.get(user_id)
|
||||
except NoResultFound:
|
||||
raise NotFoundError("user")
|
||||
|
||||
@@ -31,18 +37,18 @@ class WorkspaceUsers(object):
|
||||
|
||||
for user_dict in workspace_user_dicts:
|
||||
try:
|
||||
user = User.query.filter_by(id=user_dict["id"]).one()
|
||||
user = self.users_repo.get(user_dict["id"])
|
||||
except NoResultFound:
|
||||
default_role = Role.query.filter_by(name="developer").one_or_none()
|
||||
default_role = self.roles_repo.get("developer")
|
||||
user = User(id=user_dict["id"], atat_role=default_role)
|
||||
|
||||
try:
|
||||
role = Role.query.filter_by(name=user_dict["workspace_role"]).one()
|
||||
role = self.roles_repo.get(user_dict["workspace_role"])
|
||||
except NoResultFound:
|
||||
raise NotFoundError("role")
|
||||
|
||||
try:
|
||||
existing_workspace_role = WorkspaceRole.query.filter(
|
||||
existing_workspace_role = self.db_session.query(WorkspaceRole).filter(
|
||||
WorkspaceRole.user == user,
|
||||
WorkspaceRole.workspace_id == workspace_id,
|
||||
).one()
|
||||
|
Reference in New Issue
Block a user