Detach users from shows

server/alembic_config/versions/be353176c064_detach_users_from_shows.py

@@ -0,0 +1,41 @@
+"""Detach users from shows
+
+Revision ID: be353176c064
+Revises: a39ac9e9f085
+Create Date: 2025-04-20 22:27:07.342092
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "be353176c064"
+down_revision: Union[str, None] = "a39ac9e9f085"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.drop_index("ix_user_show_id")
+        try:
+            batch_op.drop_constraint(None, type_="foreignkey")
+        except IndexError:
+            pass
+        batch_op.drop_column("show_id")
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("show_id", sa.INTEGER(), nullable=True))
+        batch_op.create_foreign_key(None, "shows", ["show_id"], ["id"])
+        batch_op.create_index("ix_user_show_id", ["show_id"], unique=False)
+
+    # ### end Alembic commands ###

server/controllers/api/auth.py

@@ -5,7 +5,6 @@
 from tornado.ioloop import IOLoop
 
 from models.session import Session
-from models.show import Show
 from models.user import User
 from schemas.schemas import UserSchema
 from utils.web.base_controller import BaseAPIController
@@ -37,27 +36,9 @@ async def post(self):
             )
             return
 
-        show_id = data.get("show_id", None)
         is_admin = data.get("is_admin", False)
 
-        if not show_id and not is_admin:
-            self.set_status(400)
-            await self.finish({"message": "Non admin user requires a show allocation"})
-            return
-
-        if is_admin and show_id:
-            self.set_status(400)
-            await self.finish({"message": "Admin user cannot have a show allocation"})
-            return
-
         with self.make_session() as session:
-            if show_id:
-                show = session.query(Show).get(show_id)
-                if not show:
-                    self.set_status(400)
-                    await self.finish({"message": "Show not found"})
-                    return
-
             conflict_user = (
                 session.query(User).filter(User.username == username).first()
             )
@@ -75,7 +56,6 @@ async def post(self):
                 User(
                     username=username,
                     password=hashed_password,
-                    show_id=show_id,
                     is_admin=is_admin,
                 )
             )
@@ -113,21 +93,6 @@ async def post(self):
                 await self.finish({"message": "Invalid username/password"})
                 return
 
-            if not user.is_admin:
-                if not self.get_current_show():
-                    self.set_status(403)
-                    await self.finish(
-                        {
-                            "message": "Non admin user cannot log in without a loaded show"
-                        }
-                    )
-                    return
-
-                if user.show_id != self.get_current_show()["id"]:
-                    self.set_status(403)
-                    await self.finish({"message": "Loaded show does not match user"})
-                    return
-
             password_equal = await IOLoop.current().run_in_executor(
                 None,
                 bcrypt.checkpw,
@@ -201,13 +166,7 @@ class UsersHandler(BaseAPIController):
     def get(self):
         user_schema = UserSchema()
         with self.make_session() as session:
-            users = (
-                session.query(User)
-                .filter(
-                    (User.show_id == self.get_current_show()["id"]) | (User.is_admin)
-                )
-                .all()
-            )
+            users = session.query(User).all()
             self.set_status(200)
             self.finish({"users": [user_schema.dump(u) for u in users]})
 

server/models/show.py

@@ -39,7 +39,6 @@ class Show(db.Model):
     )
     scene_list = relationship("Scene", cascade="all, delete-orphan")
     cue_type_list = relationship("CueType", cascade="all, delete-orphan")
-    users = relationship("User", uselist=True, cascade="all, delete-orphan")
 
 
 class Cast(db.Model):

server/models/user.py

@@ -15,7 +15,6 @@ class User(db.Model):
     id = Column(Integer(), primary_key=True, autoincrement=True)
     username = Column(String(), index=True)
     password = Column(String())
-    show_id = Column(Integer(), ForeignKey("shows.id"), index=True)
     is_admin = Column(Boolean())
     last_login = Column(DateTime())
 

server/rbac/rbac_db.py

@@ -9,6 +9,7 @@
 from digi_server.logger import get_logger
 from models.models import db
 from models.show import Show
+from models.user import User
 from rbac.exceptions import RBACException
 from rbac.role import Role
 from utils import tree
@@ -88,7 +89,7 @@ def add_mapping(self, actor: type, resource: type) -> None:
         actor_inspect = inspect(actor)
         resource_inspect = inspect(resource)
 
-        if not self._has_link_to_show(actor_inspect.mapped_table):
+        if actor is not User and not self._has_link_to_show(actor_inspect.mapped_table):
             raise RBACException(
                 "actor class does not have a reference back to Show table"
             )

server/test/test_auth_api.py

@@ -50,67 +50,6 @@ def test_create_admin(self):
         self.assertTrue("message" in response_body)
         self.assertEqual("Successfully created user", response_body["message"])
 
-    def test_invalid_admin(self):
-        response = self.fetch(
-            "/api/v1/auth/create",
-            method="POST",
-            body=escape.json_encode(
-                {
-                    "username": "foobar",
-                    "password": "password",
-                    "is_admin": True,
-                    "show_id": 1,
-                }
-            ),
-        )
-        response_body = escape.json_decode(response.body)
-
-        self.assertEqual(400, response.code)
-        self.assertTrue("message" in response_body)
-        self.assertEqual(
-            "Admin user cannot have a show allocation", response_body["message"]
-        )
-
-    def test_invalid_user(self):
-        response = self.fetch(
-            "/api/v1/auth/create",
-            method="POST",
-            body=escape.json_encode(
-                {
-                    "username": "foobar",
-                    "password": "password",
-                    "is_admin": False,
-                    "show_id": None,
-                }
-            ),
-        )
-        response_body = escape.json_decode(response.body)
-
-        self.assertEqual(400, response.code)
-        self.assertTrue("message" in response_body)
-        self.assertEqual(
-            "Non admin user requires a show allocation", response_body["message"]
-        )
-
-    def test_invalid_show(self):
-        response = self.fetch(
-            "/api/v1/auth/create",
-            method="POST",
-            body=escape.json_encode(
-                {
-                    "username": "foobar",
-                    "password": "password",
-                    "is_admin": False,
-                    "show_id": 1,
-                }
-            ),
-        )
-        response_body = escape.json_decode(response.body)
-
-        self.assertEqual(400, response.code)
-        self.assertTrue("message" in response_body)
-        self.assertEqual("Show not found", response_body["message"])
-
     def test_login_success(self):
         self.fetch(
             "/api/v1/auth/create",