"""Phase 9 — Workspace Types: Operational Context + Functional Workspace templates. Tests cover: - OrganizationNodeType additions (BRANCH, WAREHOUSE) - WorkspaceTemplate model & service (list, create, instantiate from template) - UserOrgContextPreference / UserWorkspacePreference models & services - Context API (GET /workspaces/context/, POST /workspaces/context/) - Workspace-from-template API (POST /workspaces/from-template//) - Template list API (GET /workspaces/templates/) """ from __future__ import annotations import json import pytest from simorgh.apps.iam.registry import sync_registry_to_db from simorgh.apps.memberships.models import Membership from simorgh.apps.organizations.models import OrganizationNode, OrganizationNodeType from simorgh.apps.organizations.services import create_node from simorgh.apps.workspaces.models import ( UserOrgContextPreference, UserWorkspacePreference, Workspace, WorkspaceTemplate, ) from simorgh.apps.workspaces.services import ( add_member, create_workspace, create_workspace_from_template, get_user_org_context, get_user_workspace_context, list_workspace_templates, switch_active_node, switch_active_workspace, ) # --------------------------------------------------------------------------- # Shared fixtures # --------------------------------------------------------------------------- @pytest.fixture def alice_admin_membership(alice, tenant_acme, acme_tree, role_admin) -> Membership: m = Membership.objects.create( tenant=tenant_acme, organization_node=acme_tree["eu"], role=role_admin, ) m.users.add(alice) return m @pytest.fixture def basic_template(db) -> WorkspaceTemplate: return WorkspaceTemplate.objects.create( slug="support-supervisor", name="Support Supervisor", description="Dashboard for support team leads", kind="internal", icon="headset", default_module="helpdesk", navigation_schema=[ {"key": "tickets", "label_key": "nav.tickets", "route": "/tickets", "order": 10}, {"key": "reports", "label_key": "nav.reports", "route": "/reports", "order": 20}, ], ) # --------------------------------------------------------------------------- # OrganizationNodeType — new values # --------------------------------------------------------------------------- class TestOrganizationNodeTypes: def test_branch_choice_exists(self): assert OrganizationNodeType.BRANCH == "branch" def test_warehouse_choice_exists(self): assert OrganizationNodeType.WAREHOUSE == "warehouse" def test_branch_label(self): # Choice display label is translated; check raw value assert "branch" in [c.value for c in OrganizationNodeType] def test_warehouse_label(self): assert "warehouse" in [c.value for c in OrganizationNodeType] @pytest.mark.django_db def test_create_branch_node(self, tenant_acme, acme_tree): branch = OrganizationNode.objects.create( tenant=tenant_acme, parent=acme_tree["root"], name="Branch A", type=OrganizationNodeType.BRANCH, ) assert branch.type == "branch" @pytest.mark.django_db def test_create_warehouse_node(self, tenant_acme, acme_tree): wh = OrganizationNode.objects.create( tenant=tenant_acme, parent=acme_tree["root"], name="WH-01", type=OrganizationNodeType.WAREHOUSE, ) assert wh.type == "warehouse" @pytest.mark.django_db def test_all_six_choices_available(self): expected = {"organization", "holding", "company", "branch", "department", "warehouse"} actual = {c.value for c in OrganizationNodeType} assert expected == actual # --------------------------------------------------------------------------- # WorkspaceTemplate model # --------------------------------------------------------------------------- class TestWorkspaceTemplateModel: @pytest.mark.django_db def test_create_template(self, basic_template): assert WorkspaceTemplate.objects.count() == 1 assert basic_template.slug == "support-supervisor" @pytest.mark.django_db def test_str_representation(self, basic_template): assert str(basic_template) == "support-supervisor" @pytest.mark.django_db def test_is_system_defaults_false(self, basic_template): assert basic_template.is_system is False @pytest.mark.django_db def test_navigation_schema_stored_as_list(self, basic_template): assert isinstance(basic_template.navigation_schema, list) assert len(basic_template.navigation_schema) == 2 @pytest.mark.django_db def test_slug_uniqueness(self, basic_template): with pytest.raises(Exception): # noqa: B017 WorkspaceTemplate.objects.create(slug="support-supervisor", name="Dup") @pytest.mark.django_db def test_list_workspace_templates_service(self, basic_template): results = list(list_workspace_templates()) assert len(results) == 1 assert results[0].slug == "support-supervisor" @pytest.mark.django_db def test_multiple_templates_ordered_by_name(self, db): WorkspaceTemplate.objects.create(slug="z-tmpl", name="Z Template") WorkspaceTemplate.objects.create(slug="a-tmpl", name="A Template") names = [t.name for t in list_workspace_templates()] assert names == sorted(names) # --------------------------------------------------------------------------- # create_workspace_from_template # --------------------------------------------------------------------------- class TestCreateWorkspaceFromTemplate: @pytest.mark.django_db def test_creates_workspace_with_template_attrs(self, tenant_acme, acme_tree, basic_template): ws = create_workspace_from_template(tenant_acme, "support-supervisor") assert ws.slug == "support-supervisor" assert ws.name == "Support Supervisor" assert ws.icon == "headset" assert ws.default_module == "helpdesk" assert ws.tenant_id == tenant_acme.pk @pytest.mark.django_db def test_creates_nav_items_from_schema(self, tenant_acme, acme_tree, basic_template): from simorgh.apps.workspaces.models import NavigationItem ws = create_workspace_from_template(tenant_acme, "support-supervisor") items = NavigationItem.objects.filter(workspace=ws).order_by("order") assert [i.key for i in items] == ["tickets", "reports"] @pytest.mark.django_db def test_custom_name_and_slug_override(self, tenant_acme, acme_tree, basic_template): ws = create_workspace_from_template( tenant_acme, "support-supervisor", name="Custom Name", slug="custom-slug", ) assert ws.slug == "custom-slug" assert ws.name == "Custom Name" @pytest.mark.django_db def test_raises_for_unknown_template(self, tenant_acme, acme_tree): with pytest.raises(WorkspaceTemplate.DoesNotExist): create_workspace_from_template(tenant_acme, "no-such-template") # --------------------------------------------------------------------------- # UserOrgContextPreference model + services # --------------------------------------------------------------------------- class TestUserOrgContextPreference: @pytest.mark.django_db def test_get_or_create_preference(self, alice, tenant_acme, acme_tree): pref = get_user_org_context(alice, tenant_acme) assert pref.user_id == alice.pk assert pref.tenant_id == tenant_acme.pk assert pref.active_organization_node is None @pytest.mark.django_db def test_idempotent_get(self, alice, tenant_acme, acme_tree): p1 = get_user_org_context(alice, tenant_acme) p2 = get_user_org_context(alice, tenant_acme) assert p1.pk == p2.pk @pytest.mark.django_db def test_unique_per_user_tenant(self, alice, tenant_acme, acme_tree): get_user_org_context(alice, tenant_acme) with pytest.raises(Exception): # noqa: B017 UserOrgContextPreference.objects.create(user=alice, tenant=tenant_acme) @pytest.mark.django_db def test_str_representation(self, alice, tenant_acme, acme_tree): pref = get_user_org_context(alice, tenant_acme) assert "u:" in str(pref) assert "t:" in str(pref) @pytest.mark.django_db def test_switch_active_node(self, alice, tenant_acme, acme_tree): pref = switch_active_node(alice, tenant_acme, node_id=acme_tree["eu"].pk) assert pref.active_organization_node_id == acme_tree["eu"].pk @pytest.mark.django_db def test_switch_active_node_to_none_clears(self, alice, tenant_acme, acme_tree): switch_active_node(alice, tenant_acme, node_id=acme_tree["eu"].pk) pref = switch_active_node(alice, tenant_acme, node_id=None) assert pref.active_organization_node is None @pytest.mark.django_db def test_switch_node_rejects_wrong_tenant(self, alice, tenant_acme, tenant_globex, acme_tree): with pytest.raises(Exception): # noqa: B017 switch_active_node(alice, tenant_globex, node_id=acme_tree["eu"].pk) @pytest.mark.django_db def test_switch_active_workspace(self, alice, tenant_acme, acme_tree): ws = create_workspace(tenant_acme, slug="main", name="Main") pref = switch_active_workspace(alice, tenant_acme, workspace_slug="main") assert pref.active_workspace_id == ws.pk @pytest.mark.django_db def test_switch_active_workspace_to_none(self, alice, tenant_acme, acme_tree): create_workspace(tenant_acme, slug="main", name="Main") switch_active_workspace(alice, tenant_acme, workspace_slug="main") pref = switch_active_workspace(alice, tenant_acme, workspace_slug=None) assert pref.active_workspace is None # --------------------------------------------------------------------------- # Context API (GET/POST /api/v1/workspaces/context/) # --------------------------------------------------------------------------- class TestContextAPI: @pytest.mark.django_db def test_get_context_unauthenticated(self, api_client, tenant_acme): resp = api_client.get("/api/v1/workspaces/context/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code in (401, 403) @pytest.mark.django_db def test_get_context_empty(self, api_client, alice, alice_admin_membership, tenant_acme): sync_registry_to_db() api_client.force_login(alice) resp = api_client.get("/api/v1/workspaces/context/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code == 200, resp.content body = resp.json() assert "active_organization_node" in body assert body["active_organization_node"] is None assert "available_nodes" in body @pytest.mark.django_db def test_post_switch_node( self, api_client, alice, alice_admin_membership, tenant_acme, acme_tree, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/context/", data=json.dumps({"node_id": acme_tree["eu"].pk}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200, resp.content body = resp.json() assert body["active_organization_node"]["id"] == acme_tree["eu"].pk assert body["active_organization_node"]["type"] == "organization" @pytest.mark.django_db def test_post_clear_node( self, api_client, alice, alice_admin_membership, tenant_acme, acme_tree, ): sync_registry_to_db() api_client.force_login(alice) # Set first api_client.post( "/api/v1/workspaces/context/", data=json.dumps({"node_id": acme_tree["eu"].pk}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) # Clear resp = api_client.post( "/api/v1/workspaces/context/", data=json.dumps({"node_id": None}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200, resp.content assert resp.json()["active_organization_node"] is None @pytest.mark.django_db def test_post_switch_workspace( self, api_client, alice, alice_admin_membership, tenant_acme, acme_tree, ): sync_registry_to_db() ws = create_workspace(tenant_acme, slug="main", name="Main") add_member(ws, alice) api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/context/", data=json.dumps({"workspace_slug": "main"}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200, resp.content body = resp.json() assert body["active_workspace"]["slug"] == "main" @pytest.mark.django_db def test_post_context_missing_fields( self, api_client, alice, alice_admin_membership, tenant_acme, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/context/", data=json.dumps({"unrelated": "data"}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code in (400, 422) @pytest.mark.django_db def test_available_nodes_in_response( self, api_client, alice, alice_admin_membership, tenant_acme, acme_tree, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.get("/api/v1/workspaces/context/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code == 200 nodes = resp.json()["available_nodes"] assert isinstance(nodes, list) assert len(nodes) >= 4 # root, eu, de, us from acme_tree # --------------------------------------------------------------------------- # Template list API (GET /api/v1/workspaces/templates/) # --------------------------------------------------------------------------- class TestWorkspaceTemplateAPI: @pytest.mark.django_db def test_list_templates_unauthenticated(self, api_client, tenant_acme): resp = api_client.get("/api/v1/workspaces/templates/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code in (401, 403) @pytest.mark.django_db def test_list_templates_empty( self, api_client, alice, alice_admin_membership, tenant_acme, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.get("/api/v1/workspaces/templates/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code == 200 assert resp.json()["results"] == [] @pytest.mark.django_db def test_list_templates_returns_all( self, api_client, alice, alice_admin_membership, tenant_acme, basic_template, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.get("/api/v1/workspaces/templates/", HTTP_X_TENANT=tenant_acme.slug) assert resp.status_code == 200 results = resp.json()["results"] assert len(results) == 1 assert results[0]["slug"] == "support-supervisor" assert results[0]["navigation_schema"] is not None @pytest.mark.django_db def test_template_response_shape( self, api_client, alice, alice_admin_membership, tenant_acme, basic_template, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.get("/api/v1/workspaces/templates/", HTTP_X_TENANT=tenant_acme.slug) t = resp.json()["results"][0] for field in ("slug", "name", "description", "kind", "icon", "theme", "default_module", "navigation_schema", "is_system"): assert field in t, f"missing field: {field}" # --------------------------------------------------------------------------- # Workspace-from-template API (POST /api/v1/workspaces/from-template//) # --------------------------------------------------------------------------- class TestWorkspaceFromTemplateAPI: @pytest.mark.django_db def test_requires_authentication(self, api_client, tenant_acme, basic_template): resp = api_client.post( "/api/v1/workspaces/from-template/support-supervisor/", content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code in (401, 403) @pytest.mark.django_db def test_requires_manage_permission( self, api_client, alice, alice_membership, tenant_acme, basic_template, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/from-template/support-supervisor/", content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403 @pytest.mark.django_db def test_create_workspace_from_template( self, api_client, alice, alice_admin_membership, tenant_acme, basic_template, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/from-template/support-supervisor/", content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 201, resp.content body = resp.json() assert body["slug"] == "support-supervisor" @pytest.mark.django_db def test_create_with_overrides( self, api_client, alice, alice_admin_membership, tenant_acme, basic_template, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/from-template/support-supervisor/", data=json.dumps({"name": "My Workspace", "slug": "my-ws"}), content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 201, resp.content assert resp.json()["slug"] == "my-ws" @pytest.mark.django_db def test_not_found_for_missing_template( self, api_client, alice, alice_admin_membership, tenant_acme, ): sync_registry_to_db() api_client.force_login(alice) resp = api_client.post( "/api/v1/workspaces/from-template/does-not-exist/", content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 404 @pytest.mark.django_db def test_creates_navigation_items( self, api_client, alice, alice_admin_membership, tenant_acme, basic_template, ): from simorgh.apps.workspaces.models import NavigationItem sync_registry_to_db() api_client.force_login(alice) api_client.post( "/api/v1/workspaces/from-template/support-supervisor/", content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) ws = Workspace.objects.get(tenant=tenant_acme, slug="support-supervisor") items = NavigationItem.objects.filter(workspace=ws) assert items.count() == 2