758cd61621
Service layer (production code): - branch_manager: set account_id on SessionBranch (root + fork) and ForkPoint from session.account_id; load session in create_fork for this purpose - handoff_manager: set account_id on SessionHandoff from session.account_id - ai_suggestions endpoint: set account_id on AISuggestion from current_user - steps endpoint (/feedback): set account_id on StepRating from current_user - ratings endpoint: set account_id on StepRating from current_user Test infrastructure: - conftest.py: seed PLATFORM_ACCOUNT_ID (00000000-...-0001) account after Base.metadata.create_all so global categories and gallery items have a valid FK - test_rls_isolation: add _ensure_rls_schema fixture that runs 'alembic upgrade head' before module tests — previous function-scoped test_db fixtures drop the schema, leaving the RLS tests with no tables - test_branding: create Account before User in helper functions - test_admin_gallery: set account_id=PLATFORM_ACCOUNT_ID on Tree/ScriptTemplate - test_public_templates: set account_id=PLATFORM_ACCOUNT_ID on Tree, ScriptTemplate, TreeCategory - test_resolution_outputs: set account_id=session.account_id on SessionResolutionOutput - test_analytics_phase5: set account_id on PsaPostLog - test_draft_trees: replace account_id=None with PLATFORM_ACCOUNT_ID in migration default test (NOT NULL now enforced) - test_maintenance_schedules: set account_id on other_tree - test_save_session_as_tree: set account_id on all 5 Session() constructors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
316 lines
10 KiB
Python
316 lines
10 KiB
Python
"""Integration tests for admin gallery curation endpoints."""
|
|
|
|
import uuid
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.tree import Tree
|
|
from app.models.script_template import ScriptTemplate, ScriptCategory
|
|
|
|
_PLATFORM_ACCOUNT_ID = uuid.UUID("00000000-0000-0000-0000-000000000001")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _create_tree(db: AsyncSession, admin_user_id: str) -> Tree:
|
|
"""Insert a minimal Tree directly into the DB."""
|
|
tree = Tree(
|
|
id=uuid.uuid4(),
|
|
name="Gallery Test Flow",
|
|
tree_type="troubleshooting",
|
|
visibility="public",
|
|
account_id=_PLATFORM_ACCOUNT_ID,
|
|
is_gallery_featured=False,
|
|
gallery_sort_order=0,
|
|
tree_structure={
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Test?",
|
|
"options": [],
|
|
"children": [],
|
|
},
|
|
author_id=uuid.UUID(admin_user_id),
|
|
)
|
|
db.add(tree)
|
|
await db.commit()
|
|
await db.refresh(tree)
|
|
return tree
|
|
|
|
|
|
async def _create_script(db: AsyncSession, admin_user_id: str) -> ScriptTemplate:
|
|
"""Insert a minimal ScriptTemplate directly into the DB."""
|
|
# Need a category first
|
|
category = ScriptCategory(
|
|
id=uuid.uuid4(),
|
|
name="Test Category",
|
|
slug=f"test-category-{uuid.uuid4().hex[:6]}",
|
|
)
|
|
db.add(category)
|
|
await db.flush()
|
|
|
|
script = ScriptTemplate(
|
|
id=uuid.uuid4(),
|
|
category_id=category.id,
|
|
account_id=_PLATFORM_ACCOUNT_ID,
|
|
name="Gallery Test Script",
|
|
slug=f"gallery-test-script-{uuid.uuid4().hex[:6]}",
|
|
script_body="Write-Host 'Test'",
|
|
is_gallery_featured=False,
|
|
gallery_sort_order=0,
|
|
created_by=uuid.UUID(admin_user_id) if admin_user_id else None,
|
|
)
|
|
db.add(script)
|
|
await db.commit()
|
|
await db.refresh(script)
|
|
return script
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAdminGallery:
|
|
"""Test suite for admin gallery curation endpoints."""
|
|
|
|
# -- Auth guard tests --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_featured_list_requires_admin(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Non-admin users get 403 on featured list."""
|
|
response = await client.get("/api/v1/admin/gallery/featured", headers=auth_headers)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feature_toggle_flow_requires_admin(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Non-admin users get 403 when trying to toggle flow feature."""
|
|
fake_id = str(uuid.uuid4())
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/flows/{fake_id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feature_toggle_script_requires_admin(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Non-admin users get 403 when trying to toggle script feature."""
|
|
fake_id = str(uuid.uuid4())
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/scripts/{fake_id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
# -- Feature toggle tests --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_flow_featured_on(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""Admin can feature a flow."""
|
|
tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/flows/{tree.id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_gallery_featured"] is True
|
|
assert data["id"] == str(tree.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_flow_featured_off(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""Admin can un-feature a previously featured flow."""
|
|
tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
# Feature it first
|
|
tree.is_gallery_featured = True
|
|
await test_db.commit()
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/flows/{tree.id}/feature",
|
|
json={"is_gallery_featured": False},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["is_gallery_featured"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_flow_not_found(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
"""Returns 404 when flow ID doesn't exist."""
|
|
fake_id = str(uuid.uuid4())
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/flows/{fake_id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_script_featured_on(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""Admin can feature a script."""
|
|
script = await _create_script(test_db, test_admin["user_data"]["id"])
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/scripts/{script.id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_gallery_featured"] is True
|
|
assert data["id"] == str(script.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_script_not_found(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
"""Returns 404 when script ID doesn't exist."""
|
|
fake_id = str(uuid.uuid4())
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/scripts/{fake_id}/feature",
|
|
json={"is_gallery_featured": True},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
# -- Sort order tests --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_flow_sort_order(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""Admin can update gallery sort order for a flow."""
|
|
tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/flows/{tree.id}/sort-order",
|
|
json={"gallery_sort_order": 42},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["gallery_sort_order"] == 42
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_script_sort_order(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""Admin can update gallery sort order for a script."""
|
|
script = await _create_script(test_db, test_admin["user_data"]["id"])
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/admin/gallery/scripts/{script.id}/sort-order",
|
|
json={"gallery_sort_order": 7},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["gallery_sort_order"] == 7
|
|
|
|
# -- Featured list tests --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_featured_list_returns_only_featured(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""GET /featured returns only items where is_gallery_featured=True."""
|
|
# Create two flows: one featured, one not
|
|
featured_tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
_unfeatured_tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
_unfeatured_tree.name = "Unfeatured Flow"
|
|
|
|
# Feature only the first
|
|
featured_tree.is_gallery_featured = True
|
|
await test_db.commit()
|
|
|
|
# Create two scripts: one featured, one not
|
|
featured_script = await _create_script(test_db, test_admin["user_data"]["id"])
|
|
_unfeatured_script = await _create_script(test_db, test_admin["user_data"]["id"])
|
|
featured_script.is_gallery_featured = True
|
|
await test_db.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/admin/gallery/featured", headers=admin_auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
flow_ids = [f["id"] for f in data["flows"]]
|
|
script_ids = [s["id"] for s in data["scripts"]]
|
|
|
|
assert str(featured_tree.id) in flow_ids
|
|
assert str(_unfeatured_tree.id) not in flow_ids
|
|
assert str(featured_script.id) in script_ids
|
|
assert str(_unfeatured_script.id) not in script_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_items_list_requires_admin(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Non-admin users get 403 on all items list."""
|
|
response = await client.get("/api/v1/admin/gallery/items", headers=auth_headers)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_items_list_returns_all(
|
|
self,
|
|
client: AsyncClient,
|
|
admin_auth_headers: dict,
|
|
test_admin: dict,
|
|
test_db: AsyncSession,
|
|
):
|
|
"""GET /items returns all flows and scripts regardless of featured status."""
|
|
tree = await _create_tree(test_db, test_admin["user_data"]["id"])
|
|
|
|
response = await client.get(
|
|
"/api/v1/admin/gallery/items", headers=admin_auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "flows" in data
|
|
assert "scripts" in data
|
|
flow_ids = [f["id"] for f in data["flows"]]
|
|
assert str(tree.id) in flow_ids
|