diff --git a/client/package-lock.json b/client/package-lock.json index a93003c0..629e1457 100644 --- a/client/package-lock.json +++ b/client/package-lock.json @@ -23,6 +23,7 @@ "lodash": "4.17.23", "loglevel": "1.9.2", "marked": "11.2.0", + "splitpanes": "^2.4.1", "vue": "2.7.14", "vue-multiselect": "2.1.9", "vue-native-websocket": "2.0.15", @@ -6268,6 +6269,12 @@ "node": ">=0.10.0" } }, + "node_modules/splitpanes": { + "version": "2.4.1", + "resolved": "https://registry.npmjs.org/splitpanes/-/splitpanes-2.4.1.tgz", + "integrity": "sha512-kpEo1WuMXuc6QfdQdO2V/fl/trONlkUKp+pputsLTiW9RMtwEvjb4/aYGm2m3+KAzjmb+zLwr4A4SYZu74+pgQ==", + "license": "MIT" + }, "node_modules/stackback": { "version": "0.0.2", "resolved": "https://registry.npmjs.org/stackback/-/stackback-0.0.2.tgz", diff --git a/client/package.json b/client/package.json index a1d43e4f..45e612e0 100644 --- a/client/package.json +++ b/client/package.json @@ -41,6 +41,7 @@ "lodash": "4.17.23", "loglevel": "1.9.2", "marked": "11.2.0", + "splitpanes": "^2.4.1", "vue": "2.7.14", "vue-multiselect": "2.1.9", "vue-native-websocket": "2.0.15", diff --git a/client/src/App.vue b/client/src/App.vue index 068f974d..27fe71f4 100644 --- a/client/src/App.vue +++ b/client/src/App.vue @@ -60,6 +60,17 @@ > Jump To Page + + {{ STAGE_MANAGER_MODE ? 'Disable' : 'Enable' }} Stage Manager + diff --git a/client/src/views/show/config/ConfigStage.vue b/client/src/views/show/config/ConfigStage.vue new file mode 100644 index 00000000..4dc317c3 --- /dev/null +++ b/client/src/views/show/config/ConfigStage.vue @@ -0,0 +1,39 @@ + + + diff --git a/client/src/vue_components/show/config/stage/CrewList.vue b/client/src/vue_components/show/config/stage/CrewList.vue new file mode 100644 index 00000000..d1c0b566 --- /dev/null +++ b/client/src/vue_components/show/config/stage/CrewList.vue @@ -0,0 +1,220 @@ + + + diff --git a/client/src/vue_components/show/config/stage/PropsList.vue b/client/src/vue_components/show/config/stage/PropsList.vue new file mode 100644 index 00000000..ef0b82bf --- /dev/null +++ b/client/src/vue_components/show/config/stage/PropsList.vue @@ -0,0 +1,489 @@ + + + diff --git a/client/src/vue_components/show/config/stage/SceneryList.vue b/client/src/vue_components/show/config/stage/SceneryList.vue new file mode 100644 index 00000000..1ca44dab --- /dev/null +++ b/client/src/vue_components/show/config/stage/SceneryList.vue @@ -0,0 +1,497 @@ + + + diff --git a/client/src/vue_components/show/config/stage/StageManager.vue b/client/src/vue_components/show/config/stage/StageManager.vue new file mode 100644 index 00000000..693189b0 --- /dev/null +++ b/client/src/vue_components/show/config/stage/StageManager.vue @@ -0,0 +1,518 @@ + + + + + diff --git a/client/src/vue_components/show/live/ScriptViewPane.vue b/client/src/vue_components/show/live/ScriptViewPane.vue new file mode 100644 index 00000000..e5d76612 --- /dev/null +++ b/client/src/vue_components/show/live/ScriptViewPane.vue @@ -0,0 +1,1122 @@ + + + + + diff --git a/client/src/vue_components/show/live/StageManagerPane.vue b/client/src/vue_components/show/live/StageManagerPane.vue new file mode 100644 index 00000000..75341f58 --- /dev/null +++ b/client/src/vue_components/show/live/StageManagerPane.vue @@ -0,0 +1,426 @@ + + + + + diff --git a/server/alembic_config/versions/625ac1e96e88_add_unique_constraints_to_allocation_.py b/server/alembic_config/versions/625ac1e96e88_add_unique_constraints_to_allocation_.py new file mode 100644 index 00000000..e9bb6c6c --- /dev/null +++ b/server/alembic_config/versions/625ac1e96e88_add_unique_constraints_to_allocation_.py @@ -0,0 +1,42 @@ +"""add unique constraints to allocation tables + +Revision ID: 625ac1e96e88 +Revises: 9849eb6d381a +Create Date: 2026-01-16 00:31:42.000334 + +""" + +from typing import Sequence, Union + +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "625ac1e96e88" +down_revision: Union[str, None] = "9849eb6d381a" +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("props_allocation", schema=None) as batch_op: + batch_op.create_unique_constraint("uq_props_scene", ["props_id", "scene_id"]) + + with op.batch_alter_table("scenery_allocation", schema=None) as batch_op: + batch_op.create_unique_constraint( + "uq_scenery_scene", ["scenery_id", "scene_id"] + ) + + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table("scenery_allocation", schema=None) as batch_op: + batch_op.drop_constraint("uq_scenery_scene", type_="unique") + + with op.batch_alter_table("props_allocation", schema=None) as batch_op: + batch_op.drop_constraint("uq_props_scene", type_="unique") + + # ### end Alembic commands ### diff --git a/server/alembic_config/versions/9849eb6d381a_add_prop_and_scenery_types.py b/server/alembic_config/versions/9849eb6d381a_add_prop_and_scenery_types.py new file mode 100644 index 00000000..22f8f34c --- /dev/null +++ b/server/alembic_config/versions/9849eb6d381a_add_prop_and_scenery_types.py @@ -0,0 +1,93 @@ +"""Add prop and scenery types + +Revision ID: 9849eb6d381a +Revises: fa27b233d26c +Create Date: 2026-01-14 01:01:31.586812 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "9849eb6d381a" +down_revision: Union[str, None] = "fa27b233d26c" +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! ### + op.create_table( + "prop_type", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("show_id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(), nullable=False), + sa.Column("description", sa.String(), nullable=True), + sa.ForeignKeyConstraint( + ["show_id"], ["shows.id"], name=op.f("fk_prop_type_show_id_shows") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_prop_type")), + ) + op.create_table( + "scenery_type", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("show_id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(), nullable=False), + sa.Column("description", sa.String(), nullable=True), + sa.ForeignKeyConstraint( + ["show_id"], ["shows.id"], name=op.f("fk_scenery_type_show_id_shows") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_scenery_type")), + ) + with op.batch_alter_table("crew", schema=None) as batch_op: + batch_op.alter_column("first_name", existing_type=sa.String(), nullable=False) + + with op.batch_alter_table("props", schema=None) as batch_op: + batch_op.add_column(sa.Column("prop_type_id", sa.Integer(), nullable=False)) + batch_op.alter_column("name", existing_type=sa.String(), nullable=False) + batch_op.create_foreign_key( + batch_op.f("fk_props_prop_type_id_prop_type"), + "prop_type", + ["prop_type_id"], + ["id"], + ) + + with op.batch_alter_table("scenery", schema=None) as batch_op: + batch_op.add_column(sa.Column("scenery_type_id", sa.Integer(), nullable=False)) + batch_op.alter_column("name", existing_type=sa.String(), nullable=False) + batch_op.create_foreign_key( + batch_op.f("fk_scenery_scenery_type_id_scenery_type"), + "scenery_type", + ["scenery_type_id"], + ["id"], + ) + + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table("scenery", schema=None) as batch_op: + batch_op.drop_constraint( + batch_op.f("fk_scenery_scenery_type_id_scenery_type"), type_="foreignkey" + ) + batch_op.alter_column("name", existing_type=sa.String(), nullable=True) + batch_op.drop_column("scenery_type_id") + + with op.batch_alter_table("props", schema=None) as batch_op: + batch_op.drop_constraint( + batch_op.f("fk_props_prop_type_id_prop_type"), type_="foreignkey" + ) + batch_op.alter_column("name", existing_type=sa.String(), nullable=True) + batch_op.drop_column("prop_type_id") + + with op.batch_alter_table("crew", schema=None) as batch_op: + batch_op.alter_column("first_name", existing_type=sa.String(), nullable=True) + + op.drop_table("scenery_type") + op.drop_table("prop_type") + # ### end Alembic commands ### diff --git a/server/alembic_config/versions/fa27b233d26c_add_crew_props_and_scenery_tables_and_.py b/server/alembic_config/versions/fa27b233d26c_add_crew_props_and_scenery_tables_and_.py new file mode 100644 index 00000000..920b08ba --- /dev/null +++ b/server/alembic_config/versions/fa27b233d26c_add_crew_props_and_scenery_tables_and_.py @@ -0,0 +1,105 @@ +"""Add crew, props and scenery tables and allocations + +Revision ID: fa27b233d26c +Revises: 01fb1d6c6b08 +Create Date: 2026-01-14 00:38:40.210710 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "fa27b233d26c" +down_revision: Union[str, None] = "01fb1d6c6b08" +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! ### + op.create_table( + "crew", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("show_id", sa.Integer(), nullable=False), + sa.Column("first_name", sa.String(), nullable=True), + sa.Column("last_name", sa.String(), nullable=True), + sa.ForeignKeyConstraint( + ["show_id"], ["shows.id"], name=op.f("fk_crew_show_id_shows") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_crew")), + ) + op.create_table( + "props", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("show_id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(), nullable=True), + sa.Column("description", sa.String(), nullable=True), + sa.ForeignKeyConstraint( + ["show_id"], ["shows.id"], name=op.f("fk_props_show_id_shows") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_props")), + ) + op.create_table( + "scenery", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("show_id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(), nullable=True), + sa.Column("description", sa.String(), nullable=True), + sa.ForeignKeyConstraint( + ["show_id"], ["shows.id"], name=op.f("fk_scenery_show_id_shows") + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_scenery")), + ) + op.create_table( + "props_allocation", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("props_id", sa.Integer(), nullable=False), + sa.Column("scene_id", sa.Integer(), nullable=False), + sa.ForeignKeyConstraint( + ["props_id"], + ["props.id"], + name=op.f("fk_props_allocation_props_id_props"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["scene_id"], + ["scene.id"], + name=op.f("fk_props_allocation_scene_id_scene"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_props_allocation")), + ) + op.create_table( + "scenery_allocation", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("scenery_id", sa.Integer(), nullable=False), + sa.Column("scene_id", sa.Integer(), nullable=False), + sa.ForeignKeyConstraint( + ["scene_id"], + ["scene.id"], + name=op.f("fk_scenery_allocation_scene_id_scene"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["scenery_id"], + ["scenery.id"], + name=op.f("fk_scenery_allocation_scenery_id_scenery"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_scenery_allocation")), + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("scenery_allocation") + op.drop_table("props_allocation") + op.drop_table("scenery") + op.drop_table("props") + op.drop_table("crew") + # ### end Alembic commands ### diff --git a/server/controllers/api/show/stage/__init__.py b/server/controllers/api/show/stage/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/server/controllers/api/show/stage/crew.py b/server/controllers/api/show/stage/crew.py new file mode 100644 index 00000000..168556c2 --- /dev/null +++ b/server/controllers/api/show/stage/crew.py @@ -0,0 +1,162 @@ +from tornado import escape + +from models.show import Show +from models.stage import Crew +from rbac.role import Role +from schemas.schemas import CrewSchema +from utils.web.base_controller import BaseAPIController +from utils.web.route import ApiRoute, ApiVersion +from utils.web.web_decorators import no_live_session, requires_show + + +@ApiRoute("show/stage/crew", ApiVersion.V1) +class CrewController(BaseAPIController): + @requires_show + def get(self): + current_show = self.get_current_show() + show_id = current_show["id"] + crew_schema = CrewSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + crew = [crew_schema.dump(c) for c in show.crew_list] + self.set_status(200) + self.finish({"crew": crew}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + first_name = data.get("firstName", None) + if not first_name: + self.set_status(400) + await self.finish({"message": "First name missing"}) + return + + last_name = data.get("lastName", None) + if not last_name: + self.set_status(400) + await self.finish({"message": "Last name missing"}) + return + + new_crew = Crew( + show_id=show.id, first_name=first_name, last_name=last_name + ) + session.add(new_crew) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_crew.id, "message": "Successfully added crew member"} + ) + + await self.application.ws_send_to_all( + "GET_CREW_LIST", "GET_CREW_LIST", {} + ) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def patch(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + crew_id = data.get("id", None) + if not crew_id: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + entry: Crew = session.get(Crew, crew_id) + if entry: + first_name = data.get("firstName", None) + if not first_name: + self.set_status(400) + await self.finish({"message": "First name missing"}) + return + entry.first_name = first_name + + last_name = data.get("lastName", None) + if not last_name: + self.set_status(400) + await self.finish({"message": "Last name missing"}) + return + entry.last_name = last_name + + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully updated crew member"}) + + await self.application.ws_send_to_all( + "GET_CREW_LIST", "GET_CREW_LIST", {} + ) + else: + self.set_status(404) + await self.finish({"message": "404 cast member not found"}) + return + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def delete(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + + crew_id_str = self.get_argument("id", None) + if not crew_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + crew_id = int(crew_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + entry = session.get(Crew, crew_id) + if entry: + session.delete(entry) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted crew member"}) + + await self.application.ws_send_to_all( + "GET_CREW_LIST", "GET_CREW_LIST", {} + ) + else: + self.set_status(404) + await self.finish({"message": "404 crew member not found"}) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) diff --git a/server/controllers/api/show/stage/props.py b/server/controllers/api/show/stage/props.py new file mode 100644 index 00000000..2234550c --- /dev/null +++ b/server/controllers/api/show/stage/props.py @@ -0,0 +1,504 @@ +from sqlalchemy import select +from tornado import escape + +from models.show import Scene, Show +from models.stage import Props, PropsAllocation, PropType +from rbac.role import Role +from schemas.schemas import PropsAllocationSchema, PropsSchema, PropTypeSchema +from utils.web.base_controller import BaseAPIController +from utils.web.route import ApiRoute, ApiVersion +from utils.web.web_decorators import no_live_session, requires_show + + +@ApiRoute("show/stage/props/types", ApiVersion.V1) +class PropsTypesController(BaseAPIController): + @requires_show + def get(self): + current_show = self.get_current_show() + show_id = current_show["id"] + prop_type_schema = PropTypeSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + prop_types = [prop_type_schema.dump(c) for c in show.prop_types] + self.set_status(200) + self.finish({"prop_types": prop_types}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + description = data.get("description", "") + + new_prop_type = PropType( + show_id=show.id, name=name, description=description + ) + session.add(new_prop_type) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_prop_type.id, "message": "Successfully added prop type"} + ) + + await self.application.ws_send_to_all("NOOP", "GET_PROP_TYPES", {}) + + @requires_show + @no_live_session + async def patch(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + prop_type = data.get("id", None) + if not prop_type: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + entry: PropType = session.get(PropType, prop_type) + if not entry: + self.set_status(404) + await self.finish({"message": "404 prop type not found"}) + return + + name = data.get("name", None) + description = data.get("description", "") + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + entry.name = name + entry.description = description + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully updated prop type"}) + + await self.application.ws_send_to_all("NOOP", "GET_PROP_TYPES", {}) + + @requires_show + @no_live_session + async def delete(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + + prop_type_id_str = self.get_argument("id", None) + if not prop_type_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + prop_type_id = int(prop_type_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + entry = session.get(PropType, prop_type_id) + if not entry: + self.set_status(404) + await self.finish({"message": "404 prop type not found"}) + return + + session.delete(entry) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted prop type"}) + + await self.application.ws_send_to_all("NOOP", "GET_PROP_TYPES", {}) + await self.application.ws_send_to_all("NOOP", "GET_PROPS_LIST", {}) + + +@ApiRoute("show/stage/props", ApiVersion.V1) +class PropsController(BaseAPIController): + @requires_show + def get(self): + current_show = self.get_current_show() + show_id = current_show["id"] + props_schema = PropsSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + props = [props_schema.dump(c) for c in show.props_list] + self.set_status(200) + self.finish({"props": props}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + prop_type_id = data.get("prop_type_id", None) + if not prop_type_id: + self.set_status(400) + await self.finish({"message": "Prop type ID missing"}) + return + try: + prop_type_id = int(prop_type_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid prop type ID"}) + return + prop_type: PropType = session.get(PropType, prop_type_id) + if not prop_type: + self.set_status(404) + await self.finish({"message": "Prop type not found"}) + return + if prop_type.show_id != show.id: + self.set_status(400) + await self.finish({"message": "Invalid prop type for show"}) + return + + description = data.get("description", "") + + new_props = Props( + show_id=show.id, + name=name, + description=description, + prop_type_id=prop_type.id, + ) + session.add(new_props) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_props.id, "message": "Successfully added props"} + ) + + await self.application.ws_send_to_all("NOOP", "GET_PROPS_LIST", {}) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def patch(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + props = data.get("id", None) + if not props: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + entry: Props = session.get(Props, props) + if entry: + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + entry.name = name + + prop_type_id = data.get("prop_type_id", None) + if not prop_type_id: + self.set_status(400) + await self.finish({"message": "Prop type ID missing"}) + return + try: + prop_type_id = int(prop_type_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid prop type ID"}) + return + prop_type: PropType = session.get(PropType, prop_type_id) + if not prop_type: + self.set_status(404) + await self.finish({"message": "Prop type not found"}) + return + if prop_type.show_id != show.id: + self.set_status(400) + await self.finish({"message": "Invalid prop type for show"}) + return + entry.prop_type_id = prop_type.id + + description = data.get("description", "") + entry.description = description + + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully updated props"}) + + await self.application.ws_send_to_all("NOOP", "GET_PROPS_LIST", {}) + else: + self.set_status(404) + await self.finish({"message": "404 prop not found"}) + return + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def delete(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + + props_id_str = self.get_argument("id", None) + if not props_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + props_id = int(props_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + entry = session.get(Props, props_id) + if entry: + session.delete(entry) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted props"}) + + await self.application.ws_send_to_all("NOOP", "GET_PROPS_LIST", {}) + else: + self.set_status(404) + await self.finish({"message": "404 props not found"}) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + +@ApiRoute("show/stage/props/allocations", ApiVersion.V1) +class PropsAllocationController(BaseAPIController): + """Controller for managing props allocations to scenes.""" + + @requires_show + def get(self): + """Get all props allocations for the current show.""" + current_show = self.get_current_show() + show_id = current_show["id"] + allocation_schema = PropsAllocationSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + allocations = session.scalars( + select(PropsAllocation) + .join(Props, PropsAllocation.props_id == Props.id) + .where(Props.show_id == show_id) + ).all() + allocations = [allocation_schema.dump(a) for a in allocations] + self.set_status(200) + self.finish({"allocations": allocations}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + """Create a new props allocation.""" + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + # Validate props_id + props_id = data.get("props_id", None) + if props_id is None: + self.set_status(400) + await self.finish({"message": "props_id missing"}) + return + + try: + props_id = int(props_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid props_id"}) + return + + prop: Props = session.get(Props, props_id) + if not prop: + self.set_status(404) + await self.finish({"message": "404 prop not found"}) + return + + if prop.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 prop not found"}) + return + + # Validate scene_id + scene_id = data.get("scene_id", None) + if scene_id is None: + self.set_status(400) + await self.finish({"message": "scene_id missing"}) + return + + try: + scene_id = int(scene_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid scene_id"}) + return + + scene: Scene = session.get(Scene, scene_id) + if not scene: + self.set_status(404) + await self.finish({"message": "404 scene not found"}) + return + + if scene.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 scene not found"}) + return + + # Check for duplicate allocation + existing = session.scalars( + select(PropsAllocation).where( + PropsAllocation.props_id == props_id, + PropsAllocation.scene_id == scene_id, + ) + ).first() + if existing: + self.set_status(400) + await self.finish( + {"message": "Allocation already exists for this prop and scene"} + ) + return + + new_allocation = PropsAllocation(props_id=props_id, scene_id=scene_id) + session.add(new_allocation) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_allocation.id, "message": "Successfully added allocation"} + ) + + await self.application.ws_send_to_all("NOOP", "GET_PROPS_ALLOCATIONS", {}) + + @requires_show + @no_live_session + async def delete(self): + """Delete a props allocation by ID (query parameter).""" + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + + self.requires_role(show, Role.WRITE) + + allocation_id_str = self.get_argument("id", None) + if not allocation_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + allocation_id = int(allocation_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + allocation: PropsAllocation = session.get(PropsAllocation, allocation_id) + if not allocation: + self.set_status(404) + await self.finish({"message": "404 allocation not found"}) + return + + # Verify the allocation belongs to a prop in this show + prop: Props = session.get(Props, allocation.props_id) + if not prop or prop.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 allocation not found"}) + return + + session.delete(allocation) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted allocation"}) + + await self.application.ws_send_to_all("NOOP", "GET_PROPS_ALLOCATIONS", {}) diff --git a/server/controllers/api/show/stage/scenery.py b/server/controllers/api/show/stage/scenery.py new file mode 100644 index 00000000..663945cf --- /dev/null +++ b/server/controllers/api/show/stage/scenery.py @@ -0,0 +1,517 @@ +from sqlalchemy import select +from tornado import escape + +from models.show import Scene, Show +from models.stage import Scenery, SceneryAllocation, SceneryType +from rbac.role import Role +from schemas.schemas import SceneryAllocationSchema, ScenerySchema, SceneryTypeSchema +from utils.web.base_controller import BaseAPIController +from utils.web.route import ApiRoute, ApiVersion +from utils.web.web_decorators import no_live_session, requires_show + + +@ApiRoute("show/stage/scenery/types", ApiVersion.V1) +class SceneryTypesController(BaseAPIController): + @requires_show + def get(self): + current_show = self.get_current_show() + show_id = current_show["id"] + scenery_type_schema = SceneryTypeSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + scenery_types = [ + scenery_type_schema.dump(c) for c in show.scenery_types + ] + self.set_status(200) + self.finish({"scenery_types": scenery_types}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + description = data.get("description", "") + + new_scenery_type = SceneryType( + show_id=show.id, name=name, description=description + ) + session.add(new_scenery_type) + session.commit() + + self.set_status(200) + await self.finish( + { + "id": new_scenery_type.id, + "message": "Successfully added scenery type", + } + ) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_TYPES", {}) + + @requires_show + @no_live_session + async def patch(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + prop_type = data.get("id", None) + if not prop_type: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + entry: SceneryType = session.get(SceneryType, prop_type) + if not entry: + self.set_status(404) + await self.finish({"message": "404 scenery type not found"}) + return + + name = data.get("name", None) + description = data.get("description", "") + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + entry.name = name + entry.description = description + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully updated scenery type"}) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_TYPES", {}) + + @requires_show + @no_live_session + async def delete(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + self.requires_role(show, Role.WRITE) + + scenery_type_id_str = self.get_argument("id", None) + if not scenery_type_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + scenery_type_id = int(scenery_type_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + entry = session.get(SceneryType, scenery_type_id) + if not entry: + self.set_status(404) + await self.finish({"message": "404 scenery type not found"}) + return + + session.delete(entry) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted scenery type"}) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_TYPES", {}) + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_LIST", {}) + + +@ApiRoute("show/stage/scenery", ApiVersion.V1) +class SceneryController(BaseAPIController): + @requires_show + def get(self): + current_show = self.get_current_show() + show_id = current_show["id"] + scenery_schema = ScenerySchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + scenery = [scenery_schema.dump(c) for c in show.scenery_list] + self.set_status(200) + self.finish({"scenery": scenery}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + + scenery_type_id = data.get("scenery_type_id", None) + if not scenery_type_id: + self.set_status(400) + await self.finish({"message": "Scenery type ID missing"}) + return + try: + scenery_type_id = int(scenery_type_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Scenery prop type ID"}) + return + scenery_type: SceneryType = session.get(SceneryType, scenery_type_id) + if not scenery_type: + self.set_status(404) + await self.finish({"message": "Scenery type not found"}) + return + if scenery_type.show_id != show.id: + self.set_status(400) + await self.finish({"message": "Invalid scenery type for show"}) + return + + description = data.get("description", "") + + new_scenery = Scenery( + show_id=show.id, + name=name, + description=description, + scenery_type_id=scenery_type.id, + ) + session.add(new_scenery) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_scenery.id, "message": "Successfully added scenery"} + ) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_LIST", {}) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def patch(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + scenery = data.get("id", None) + if not scenery: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + entry: Scenery = session.get(Scenery, scenery) + if entry: + name = data.get("name", None) + if not name: + self.set_status(400) + await self.finish({"message": "Name missing"}) + return + entry.name = name + + scenery_type_id = data.get("scenery_type_id", None) + if not scenery_type_id: + self.set_status(400) + await self.finish({"message": "Scenery type ID missing"}) + return + try: + scenery_type_id = int(scenery_type_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Scenery prop type ID"}) + return + scenery_type: SceneryType = session.get( + SceneryType, scenery_type_id + ) + if not scenery_type: + self.set_status(404) + await self.finish({"message": "Scenery type not found"}) + return + if scenery_type.show_id != show.id: + self.set_status(400) + await self.finish({"message": "Invalid scenery type for show"}) + return + entry.scenery_type_id = scenery_type.id + + description = data.get("description", "") + entry.description = description + + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully updated scenery"}) + + await self.application.ws_send_to_all( + "NOOP", "GET_SCENERY_LIST", {} + ) + else: + self.set_status(404) + await self.finish({"message": "404 cast member not found"}) + return + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def delete(self): + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + self.requires_role(show, Role.WRITE) + + scenery_id_str = self.get_argument("id", None) + if not scenery_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + scenery_id = int(scenery_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + entry = session.get(Scenery, scenery_id) + if entry: + session.delete(entry) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted scenery"}) + + await self.application.ws_send_to_all( + "NOOP", "GET_SCENERY_LIST", {} + ) + else: + self.set_status(404) + await self.finish({"message": "404 scenery not found"}) + else: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + + +@ApiRoute("show/stage/scenery/allocations", ApiVersion.V1) +class SceneryAllocationController(BaseAPIController): + """Controller for managing scenery allocations to scenes.""" + + @requires_show + def get(self): + """Get all scenery allocations for the current show.""" + current_show = self.get_current_show() + show_id = current_show["id"] + allocation_schema = SceneryAllocationSchema() + + with self.make_session() as session: + show = session.get(Show, show_id) + if show: + allocations = session.scalars( + select(SceneryAllocation) + .join(Scenery, SceneryAllocation.scenery_id == Scenery.id) + .where(Scenery.show_id == show_id) + ).all() + allocations = [allocation_schema.dump(a) for a in allocations] + self.set_status(200) + self.finish({"allocations": allocations}) + else: + self.set_status(404) + self.finish({"message": "404 show not found"}) + + @requires_show + @no_live_session + async def post(self): + """Create a new scenery allocation.""" + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + + self.requires_role(show, Role.WRITE) + data = escape.json_decode(self.request.body) + + # Validate scenery_id + scenery_id = data.get("scenery_id", None) + if scenery_id is None: + self.set_status(400) + await self.finish({"message": "scenery_id missing"}) + return + + try: + scenery_id = int(scenery_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid scenery_id"}) + return + + scenery: Scenery = session.get(Scenery, scenery_id) + if not scenery: + self.set_status(404) + await self.finish({"message": "404 scenery not found"}) + return + + if scenery.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 scenery not found"}) + return + + # Validate scene_id + scene_id = data.get("scene_id", None) + if scene_id is None: + self.set_status(400) + await self.finish({"message": "scene_id missing"}) + return + + try: + scene_id = int(scene_id) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid scene_id"}) + return + + scene: Scene = session.get(Scene, scene_id) + if not scene: + self.set_status(404) + await self.finish({"message": "404 scene not found"}) + return + + if scene.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 scene not found"}) + return + + # Check for duplicate allocation + existing = session.scalars( + select(SceneryAllocation).where( + SceneryAllocation.scenery_id == scenery_id, + SceneryAllocation.scene_id == scene_id, + ) + ).first() + if existing: + self.set_status(400) + await self.finish( + {"message": "Allocation already exists for this scenery and scene"} + ) + return + + new_allocation = SceneryAllocation(scenery_id=scenery_id, scene_id=scene_id) + session.add(new_allocation) + session.commit() + + self.set_status(200) + await self.finish( + {"id": new_allocation.id, "message": "Successfully added allocation"} + ) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_ALLOCATIONS", {}) + + @requires_show + @no_live_session + async def delete(self): + """Delete a scenery allocation by ID (query parameter).""" + current_show = self.get_current_show() + show_id = current_show["id"] + + with self.make_session() as session: + show = session.get(Show, show_id) + if not show: + self.set_status(404) + await self.finish({"message": "404 show not found"}) + return + + self.requires_role(show, Role.WRITE) + + allocation_id_str = self.get_argument("id", None) + if not allocation_id_str: + self.set_status(400) + await self.finish({"message": "ID missing"}) + return + + try: + allocation_id = int(allocation_id_str) + except ValueError: + self.set_status(400) + await self.finish({"message": "Invalid ID"}) + return + + allocation: SceneryAllocation = session.get( + SceneryAllocation, allocation_id + ) + if not allocation: + self.set_status(404) + await self.finish({"message": "404 allocation not found"}) + return + + # Verify the allocation belongs to scenery in this show + scenery: Scenery = session.get(Scenery, allocation.scenery_id) + if not scenery or scenery.show_id != show_id: + self.set_status(404) + await self.finish({"message": "404 allocation not found"}) + return + + session.delete(allocation) + session.commit() + + self.set_status(200) + await self.finish({"message": "Successfully deleted allocation"}) + + await self.application.ws_send_to_all("NOOP", "GET_SCENERY_ALLOCATIONS", {}) diff --git a/server/models/show.py b/server/models/show.py index ce739b63..7ba93f54 100644 --- a/server/models/show.py +++ b/server/models/show.py @@ -15,6 +15,15 @@ from models.mics import MicrophoneAllocation from models.script import ScriptLine from models.session import ShowSession + from models.stage import ( + Crew, + Props, + PropsAllocation, + PropType, + Scenery, + SceneryAllocation, + SceneryType, + ) class ShowScriptType(enum.IntEnum): @@ -66,8 +75,27 @@ class Show(db.Model): current_session: Mapped[ShowSession] = relationship( foreign_keys=[current_session_id] ) - cast_list: Mapped[List[Cast]] = relationship(cascade="all, delete-orphan") + crew_list: Mapped[List[Crew]] = relationship( + back_populates="show", + cascade="all, delete-orphan", + ) + scenery_types: Mapped[List[SceneryType]] = relationship( + back_populates="show", + cascade="all, delete-orphan", + ) + scenery_list: Mapped[List[Scenery]] = relationship( + back_populates="show", + cascade="all, delete-orphan", + ) + prop_types: Mapped[List[PropType]] = relationship( + back_populates="show", + cascade="all, delete-orphan", + ) + props_list: Mapped[List[Props]] = relationship( + back_populates="show", + cascade="all, delete-orphan", + ) character_list: Mapped[List[Character]] = relationship(cascade="all, delete-orphan") character_group_list: Mapped[List[CharacterGroup]] = relationship( cascade="all, delete-orphan" @@ -192,3 +220,11 @@ class Scene(db.Model): mic_allocations: Mapped[List[MicrophoneAllocation]] = relationship( cascade="all, delete-orphan", back_populates="scene" ) + scenery_allocations: Mapped[List["SceneryAllocation"]] = relationship( + back_populates="scene", + cascade="all, delete-orphan", + ) + props_allocations: Mapped[List["PropsAllocation"]] = relationship( + back_populates="scene", + cascade="all, delete-orphan", + ) diff --git a/server/models/stage.py b/server/models/stage.py new file mode 100644 index 00000000..795aefd3 --- /dev/null +++ b/server/models/stage.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from typing import List + +from sqlalchemy import ForeignKey, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from models.models import db +from models.show import Scene, Show + + +class Crew(db.Model): + __tablename__ = "crew" + + id: Mapped[int] = mapped_column(primary_key=True) + show_id: Mapped[int] = mapped_column(ForeignKey("shows.id")) + first_name: Mapped[str] = mapped_column() + last_name: Mapped[str | None] = mapped_column() + + show: Mapped[Show] = relationship(back_populates="crew_list") + + +class SceneryAllocation(db.Model): + __tablename__ = "scenery_allocation" + __table_args__ = ( + UniqueConstraint("scenery_id", "scene_id", name="uq_scenery_scene"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + scenery_id: Mapped[int] = mapped_column( + ForeignKey("scenery.id", ondelete="CASCADE") + ) + scene_id: Mapped[int] = mapped_column(ForeignKey("scene.id", ondelete="CASCADE")) + + scenery: Mapped[Scenery] = relationship( + back_populates="scene_allocations", + foreign_keys=[scenery_id], + ) + scene: Mapped[Scene] = relationship( + back_populates="scenery_allocations", + foreign_keys=[scene_id], + ) + + +class PropsAllocation(db.Model): + __tablename__ = "props_allocation" + __table_args__ = (UniqueConstraint("props_id", "scene_id", name="uq_props_scene"),) + + id: Mapped[int] = mapped_column(primary_key=True) + props_id: Mapped[int] = mapped_column(ForeignKey("props.id", ondelete="CASCADE")) + scene_id: Mapped[int] = mapped_column(ForeignKey("scene.id", ondelete="CASCADE")) + + prop: Mapped[Props] = relationship( + back_populates="scene_allocations", + foreign_keys=[props_id], + ) + scene: Mapped[Scene] = relationship( + back_populates="props_allocations", + foreign_keys=[scene_id], + ) + + +class SceneryType(db.Model): + __tablename__ = "scenery_type" + + id: Mapped[int] = mapped_column(primary_key=True) + show_id: Mapped[int] = mapped_column(ForeignKey("shows.id")) + name: Mapped[str] = mapped_column() + description: Mapped[str | None] = mapped_column() + + show: Mapped[Show] = relationship(back_populates="scenery_types") + scenery_items: Mapped[list[Scenery]] = relationship( + back_populates="scenery_type", + cascade="all, delete-orphan", + ) + + +class Scenery(db.Model): + __tablename__ = "scenery" + + id: Mapped[int] = mapped_column(primary_key=True) + show_id: Mapped[int] = mapped_column(ForeignKey("shows.id")) + scenery_type_id: Mapped[int] = mapped_column(ForeignKey("scenery_type.id")) + name: Mapped[str] = mapped_column() + description: Mapped[str | None] = mapped_column() + + show: Mapped[Show] = relationship(back_populates="scenery_list") + scenery_type: Mapped[SceneryType] = relationship(back_populates="scenery_items") + scene_allocations: Mapped[List[SceneryAllocation]] = relationship( + back_populates="scenery", + cascade="all, delete-orphan", + ) + + +class PropType(db.Model): + __tablename__ = "prop_type" + + id: Mapped[int] = mapped_column(primary_key=True) + show_id: Mapped[int] = mapped_column(ForeignKey("shows.id")) + name: Mapped[str] = mapped_column() + description: Mapped[str | None] = mapped_column() + + show: Mapped[Show] = relationship(back_populates="prop_types") + prop_items: Mapped[List[Props]] = relationship( + back_populates="prop_type", + cascade="all, delete-orphan", + ) + + +class Props(db.Model): + __tablename__ = "props" + + id: Mapped[int] = mapped_column(primary_key=True) + show_id: Mapped[int] = mapped_column(ForeignKey("shows.id")) + prop_type_id: Mapped[int] = mapped_column(ForeignKey("prop_type.id")) + name: Mapped[str] = mapped_column() + description: Mapped[str | None] = mapped_column() + + show: Mapped[Show] = relationship(back_populates="props_list") + prop_type: Mapped[PropType] = relationship(back_populates="prop_items") + scene_allocations: Mapped[list[PropsAllocation]] = relationship( + back_populates="prop", + cascade="all, delete-orphan", + ) diff --git a/server/schemas/schemas.py b/server/schemas/schemas.py index 84b4a383..d6026206 100644 --- a/server/schemas/schemas.py +++ b/server/schemas/schemas.py @@ -14,6 +14,15 @@ ) from models.session import Interval, Session, SessionTag, ShowSession from models.show import Act, Cast, Character, CharacterGroup, Scene, Show +from models.stage import ( + Crew, + Props, + PropsAllocation, + PropType, + Scenery, + SceneryAllocation, + SceneryType, +) from models.user import User, UserSettings from registry.schema import get_registry @@ -72,6 +81,62 @@ class Meta: ) +@schema +class CrewSchema(SQLAlchemyAutoSchema): + class Meta: + model = Crew + load_instance = True + include_fk = True + + +@schema +class SceneryTypeSchema(SQLAlchemyAutoSchema): + class Meta: + model = SceneryType + load_instance = True + include_fk = True + + +@schema +class ScenerySchema(SQLAlchemyAutoSchema): + class Meta: + model = Scenery + load_instance = True + include_fk = True + + +@schema +class PropsSchema(SQLAlchemyAutoSchema): + class Meta: + model = Props + load_instance = True + include_fk = True + + +@schema +class PropTypeSchema(SQLAlchemyAutoSchema): + class Meta: + model = PropType + load_instance = True + include_fk = True + + +@schema +class PropsAllocationSchema(SQLAlchemyAutoSchema): + class Meta: + model = PropsAllocation + load_instance = True + include_fk = True + + +@schema +class SceneryAllocationSchema(SQLAlchemyAutoSchema): + class Meta: + model = SceneryAllocation + load_instance = True + include_fk = True + + @schema class CharacterSchema(SQLAlchemyAutoSchema): class Meta: diff --git a/server/test/controllers/api/show/stage/__init__.py b/server/test/controllers/api/show/stage/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/server/test/controllers/api/show/stage/test_crew.py b/server/test/controllers/api/show/stage/test_crew.py new file mode 100644 index 00000000..49550867 --- /dev/null +++ b/server/test/controllers/api/show/stage/test_crew.py @@ -0,0 +1,288 @@ +import tornado.escape + +from models.show import Show, ShowScriptType +from models.stage import Crew +from models.user import User +from test.conftest import DigiScriptTestCase + + +class TestCrewController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/crew endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + # GET tests + + def test_get_crew_empty(self): + """Test GET with no crew returns empty list.""" + response = self.fetch("/api/v1/show/stage/crew") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("crew", response_body) + self.assertEqual([], response_body["crew"]) + + def test_get_crew_returns_all(self): + """Test GET returns all crew members for the show.""" + with self._app.get_db().sessionmaker() as session: + crew_member = Crew(show_id=self.show_id, first_name="John", last_name="Doe") + session.add(crew_member) + session.commit() + + response = self.fetch("/api/v1/show/stage/crew") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["crew"])) + self.assertEqual("John", response_body["crew"][0]["first_name"]) + self.assertEqual("Doe", response_body["crew"][0]["last_name"]) + + def test_get_crew_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/crew") + self.assertEqual(400, response.code) + + # POST tests + + def test_create_crew_success(self): + """Test POST creates a new crew member.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="POST", + body=tornado.escape.json_encode({"firstName": "Jane", "lastName": "Smith"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify crew member was created + with self._app.get_db().sessionmaker() as session: + crew = session.get(Crew, response_body["id"]) + self.assertIsNotNone(crew) + self.assertEqual("Jane", crew.first_name) + self.assertEqual("Smith", crew.last_name) + + def test_create_crew_missing_first_name(self): + """Test POST returns 400 when firstName is missing.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="POST", + body=tornado.escape.json_encode({"lastName": "Smith"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("First name missing", response_body["message"]) + + def test_create_crew_missing_last_name(self): + """Test POST returns 400 when lastName is missing.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="POST", + body=tornado.escape.json_encode({"firstName": "Jane"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Last name missing", response_body["message"]) + + def test_create_crew_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/crew", + method="POST", + body=tornado.escape.json_encode({"firstName": "Jane", "lastName": "Smith"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # PATCH tests + + def test_update_crew_success(self): + """Test PATCH updates an existing crew member.""" + # Create a crew member first + with self._app.get_db().sessionmaker() as session: + crew_member = Crew(show_id=self.show_id, first_name="John", last_name="Doe") + session.add(crew_member) + session.flush() + crew_id = crew_member.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode( + {"id": crew_id, "firstName": "Jane", "lastName": "Smith"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify update + with self._app.get_db().sessionmaker() as session: + crew = session.get(Crew, crew_id) + self.assertEqual("Jane", crew.first_name) + self.assertEqual("Smith", crew.last_name) + + def test_update_crew_missing_id(self): + """Test PATCH returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode({"firstName": "Jane", "lastName": "Smith"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_update_crew_not_found(self): + """Test PATCH returns 404 for non-existent crew member.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode( + {"id": 99999, "firstName": "Jane", "lastName": "Smith"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_update_crew_missing_first_name(self): + """Test PATCH returns 400 when firstName is missing.""" + # Create a crew member first + with self._app.get_db().sessionmaker() as session: + crew_member = Crew(show_id=self.show_id, first_name="John", last_name="Doe") + session.add(crew_member) + session.flush() + crew_id = crew_member.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode({"id": crew_id, "lastName": "Smith"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("First name missing", response_body["message"]) + + def test_update_crew_missing_last_name(self): + """Test PATCH returns 400 when lastName is missing.""" + # Create a crew member first + with self._app.get_db().sessionmaker() as session: + crew_member = Crew(show_id=self.show_id, first_name="John", last_name="Doe") + session.add(crew_member) + session.flush() + crew_id = crew_member.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode({"id": crew_id, "firstName": "Jane"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Last name missing", response_body["message"]) + + def test_update_crew_no_show(self): + """Test PATCH returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/crew", + method="PATCH", + body=tornado.escape.json_encode( + {"id": 1, "firstName": "Jane", "lastName": "Smith"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # DELETE tests + + def test_delete_crew_success(self): + """Test DELETE removes a crew member.""" + # Create a crew member first + with self._app.get_db().sessionmaker() as session: + crew_member = Crew(show_id=self.show_id, first_name="John", last_name="Doe") + session.add(crew_member) + session.flush() + crew_id = crew_member.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/crew?id={crew_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify deletion + with self._app.get_db().sessionmaker() as session: + crew = session.get(Crew, crew_id) + self.assertIsNone(crew) + + def test_delete_crew_missing_id(self): + """Test DELETE returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/crew", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_crew_not_found(self): + """Test DELETE returns 404 for non-existent crew member.""" + response = self.fetch( + "/api/v1/show/stage/crew?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_crew_invalid_id(self): + """Test DELETE returns 400 when id is not a valid integer.""" + response = self.fetch( + "/api/v1/show/stage/crew?id=not-a-number", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_crew_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/crew?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) diff --git a/server/test/controllers/api/show/stage/test_props.py b/server/test/controllers/api/show/stage/test_props.py new file mode 100644 index 00000000..135c9198 --- /dev/null +++ b/server/test/controllers/api/show/stage/test_props.py @@ -0,0 +1,1058 @@ +import tornado.escape + +from models.show import Act, Scene, Show, ShowScriptType +from models.stage import Props, PropsAllocation, PropType +from models.user import User +from test.conftest import DigiScriptTestCase + + +class TestPropsAllocationController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/props/allocations endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create an act and scene + act = Act(show_id=show.id, name="Act 1", interval_after=False) + session.add(act) + session.flush() + self.act_id = act.id + + scene = Scene( + show_id=show.id, + act_id=act.id, + name="Scene 1", + previous_scene_id=None, + ) + session.add(scene) + session.flush() + self.scene_id = scene.id + + # Create a prop type and prop + prop_type = PropType(show_id=show.id, name="Hand Props", description="") + session.add(prop_type) + session.flush() + self.prop_type_id = prop_type.id + + prop = Props( + show_id=show.id, + prop_type_id=prop_type.id, + name="Sword", + description="Test prop", + ) + session.add(prop) + session.flush() + self.prop_id = prop.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + def test_get_allocations_empty(self): + """Test GET with no allocations returns empty list.""" + response = self.fetch("/api/v1/show/stage/props/allocations") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("allocations", response_body) + self.assertEqual([], response_body["allocations"]) + + def test_get_allocations_returns_all(self): + """Test GET returns all allocations for the show.""" + # Create an allocation + with self._app.get_db().sessionmaker() as session: + allocation = PropsAllocation(props_id=self.prop_id, scene_id=self.scene_id) + session.add(allocation) + session.commit() + + response = self.fetch("/api/v1/show/stage/props/allocations") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["allocations"])) + self.assertEqual(self.prop_id, response_body["allocations"][0]["props_id"]) + self.assertEqual(self.scene_id, response_body["allocations"][0]["scene_id"]) + + def test_get_allocations_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/props/allocations") + self.assertEqual(400, response.code) + + def test_create_allocation_success(self): + """Test POST creates a new allocation.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": self.prop_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify allocation was created + with self._app.get_db().sessionmaker() as session: + allocation = session.get(PropsAllocation, response_body["id"]) + self.assertIsNotNone(allocation) + self.assertEqual(self.prop_id, allocation.props_id) + self.assertEqual(self.scene_id, allocation.scene_id) + + def test_create_allocation_duplicate(self): + """Test POST returns 400 for duplicate allocation.""" + # Create initial allocation + with self._app.get_db().sessionmaker() as session: + allocation = PropsAllocation(props_id=self.prop_id, scene_id=self.scene_id) + session.add(allocation) + session.commit() + + # Try to create duplicate + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": self.prop_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("already exists", response_body["message"]) + + def test_create_allocation_invalid_props_id(self): + """Test POST returns 404 for non-existent prop.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": 99999, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_invalid_scene_id(self): + """Test POST returns 404 for non-existent scene.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": self.prop_id, "scene_id": 99999} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_prop_wrong_show(self): + """Test POST returns 404 for prop from different show.""" + # Create another show with a prop + with self._app.get_db().sessionmaker() as session: + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_prop_type = PropType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_prop_type) + session.flush() + + other_prop = Props( + show_id=other_show.id, + prop_type_id=other_prop_type.id, + name="Other Prop", + description="", + ) + session.add(other_prop) + session.flush() + other_prop_id = other_prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": other_prop_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_missing_props_id(self): + """Test POST returns 400 when props_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode({"scene_id": self.scene_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("props_id missing", response_body["message"]) + + def test_create_allocation_missing_scene_id(self): + """Test POST returns 400 when scene_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode({"props_id": self.prop_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("scene_id missing", response_body["message"]) + + def test_create_allocation_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="POST", + body=tornado.escape.json_encode( + {"props_id": self.prop_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + def test_delete_allocation_success(self): + """Test DELETE removes an allocation.""" + # Create an allocation + with self._app.get_db().sessionmaker() as session: + allocation = PropsAllocation(props_id=self.prop_id, scene_id=self.scene_id) + session.add(allocation) + session.flush() + allocation_id = allocation.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/props/allocations?id={allocation_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify allocation was deleted + with self._app.get_db().sessionmaker() as session: + allocation = session.get(PropsAllocation, allocation_id) + self.assertIsNone(allocation) + + def test_delete_allocation_not_found(self): + """Test DELETE returns 404 for non-existent allocation.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_allocation_missing_id(self): + """Test DELETE returns 400 when ID is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_allocation_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/props/allocations?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_allocation_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props/allocations?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + +class TestPropTypesController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/props/types endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + # GET tests + + def test_get_prop_types_empty(self): + """Test GET with no prop types returns empty list.""" + response = self.fetch("/api/v1/show/stage/props/types") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("prop_types", response_body) + self.assertEqual([], response_body["prop_types"]) + + def test_get_prop_types_returns_all(self): + """Test GET returns all prop types for the show.""" + with self._app.get_db().sessionmaker() as session: + prop_type = PropType( + show_id=self.show_id, name="Hand Props", description="Small items" + ) + session.add(prop_type) + session.commit() + + response = self.fetch("/api/v1/show/stage/props/types") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["prop_types"])) + self.assertEqual("Hand Props", response_body["prop_types"][0]["name"]) + + def test_get_prop_types_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/props/types") + self.assertEqual(400, response.code) + + # POST tests + + def test_create_prop_type_success(self): + """Test POST creates a new prop type.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="POST", + body=tornado.escape.json_encode({"name": "Furniture"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify prop type was created + with self._app.get_db().sessionmaker() as session: + prop_type = session.get(PropType, response_body["id"]) + self.assertIsNotNone(prop_type) + self.assertEqual("Furniture", prop_type.name) + + def test_create_prop_type_with_description(self): + """Test POST creates a prop type with description.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="POST", + body=tornado.escape.json_encode( + {"name": "Furniture", "description": "Tables and chairs"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + + # Verify description was saved + with self._app.get_db().sessionmaker() as session: + prop_type = session.get(PropType, response_body["id"]) + self.assertEqual("Tables and chairs", prop_type.description) + + def test_create_prop_type_missing_name(self): + """Test POST returns 400 when name is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="POST", + body=tornado.escape.json_encode({"description": "Some description"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_create_prop_type_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props/types", + method="POST", + body=tornado.escape.json_encode({"name": "Furniture"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # PATCH tests + + def test_update_prop_type_success(self): + """Test PATCH updates an existing prop type.""" + # Create a prop type first + with self._app.get_db().sessionmaker() as session: + prop_type = PropType( + show_id=self.show_id, name="Hand Props", description="" + ) + session.add(prop_type) + session.flush() + prop_type_id = prop_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props/types", + method="PATCH", + body=tornado.escape.json_encode( + {"id": prop_type_id, "name": "Stage Props", "description": "Updated"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify update + with self._app.get_db().sessionmaker() as session: + prop_type = session.get(PropType, prop_type_id) + self.assertEqual("Stage Props", prop_type.name) + self.assertEqual("Updated", prop_type.description) + + def test_update_prop_type_missing_id(self): + """Test PATCH returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="PATCH", + body=tornado.escape.json_encode({"name": "Stage Props"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_update_prop_type_not_found(self): + """Test PATCH returns 404 for non-existent prop type.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="PATCH", + body=tornado.escape.json_encode({"id": 99999, "name": "Stage Props"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_update_prop_type_missing_name(self): + """Test PATCH returns 400 when name is missing.""" + # Create a prop type first + with self._app.get_db().sessionmaker() as session: + prop_type = PropType( + show_id=self.show_id, name="Hand Props", description="" + ) + session.add(prop_type) + session.flush() + prop_type_id = prop_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props/types", + method="PATCH", + body=tornado.escape.json_encode({"id": prop_type_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_update_prop_type_no_show(self): + """Test PATCH returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props/types", + method="PATCH", + body=tornado.escape.json_encode({"id": 1, "name": "Stage Props"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # DELETE tests + + def test_delete_prop_type_success(self): + """Test DELETE removes a prop type.""" + # Create a prop type first + with self._app.get_db().sessionmaker() as session: + prop_type = PropType( + show_id=self.show_id, name="Hand Props", description="" + ) + session.add(prop_type) + session.flush() + prop_type_id = prop_type.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/props/types?id={prop_type_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify deletion + with self._app.get_db().sessionmaker() as session: + prop_type = session.get(PropType, prop_type_id) + self.assertIsNone(prop_type) + + def test_delete_prop_type_missing_id(self): + """Test DELETE returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props/types", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_prop_type_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/props/types?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_prop_type_not_found(self): + """Test DELETE returns 404 for non-existent prop type.""" + response = self.fetch( + "/api/v1/show/stage/props/types?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_prop_type_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props/types?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + +class TestPropsController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/props endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create a prop type + prop_type = PropType(show_id=show.id, name="Hand Props", description="") + session.add(prop_type) + session.flush() + self.prop_type_id = prop_type.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + # GET tests + + def test_get_props_empty(self): + """Test GET with no props returns empty list.""" + response = self.fetch("/api/v1/show/stage/props") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("props", response_body) + self.assertEqual([], response_body["props"]) + + def test_get_props_returns_all(self): + """Test GET returns all props for the show.""" + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="A prop sword", + ) + session.add(prop) + session.commit() + + response = self.fetch("/api/v1/show/stage/props") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["props"])) + self.assertEqual("Sword", response_body["props"][0]["name"]) + + def test_get_props_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/props") + self.assertEqual(400, response.code) + + # POST tests + + def test_create_props_success(self): + """Test POST creates a new prop.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode( + {"name": "Sword", "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify prop was created + with self._app.get_db().sessionmaker() as session: + prop = session.get(Props, response_body["id"]) + self.assertIsNotNone(prop) + self.assertEqual("Sword", prop.name) + self.assertEqual(self.prop_type_id, prop.prop_type_id) + + def test_create_props_with_description(self): + """Test POST creates a prop with description.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode( + { + "name": "Sword", + "prop_type_id": self.prop_type_id, + "description": "A medieval sword", + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + + # Verify description was saved + with self._app.get_db().sessionmaker() as session: + prop = session.get(Props, response_body["id"]) + self.assertEqual("A medieval sword", prop.description) + + def test_create_props_missing_name(self): + """Test POST returns 400 when name is missing.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode({"prop_type_id": self.prop_type_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_create_props_missing_prop_type_id(self): + """Test POST returns 400 when prop_type_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode({"name": "Sword"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Prop type ID missing", response_body["message"]) + + def test_create_props_invalid_prop_type_id(self): + """Test POST returns 400 for non-integer prop_type_id.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode( + {"name": "Sword", "prop_type_id": "invalid"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid prop type ID", response_body["message"]) + + def test_create_props_prop_type_not_found(self): + """Test POST returns 404 for non-existent prop type.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode({"name": "Sword", "prop_type_id": 99999}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Prop type not found", response_body["message"]) + + def test_create_props_prop_type_wrong_show(self): + """Test POST returns 400 for prop type from different show.""" + # Create another show with a prop type + with self._app.get_db().sessionmaker() as session: + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_prop_type = PropType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_prop_type) + session.flush() + other_prop_type_id = other_prop_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode( + {"name": "Sword", "prop_type_id": other_prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid prop type for show", response_body["message"]) + + def test_create_props_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props", + method="POST", + body=tornado.escape.json_encode( + {"name": "Sword", "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # PATCH tests + + def test_update_props_success(self): + """Test PATCH updates an existing prop.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + { + "id": prop_id, + "name": "Shield", + "prop_type_id": self.prop_type_id, + "description": "Updated", + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify update + with self._app.get_db().sessionmaker() as session: + prop = session.get(Props, prop_id) + self.assertEqual("Shield", prop.name) + self.assertEqual("Updated", prop.description) + + def test_update_props_missing_id(self): + """Test PATCH returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"name": "Shield", "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_update_props_not_found(self): + """Test PATCH returns 404 for non-existent prop.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": 99999, "name": "Shield", "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_update_props_missing_name(self): + """Test PATCH returns 400 when name is missing.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": prop_id, "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_update_props_missing_prop_type_id(self): + """Test PATCH returns 400 when prop_type_id is missing.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode({"id": prop_id, "name": "Shield"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Prop type ID missing", response_body["message"]) + + def test_update_props_invalid_prop_type_id(self): + """Test PATCH returns 400 for non-integer prop_type_id.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": prop_id, "name": "Shield", "prop_type_id": "invalid"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid prop type ID", response_body["message"]) + + def test_update_props_prop_type_not_found(self): + """Test PATCH returns 404 for non-existent prop type.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": prop_id, "name": "Shield", "prop_type_id": 99999} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Prop type not found", response_body["message"]) + + def test_update_props_prop_type_wrong_show(self): + """Test PATCH returns 400 for prop type from different show.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + + # Create another show with a prop type + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_prop_type = PropType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_prop_type) + session.flush() + other_prop_type_id = other_prop_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": prop_id, "name": "Shield", "prop_type_id": other_prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid prop type for show", response_body["message"]) + + def test_update_props_no_show(self): + """Test PATCH returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props", + method="PATCH", + body=tornado.escape.json_encode( + {"id": 1, "name": "Shield", "prop_type_id": self.prop_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # DELETE tests + + def test_delete_props_success(self): + """Test DELETE removes a prop.""" + # Create a prop first + with self._app.get_db().sessionmaker() as session: + prop = Props( + show_id=self.show_id, + prop_type_id=self.prop_type_id, + name="Sword", + description="", + ) + session.add(prop) + session.flush() + prop_id = prop.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/props?id={prop_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify deletion + with self._app.get_db().sessionmaker() as session: + prop = session.get(Props, prop_id) + self.assertIsNone(prop) + + def test_delete_props_missing_id(self): + """Test DELETE returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/props", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_props_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/props?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_props_not_found(self): + """Test DELETE returns 404 for non-existent prop.""" + response = self.fetch( + "/api/v1/show/stage/props?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_props_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/props?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) diff --git a/server/test/controllers/api/show/stage/test_scenery.py b/server/test/controllers/api/show/stage/test_scenery.py new file mode 100644 index 00000000..5298955c --- /dev/null +++ b/server/test/controllers/api/show/stage/test_scenery.py @@ -0,0 +1,1080 @@ +import tornado.escape + +from models.show import Act, Scene, Show, ShowScriptType +from models.stage import Scenery, SceneryAllocation, SceneryType +from models.user import User +from test.conftest import DigiScriptTestCase + + +class TestSceneryAllocationController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/scenery/allocations endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create an act and scene + act = Act(show_id=show.id, name="Act 1", interval_after=False) + session.add(act) + session.flush() + self.act_id = act.id + + scene = Scene( + show_id=show.id, + act_id=act.id, + name="Scene 1", + previous_scene_id=None, + ) + session.add(scene) + session.flush() + self.scene_id = scene.id + + # Create a scenery type and scenery + scenery_type = SceneryType( + show_id=show.id, name="Backdrops", description="" + ) + session.add(scenery_type) + session.flush() + self.scenery_type_id = scenery_type.id + + scenery = Scenery( + show_id=show.id, + scenery_type_id=scenery_type.id, + name="Forest Backdrop", + description="Test scenery", + ) + session.add(scenery) + session.flush() + self.scenery_id = scenery.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + def test_get_allocations_empty(self): + """Test GET with no allocations returns empty list.""" + response = self.fetch("/api/v1/show/stage/scenery/allocations") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("allocations", response_body) + self.assertEqual([], response_body["allocations"]) + + def test_get_allocations_returns_all(self): + """Test GET returns all allocations for the show.""" + # Create an allocation + with self._app.get_db().sessionmaker() as session: + allocation = SceneryAllocation( + scenery_id=self.scenery_id, scene_id=self.scene_id + ) + session.add(allocation) + session.commit() + + response = self.fetch("/api/v1/show/stage/scenery/allocations") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["allocations"])) + self.assertEqual(self.scenery_id, response_body["allocations"][0]["scenery_id"]) + self.assertEqual(self.scene_id, response_body["allocations"][0]["scene_id"]) + + def test_get_allocations_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/scenery/allocations") + self.assertEqual(400, response.code) + + def test_create_allocation_success(self): + """Test POST creates a new allocation.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": self.scenery_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify allocation was created + with self._app.get_db().sessionmaker() as session: + allocation = session.get(SceneryAllocation, response_body["id"]) + self.assertIsNotNone(allocation) + self.assertEqual(self.scenery_id, allocation.scenery_id) + self.assertEqual(self.scene_id, allocation.scene_id) + + def test_create_allocation_duplicate(self): + """Test POST returns 400 for duplicate allocation.""" + # Create initial allocation + with self._app.get_db().sessionmaker() as session: + allocation = SceneryAllocation( + scenery_id=self.scenery_id, scene_id=self.scene_id + ) + session.add(allocation) + session.commit() + + # Try to create duplicate + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": self.scenery_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("already exists", response_body["message"]) + + def test_create_allocation_invalid_scenery_id(self): + """Test POST returns 404 for non-existent scenery.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": 99999, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_invalid_scene_id(self): + """Test POST returns 404 for non-existent scene.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": self.scenery_id, "scene_id": 99999} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_scenery_wrong_show(self): + """Test POST returns 404 for scenery from different show.""" + # Create another show with scenery + with self._app.get_db().sessionmaker() as session: + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_scenery_type = SceneryType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_scenery_type) + session.flush() + + other_scenery = Scenery( + show_id=other_show.id, + scenery_type_id=other_scenery_type.id, + name="Other Scenery", + description="", + ) + session.add(other_scenery) + session.flush() + other_scenery_id = other_scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": other_scenery_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_create_allocation_missing_scenery_id(self): + """Test POST returns 400 when scenery_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode({"scene_id": self.scene_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("scenery_id missing", response_body["message"]) + + def test_create_allocation_missing_scene_id(self): + """Test POST returns 400 when scene_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode({"scenery_id": self.scenery_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("scene_id missing", response_body["message"]) + + def test_create_allocation_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="POST", + body=tornado.escape.json_encode( + {"scenery_id": self.scenery_id, "scene_id": self.scene_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + def test_delete_allocation_success(self): + """Test DELETE removes an allocation.""" + # Create an allocation + with self._app.get_db().sessionmaker() as session: + allocation = SceneryAllocation( + scenery_id=self.scenery_id, scene_id=self.scene_id + ) + session.add(allocation) + session.flush() + allocation_id = allocation.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/scenery/allocations?id={allocation_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify allocation was deleted + with self._app.get_db().sessionmaker() as session: + allocation = session.get(SceneryAllocation, allocation_id) + self.assertIsNone(allocation) + + def test_delete_allocation_not_found(self): + """Test DELETE returns 404 for non-existent allocation.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_allocation_missing_id(self): + """Test DELETE returns 400 when ID is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_allocation_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/scenery/allocations?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_allocation_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery/allocations?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + +class TestSceneryTypesController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/scenery/types endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + # GET tests + + def test_get_scenery_types_empty(self): + """Test GET with no scenery types returns empty list.""" + response = self.fetch("/api/v1/show/stage/scenery/types") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("scenery_types", response_body) + self.assertEqual([], response_body["scenery_types"]) + + def test_get_scenery_types_returns_all(self): + """Test GET returns all scenery types for the show.""" + with self._app.get_db().sessionmaker() as session: + scenery_type = SceneryType( + show_id=self.show_id, name="Backdrops", description="Background pieces" + ) + session.add(scenery_type) + session.commit() + + response = self.fetch("/api/v1/show/stage/scenery/types") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["scenery_types"])) + self.assertEqual("Backdrops", response_body["scenery_types"][0]["name"]) + + def test_get_scenery_types_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/scenery/types") + self.assertEqual(400, response.code) + + # POST tests + + def test_create_scenery_type_success(self): + """Test POST creates a new scenery type.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="POST", + body=tornado.escape.json_encode({"name": "Platforms"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify scenery type was created + with self._app.get_db().sessionmaker() as session: + scenery_type = session.get(SceneryType, response_body["id"]) + self.assertIsNotNone(scenery_type) + self.assertEqual("Platforms", scenery_type.name) + + def test_create_scenery_type_with_description(self): + """Test POST creates a scenery type with description.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="POST", + body=tornado.escape.json_encode( + {"name": "Platforms", "description": "Elevated surfaces"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + + # Verify description was saved + with self._app.get_db().sessionmaker() as session: + scenery_type = session.get(SceneryType, response_body["id"]) + self.assertEqual("Elevated surfaces", scenery_type.description) + + def test_create_scenery_type_missing_name(self): + """Test POST returns 400 when name is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="POST", + body=tornado.escape.json_encode({"description": "Some description"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_create_scenery_type_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="POST", + body=tornado.escape.json_encode({"name": "Platforms"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # PATCH tests + + def test_update_scenery_type_success(self): + """Test PATCH updates an existing scenery type.""" + # Create a scenery type first + with self._app.get_db().sessionmaker() as session: + scenery_type = SceneryType( + show_id=self.show_id, name="Backdrops", description="" + ) + session.add(scenery_type) + session.flush() + scenery_type_id = scenery_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="PATCH", + body=tornado.escape.json_encode( + {"id": scenery_type_id, "name": "Flats", "description": "Updated"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify update + with self._app.get_db().sessionmaker() as session: + scenery_type = session.get(SceneryType, scenery_type_id) + self.assertEqual("Flats", scenery_type.name) + self.assertEqual("Updated", scenery_type.description) + + def test_update_scenery_type_missing_id(self): + """Test PATCH returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="PATCH", + body=tornado.escape.json_encode({"name": "Flats"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_update_scenery_type_not_found(self): + """Test PATCH returns 404 for non-existent scenery type.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="PATCH", + body=tornado.escape.json_encode({"id": 99999, "name": "Flats"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_update_scenery_type_missing_name(self): + """Test PATCH returns 400 when name is missing.""" + # Create a scenery type first + with self._app.get_db().sessionmaker() as session: + scenery_type = SceneryType( + show_id=self.show_id, name="Backdrops", description="" + ) + session.add(scenery_type) + session.flush() + scenery_type_id = scenery_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="PATCH", + body=tornado.escape.json_encode({"id": scenery_type_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_update_scenery_type_no_show(self): + """Test PATCH returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="PATCH", + body=tornado.escape.json_encode({"id": 1, "name": "Flats"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # DELETE tests + + def test_delete_scenery_type_success(self): + """Test DELETE removes a scenery type.""" + # Create a scenery type first + with self._app.get_db().sessionmaker() as session: + scenery_type = SceneryType( + show_id=self.show_id, name="Backdrops", description="" + ) + session.add(scenery_type) + session.flush() + scenery_type_id = scenery_type.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/scenery/types?id={scenery_type_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify deletion + with self._app.get_db().sessionmaker() as session: + scenery_type = session.get(SceneryType, scenery_type_id) + self.assertIsNone(scenery_type) + + def test_delete_scenery_type_missing_id(self): + """Test DELETE returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_scenery_type_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_scenery_type_not_found(self): + """Test DELETE returns 404 for non-existent scenery type.""" + response = self.fetch( + "/api/v1/show/stage/scenery/types?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_scenery_type_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery/types?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + +class TestSceneryController(DigiScriptTestCase): + """Test suite for /api/v1/show/stage/scenery endpoint.""" + + def setUp(self): + super().setUp() + with self._app.get_db().sessionmaker() as session: + show = Show(name="Test Show", script_mode=ShowScriptType.FULL) + session.add(show) + session.flush() + self.show_id = show.id + + # Create a scenery type + scenery_type = SceneryType( + show_id=show.id, name="Backdrops", description="" + ) + session.add(scenery_type) + session.flush() + self.scenery_type_id = scenery_type.id + + # Create admin user for RBAC + admin = User(username="admin", is_admin=True, password="test") + session.add(admin) + session.flush() + self.user_id = admin.id + + session.commit() + + self._app.digi_settings.settings["current_show"].set_value(self.show_id) + self.token = self._app.jwt_service.create_access_token( + data={"user_id": self.user_id} + ) + + # GET tests + + def test_get_scenery_empty(self): + """Test GET with no scenery returns empty list.""" + response = self.fetch("/api/v1/show/stage/scenery") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("scenery", response_body) + self.assertEqual([], response_body["scenery"]) + + def test_get_scenery_returns_all(self): + """Test GET returns all scenery for the show.""" + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Forest Backdrop", + description="A woodland scene", + ) + session.add(scenery) + session.commit() + + response = self.fetch("/api/v1/show/stage/scenery") + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertEqual(1, len(response_body["scenery"])) + self.assertEqual("Forest Backdrop", response_body["scenery"][0]["name"]) + + def test_get_scenery_no_show(self): + """Test GET returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch("/api/v1/show/stage/scenery") + self.assertEqual(400, response.code) + + # POST tests + + def test_create_scenery_success(self): + """Test POST creates a new scenery.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + {"name": "Castle Wall", "scenery_type_id": self.scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("id", response_body) + self.assertIn("message", response_body) + + # Verify scenery was created + with self._app.get_db().sessionmaker() as session: + scenery = session.get(Scenery, response_body["id"]) + self.assertIsNotNone(scenery) + self.assertEqual("Castle Wall", scenery.name) + self.assertEqual(self.scenery_type_id, scenery.scenery_type_id) + + def test_create_scenery_with_description(self): + """Test POST creates a scenery with description.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + { + "name": "Castle Wall", + "scenery_type_id": self.scenery_type_id, + "description": "Stone castle wall", + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + response_body = tornado.escape.json_decode(response.body) + + # Verify description was saved + with self._app.get_db().sessionmaker() as session: + scenery = session.get(Scenery, response_body["id"]) + self.assertEqual("Stone castle wall", scenery.description) + + def test_create_scenery_missing_name(self): + """Test POST returns 400 when name is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode({"scenery_type_id": self.scenery_type_id}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_create_scenery_missing_scenery_type_id(self): + """Test POST returns 400 when scenery_type_id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode({"name": "Castle Wall"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery type ID missing", response_body["message"]) + + def test_create_scenery_invalid_scenery_type_id(self): + """Test POST returns 400 for non-integer scenery_type_id.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + {"name": "Castle Wall", "scenery_type_id": "invalid"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + # Note: The controller has a typo in the error message + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery prop type ID", response_body["message"]) + + def test_create_scenery_scenery_type_not_found(self): + """Test POST returns 404 for non-existent scenery type.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + {"name": "Castle Wall", "scenery_type_id": 99999} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery type not found", response_body["message"]) + + def test_create_scenery_scenery_type_wrong_show(self): + """Test POST returns 400 for scenery type from different show.""" + # Create another show with a scenery type + with self._app.get_db().sessionmaker() as session: + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_scenery_type = SceneryType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_scenery_type) + session.flush() + other_scenery_type_id = other_scenery_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + {"name": "Castle Wall", "scenery_type_id": other_scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid scenery type for show", response_body["message"]) + + def test_create_scenery_no_show(self): + """Test POST returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery", + method="POST", + body=tornado.escape.json_encode( + {"name": "Castle Wall", "scenery_type_id": self.scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # PATCH tests + + def test_update_scenery_success(self): + """Test PATCH updates an existing scenery.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + { + "id": scenery_id, + "name": "Stone Wall", + "scenery_type_id": self.scenery_type_id, + "description": "Updated", + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify update + with self._app.get_db().sessionmaker() as session: + scenery = session.get(Scenery, scenery_id) + self.assertEqual("Stone Wall", scenery.name) + self.assertEqual("Updated", scenery.description) + + def test_update_scenery_missing_id(self): + """Test PATCH returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + {"name": "Stone Wall", "scenery_type_id": self.scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_update_scenery_not_found(self): + """Test PATCH returns 404 for non-existent scenery.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + { + "id": 99999, + "name": "Stone Wall", + "scenery_type_id": self.scenery_type_id, + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_update_scenery_missing_name(self): + """Test PATCH returns 400 when name is missing.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + {"id": scenery_id, "scenery_type_id": self.scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Name missing", response_body["message"]) + + def test_update_scenery_missing_scenery_type_id(self): + """Test PATCH returns 400 when scenery_type_id is missing.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode({"id": scenery_id, "name": "Stone Wall"}), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery type ID missing", response_body["message"]) + + def test_update_scenery_invalid_scenery_type_id(self): + """Test PATCH returns 400 for non-integer scenery_type_id.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + {"id": scenery_id, "name": "Stone Wall", "scenery_type_id": "invalid"} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + # Note: The controller has a typo in the error message + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery prop type ID", response_body["message"]) + + def test_update_scenery_scenery_type_not_found(self): + """Test PATCH returns 404 for non-existent scenery type.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + {"id": scenery_id, "name": "Stone Wall", "scenery_type_id": 99999} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Scenery type not found", response_body["message"]) + + def test_update_scenery_scenery_type_wrong_show(self): + """Test PATCH returns 400 for scenery type from different show.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + + # Create another show with a scenery type + other_show = Show(name="Other Show", script_mode=ShowScriptType.FULL) + session.add(other_show) + session.flush() + + other_scenery_type = SceneryType( + show_id=other_show.id, name="Other Type", description="" + ) + session.add(other_scenery_type) + session.flush() + other_scenery_type_id = other_scenery_type.id + session.commit() + + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + { + "id": scenery_id, + "name": "Stone Wall", + "scenery_type_id": other_scenery_type_id, + } + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid scenery type for show", response_body["message"]) + + def test_update_scenery_no_show(self): + """Test PATCH returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery", + method="PATCH", + body=tornado.escape.json_encode( + {"id": 1, "name": "Stone Wall", "scenery_type_id": self.scenery_type_id} + ), + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + + # DELETE tests + + def test_delete_scenery_success(self): + """Test DELETE removes a scenery.""" + # Create a scenery first + with self._app.get_db().sessionmaker() as session: + scenery = Scenery( + show_id=self.show_id, + scenery_type_id=self.scenery_type_id, + name="Castle Wall", + description="", + ) + session.add(scenery) + session.flush() + scenery_id = scenery.id + session.commit() + + response = self.fetch( + f"/api/v1/show/stage/scenery?id={scenery_id}", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(200, response.code) + + # Verify deletion + with self._app.get_db().sessionmaker() as session: + scenery = session.get(Scenery, scenery_id) + self.assertIsNone(scenery) + + def test_delete_scenery_missing_id(self): + """Test DELETE returns 400 when id is missing.""" + response = self.fetch( + "/api/v1/show/stage/scenery", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("ID missing", response_body["message"]) + + def test_delete_scenery_invalid_id(self): + """Test DELETE returns 400 for non-integer ID.""" + response = self.fetch( + "/api/v1/show/stage/scenery?id=invalid", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code) + response_body = tornado.escape.json_decode(response.body) + self.assertIn("Invalid ID", response_body["message"]) + + def test_delete_scenery_not_found(self): + """Test DELETE returns 404 for non-existent scenery.""" + response = self.fetch( + "/api/v1/show/stage/scenery?id=99999", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(404, response.code) + + def test_delete_scenery_no_show(self): + """Test DELETE returns 400 when no show is loaded.""" + self._app.digi_settings.settings["current_show"].set_value(None) + response = self.fetch( + "/api/v1/show/stage/scenery?id=1", + method="DELETE", + headers={"Authorization": f"Bearer {self.token}"}, + ) + self.assertEqual(400, response.code)