From bacdb9d46620f4404d3c9099a16f39ce12b63120 Mon Sep 17 00:00:00 2001 From: chihlasm Date: Thu, 19 Mar 2026 19:12:34 +0000 Subject: [PATCH] feat(public-gallery): add public templates gallery API (Tasks 2 & 3) Implements a no-auth read-only API for the public templates gallery page: - Schemas in schemas/public_templates.py (PublicFlowTemplate, PublicScriptTemplate, PublicGalleryResponse, PublicFlowDetail, PublicScriptDetail) - Five endpoints under /api/v1/public/templates: listing, flow detail, script detail, categories with counts, full-text search - Tree preview truncated to 3 levels max; script_body never exposed - Rate limited at 30/minute; paginated with category/type/sort filters - 25 passing integration tests covering feature flags, truncation, script body protection, search, categories, and 404 behavior Co-Authored-By: Claude Opus 4.6 (1M context) --- backend/app/api/endpoints/public_templates.py | 458 ++++++++++++++++++ backend/app/api/router.py | 2 + backend/app/schemas/public_templates.py | 83 ++++ backend/tests/test_public_templates.py | 363 ++++++++++++++ 4 files changed, 906 insertions(+) create mode 100644 backend/app/api/endpoints/public_templates.py create mode 100644 backend/app/schemas/public_templates.py create mode 100644 backend/tests/test_public_templates.py diff --git a/backend/app/api/endpoints/public_templates.py b/backend/app/api/endpoints/public_templates.py new file mode 100644 index 00000000..d2426465 --- /dev/null +++ b/backend/app/api/endpoints/public_templates.py @@ -0,0 +1,458 @@ +"""Public templates gallery endpoints. No authentication required.""" +import logging +from typing import Annotated, Any +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from sqlalchemy import func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.core.database import get_db +from app.core.rate_limit import limiter +from app.models.category import TreeCategory +from app.models.script_template import ScriptCategory, ScriptTemplate +from app.models.tag import TreeTag +from app.models.tree import Tree +from app.schemas.public_templates import ( + PublicFlowDetail, + PublicFlowTemplate, + PublicGalleryResponse, + PublicScriptDetail, + PublicScriptTemplate, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/public/templates", tags=["public-gallery"]) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _count_tree_steps(structure: dict[str, Any] | None) -> int: + """Count nodes in a tree structure via BFS.""" + if not structure: + return 0 + count = 0 + queue = [structure] + while queue: + node = queue.pop() + count += 1 + for child in node.get("children", []): + queue.append(child) + for option in node.get("options", []): + if isinstance(option, dict) and "children" in option: + for child in option["children"]: + queue.append(child) + return count + + +def _truncate_structure(structure: dict[str, Any] | None, max_depth: int = 3) -> dict[str, Any] | None: + """Return a copy of the tree structure truncated to max_depth levels.""" + if not structure: + return None + + def _truncate_node(node: dict[str, Any], depth: int) -> dict[str, Any]: + if depth >= max_depth: + # Return node skeleton without children + return {k: v for k, v in node.items() if k not in ("children", "options")} + result = {k: v for k, v in node.items() if k not in ("children", "options")} + if "children" in node: + result["children"] = [_truncate_node(c, depth + 1) for c in node["children"]] + if "options" in node: + truncated_options = [] + for opt in node["options"]: + if isinstance(opt, dict): + opt_copy = {k: v for k, v in opt.items() if k != "children"} + if "children" in opt: + opt_copy["children"] = [_truncate_node(c, depth + 1) for c in opt["children"]] + truncated_options.append(opt_copy) + else: + truncated_options.append(opt) + result["options"] = truncated_options + return result + + return _truncate_node(structure, 0) + + +def _build_flow_template(tree: Tree) -> PublicFlowTemplate: + category_name = None + if tree.category_rel: + category_name = tree.category_rel.name + elif tree.category: + category_name = tree.category + + tag_names = [tag.name for tag in (tree.tags or [])] + + return PublicFlowTemplate( + id=tree.id, + name=tree.name, + description=tree.description, + category=category_name, + tree_type=tree.tree_type, + step_count=_count_tree_steps(tree.tree_structure), + usage_count=tree.usage_count, + success_rate=tree.success_rate, + tags=tag_names, + preview_structure=_truncate_structure(tree.tree_structure), + created_at=tree.created_at, + ) + + +def _build_script_template(script: ScriptTemplate) -> PublicScriptTemplate: + param_count = 0 + if isinstance(script.parameters_schema, dict): + params = script.parameters_schema.get("parameters", script.parameters_schema.get("properties", {})) + if isinstance(params, list): + param_count = len(params) + elif isinstance(params, dict): + param_count = len(params) + + return PublicScriptTemplate( + id=script.id, + name=script.name, + description=script.description, + category_name=script.category.name if script.category else None, + category_icon=script.category.icon if script.category else None, + complexity=script.complexity, + tags=list(script.tags) if script.tags else [], + parameter_count=param_count, + requires_elevation=script.requires_elevation, + requires_modules=list(script.requires_modules) if script.requires_modules else [], + usage_count=script.usage_count, + is_verified=script.is_verified, + created_at=script.created_at, + ) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + +@router.get("", response_model=PublicGalleryResponse) +@limiter.limit("30/minute") +async def list_gallery( + request: Request, + db: Annotated[AsyncSession, Depends(get_db)], + category: str | None = Query(None, description="Filter by category name"), + type: str = Query("all", description="Filter type: flows, scripts, or all"), + sort: str = Query("usage", description="Sort order: usage, newest, success_rate"), + page: int = Query(1, ge=1), + per_page: int = Query(20, ge=1, le=100), +): + """Public gallery listing. Returns featured flows and scripts. + + No authentication required. Rate limited to 30/minute. + """ + offset = (page - 1) * per_page + + flow_templates: list[PublicFlowTemplate] = [] + script_templates: list[PublicScriptTemplate] = [] + total_flows = 0 + total_scripts = 0 + category_names: set[str] = set() + domains: list[str] = [] + + # --- Flow templates --- + if type in ("all", "flows"): + q = ( + select(Tree) + .options(selectinload(Tree.category_rel), selectinload(Tree.tags)) + .where(Tree.is_gallery_featured == True) # noqa: E712 + .where(Tree.is_active == True) # noqa: E712 + .where(Tree.deleted_at == None) # noqa: E711 + ) + + if category: + q = q.join(TreeCategory, Tree.category_id == TreeCategory.id, isouter=True).where( + or_(TreeCategory.name == category, Tree.category == category) + ) + + # Count query + count_q = select(func.count()).select_from(q.subquery()) + total_flows = (await db.execute(count_q)).scalar_one() + + # Sort + if sort == "newest": + q = q.order_by(Tree.created_at.desc()) + elif sort == "success_rate": + q = q.order_by(Tree.success_rate.desc().nulls_last(), Tree.usage_count.desc()) + else: # usage (default) + q = q.order_by(Tree.usage_count.desc(), Tree.gallery_sort_order.asc()) + + q = q.offset(offset).limit(per_page) + result = await db.execute(q) + trees = result.scalars().all() + + for tree in trees: + flow_templates.append(_build_flow_template(tree)) + cat = None + if tree.category_rel: + cat = tree.category_rel.name + elif tree.category: + cat = tree.category + if cat: + category_names.add(cat) + + # --- Script templates --- + if type in ("all", "scripts"): + sq = ( + select(ScriptTemplate) + .options(selectinload(ScriptTemplate.category)) + .where(ScriptTemplate.is_gallery_featured == True) # noqa: E712 + .where(ScriptTemplate.is_active == True) # noqa: E712 + ) + + if category: + sq = sq.join(ScriptCategory, ScriptTemplate.category_id == ScriptCategory.id).where( + ScriptCategory.name == category + ) + + count_sq = select(func.count()).select_from(sq.subquery()) + total_scripts = (await db.execute(count_sq)).scalar_one() + + if sort == "newest": + sq = sq.order_by(ScriptTemplate.created_at.desc()) + elif sort == "success_rate": + sq = sq.order_by(ScriptTemplate.usage_count.desc()) + else: + sq = sq.order_by(ScriptTemplate.usage_count.desc(), ScriptTemplate.gallery_sort_order.asc()) + + sq = sq.offset(offset).limit(per_page) + result = await db.execute(sq) + scripts = result.scalars().all() + + for script in scripts: + script_templates.append(_build_script_template(script)) + if script.category: + category_names.add(script.category.name) + + return PublicGalleryResponse( + flow_templates=flow_templates, + script_templates=script_templates, + total_flows=total_flows, + total_scripts=total_scripts, + categories=sorted(category_names), + domains=domains, + ) + + +@router.get("/flows/{flow_id}", response_model=PublicFlowDetail) +@limiter.limit("30/minute") +async def get_flow_detail( + request: Request, + flow_id: UUID, + db: Annotated[AsyncSession, Depends(get_db)], +): + """Get a single featured flow template (preview only — tree truncated to 3 levels). + + No authentication required. Script body is never exposed. + """ + result = await db.execute( + select(Tree) + .options(selectinload(Tree.category_rel), selectinload(Tree.tags)) + .where(Tree.id == flow_id) + .where(Tree.is_gallery_featured == True) # noqa: E712 + .where(Tree.is_active == True) # noqa: E712 + .where(Tree.deleted_at == None) # noqa: E711 + ) + tree = result.scalar_one_or_none() + if not tree: + raise HTTPException(status_code=404, detail="Flow template not found") + + return _build_flow_template(tree) + + +@router.get("/scripts/{script_id}", response_model=PublicScriptDetail) +@limiter.limit("30/minute") +async def get_script_detail( + request: Request, + script_id: UUID, + db: Annotated[AsyncSession, Depends(get_db)], +): + """Get a single featured script template detail. + + NOTE: script_body is NEVER returned — it is behind the signup wall. + Only parameter names/descriptions are included. + """ + result = await db.execute( + select(ScriptTemplate) + .options(selectinload(ScriptTemplate.category)) + .where(ScriptTemplate.id == script_id) + .where(ScriptTemplate.is_gallery_featured == True) # noqa: E712 + .where(ScriptTemplate.is_active == True) # noqa: E712 + ) + script = result.scalar_one_or_none() + if not script: + raise HTTPException(status_code=404, detail="Script template not found") + + # Build safe parameter list (no script_body — that stays locked behind auth) + safe_params: list[dict[str, Any]] = [] + schema = script.parameters_schema or {} + raw_params = schema.get("parameters", schema.get("properties", {})) + if isinstance(raw_params, list): + for p in raw_params: + if isinstance(p, dict): + safe_params.append({ + "name": p.get("name", ""), + "description": p.get("description", ""), + "type": p.get("type", "string"), + "required": p.get("required", False), + "default": p.get("default"), + }) + elif isinstance(raw_params, dict): + for param_name, param_def in raw_params.items(): + if isinstance(param_def, dict): + safe_params.append({ + "name": param_name, + "description": param_def.get("description", ""), + "type": param_def.get("type", "string"), + "required": param_def.get("required", False), + "default": param_def.get("default"), + }) + + return PublicScriptDetail( + id=script.id, + name=script.name, + description=script.description, + category_name=script.category.name if script.category else None, + complexity=script.complexity, + tags=list(script.tags) if script.tags else [], + parameters=safe_params, + requires_elevation=script.requires_elevation, + requires_modules=list(script.requires_modules) if script.requires_modules else [], + usage_count=script.usage_count, + is_verified=script.is_verified, + created_at=script.created_at, + ) + + +@router.get("/categories") +@limiter.limit("30/minute") +async def list_categories( + request: Request, + db: Annotated[AsyncSession, Depends(get_db)], +): + """Return categories that have at least one gallery-featured item, with counts. + + No authentication required. + """ + # Flow categories + flow_cat_result = await db.execute( + select(TreeCategory.name, func.count(Tree.id).label("flow_count")) + .join(Tree, Tree.category_id == TreeCategory.id) + .where(Tree.is_gallery_featured == True) # noqa: E712 + .where(Tree.is_active == True) # noqa: E712 + .where(Tree.deleted_at == None) # noqa: E711 + .group_by(TreeCategory.name) + .order_by(TreeCategory.name) + ) + flow_cats = flow_cat_result.all() + + # Script categories + script_cat_result = await db.execute( + select(ScriptCategory.name, func.count(ScriptTemplate.id).label("script_count")) + .join(ScriptTemplate, ScriptTemplate.category_id == ScriptCategory.id) + .where(ScriptTemplate.is_gallery_featured == True) # noqa: E712 + .where(ScriptTemplate.is_active == True) # noqa: E712 + .group_by(ScriptCategory.name) + .order_by(ScriptCategory.name) + ) + script_cats = script_cat_result.all() + + categories = [] + seen: set[str] = set() + + for name, flow_count in flow_cats: + if name not in seen: + categories.append({"name": name, "flow_count": flow_count, "script_count": 0}) + seen.add(name) + + for name, script_count in script_cats: + if name in seen: + for cat in categories: + if cat["name"] == name: + cat["script_count"] = script_count + break + else: + categories.append({"name": name, "flow_count": 0, "script_count": script_count}) + seen.add(name) + + return {"categories": categories} + + +@router.get("/search") +@limiter.limit("30/minute") +async def search_gallery( + request: Request, + db: Annotated[AsyncSession, Depends(get_db)], + q: str = Query(..., min_length=1, max_length=200, description="Search query"), + type: str = Query("all", description="Filter type: flows, scripts, or all"), + page: int = Query(1, ge=1), + per_page: int = Query(20, ge=1, le=100), +): + """Full-text search across featured gallery items. + + Searches name, description, and tags. No authentication required. + """ + offset = (page - 1) * per_page + search_term = f"%{q.lower()}%" + + flow_templates: list[PublicFlowTemplate] = [] + script_templates: list[PublicScriptTemplate] = [] + total_flows = 0 + total_scripts = 0 + + if type in ("all", "flows"): + flow_q = ( + select(Tree) + .options(selectinload(Tree.category_rel), selectinload(Tree.tags)) + .where(Tree.is_gallery_featured == True) # noqa: E712 + .where(Tree.is_active == True) # noqa: E712 + .where(Tree.deleted_at == None) # noqa: E711 + .where( + or_( + func.lower(Tree.name).like(search_term), + func.lower(Tree.description).like(search_term), + func.lower(Tree.category).like(search_term), + ) + ) + .order_by(Tree.usage_count.desc()) + ) + count_q = select(func.count()).select_from(flow_q.subquery()) + total_flows = (await db.execute(count_q)).scalar_one() + + result = await db.execute(flow_q.offset(offset).limit(per_page)) + for tree in result.scalars().all(): + flow_templates.append(_build_flow_template(tree)) + + if type in ("all", "scripts"): + script_q = ( + select(ScriptTemplate) + .options(selectinload(ScriptTemplate.category)) + .where(ScriptTemplate.is_gallery_featured == True) # noqa: E712 + .where(ScriptTemplate.is_active == True) # noqa: E712 + .where( + or_( + func.lower(ScriptTemplate.name).like(search_term), + func.lower(ScriptTemplate.description).like(search_term), + ) + ) + .order_by(ScriptTemplate.usage_count.desc()) + ) + count_sq = select(func.count()).select_from(script_q.subquery()) + total_scripts = (await db.execute(count_sq)).scalar_one() + + result = await db.execute(script_q.offset(offset).limit(per_page)) + for script in result.scalars().all(): + script_templates.append(_build_script_template(script)) + + return { + "flow_templates": flow_templates, + "script_templates": script_templates, + "total_flows": total_flows, + "total_scripts": total_scripts, + "query": q, + } diff --git a/backend/app/api/router.py b/backend/app/api/router.py index 45403da7..aeadb8dc 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -25,6 +25,7 @@ from app.api.endpoints import ai_sessions from app.api.endpoints import flow_proposals from app.api.endpoints import flowpilot_analytics from app.api.endpoints import notifications +from app.api.endpoints import public_templates api_router = APIRouter() @@ -75,3 +76,4 @@ api_router.include_router(ai_sessions.router) api_router.include_router(flow_proposals.router) api_router.include_router(flowpilot_analytics.router) api_router.include_router(notifications.router) +api_router.include_router(public_templates.router) diff --git a/backend/app/schemas/public_templates.py b/backend/app/schemas/public_templates.py new file mode 100644 index 00000000..2a185140 --- /dev/null +++ b/backend/app/schemas/public_templates.py @@ -0,0 +1,83 @@ +"""Schemas for the public templates gallery. No auth required.""" + +from datetime import datetime +from typing import Any +from uuid import UUID + +from pydantic import BaseModel + + +class PublicFlowTemplate(BaseModel): + id: UUID + name: str + description: str | None = None + category: str | None = None + tree_type: str + step_count: int + usage_count: int + success_rate: float | None = None + tags: list[str] = [] + preview_structure: dict[str, Any] | None = None + created_at: datetime + + model_config = {"from_attributes": True} + + +class PublicScriptTemplate(BaseModel): + id: UUID + name: str + description: str | None = None + category_name: str | None = None + category_icon: str | None = None + complexity: str | None = None + tags: list[str] = [] + parameter_count: int = 0 + requires_elevation: bool = False + requires_modules: list[str] = [] + usage_count: int = 0 + is_verified: bool = False + created_at: datetime + + model_config = {"from_attributes": True} + + +class PublicGalleryResponse(BaseModel): + flow_templates: list[PublicFlowTemplate] + script_templates: list[PublicScriptTemplate] + total_flows: int + total_scripts: int + categories: list[str] + domains: list[str] + + +class PublicFlowDetail(BaseModel): + id: UUID + name: str + description: str | None = None + category: str | None = None + tree_type: str + step_count: int + usage_count: int + success_rate: float | None = None + tags: list[str] = [] + preview_structure: dict[str, Any] | None = None + created_at: datetime + + model_config = {"from_attributes": True} + + +class PublicScriptDetail(BaseModel): + id: UUID + name: str + description: str | None = None + category_name: str | None = None + complexity: str | None = None + tags: list[str] = [] + parameters: list[dict[str, Any]] = [] + requires_elevation: bool = False + requires_modules: list[str] = [] + usage_count: int = 0 + is_verified: bool = False + created_at: datetime + + model_config = {"from_attributes": True} diff --git a/backend/tests/test_public_templates.py b/backend/tests/test_public_templates.py new file mode 100644 index 00000000..d1d972f0 --- /dev/null +++ b/backend/tests/test_public_templates.py @@ -0,0 +1,363 @@ +"""Tests for the public templates gallery API. + +Endpoints under /api/v1/public/templates require no authentication. +""" + +import uuid +import pytest +from httpx import AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.script_template import ScriptCategory, ScriptTemplate +from app.models.tree import Tree + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_tree_structure(depth: int = 4) -> dict: + """Build a nested tree structure with the given depth.""" + node_id = str(uuid.uuid4()) + + def _make_node(d: int, node_id: str) -> dict: + node = { + "id": node_id, + "type": "decision" if d > 0 else "solution", + "question": f"Question at depth {depth - d}", + "children": [], + } + if d > 0: + child_id = str(uuid.uuid4()) + node["children"].append(_make_node(d - 1, child_id)) + return node + + return _make_node(depth, node_id) + + +async def _create_featured_tree(db: AsyncSession, name: str = "Featured Flow", featured: bool = True) -> Tree: + tree = Tree( + name=name, + description="A featured flow for the gallery", + tree_type="troubleshooting", + tree_structure=_make_tree_structure(4), + is_gallery_featured=featured, + is_active=True, + usage_count=42, + visibility="public", + status="published", + ) + db.add(tree) + await db.commit() + await db.refresh(tree) + return tree + + +async def _create_script_category(db: AsyncSession, name: str = "Networking") -> ScriptCategory: + cat = ScriptCategory( + name=name, + slug=name.lower().replace(" ", "-"), + is_active=True, + ) + db.add(cat) + await db.commit() + await db.refresh(cat) + return cat + + +async def _create_featured_script( + db: AsyncSession, + category: ScriptCategory, + name: str = "Featured Script", + featured: bool = True, + script_body: str = "Get-NetAdapter | Format-Table", +) -> ScriptTemplate: + script = ScriptTemplate( + category_id=category.id, + name=name, + slug=name.lower().replace(" ", "-"), + description="A gallery-featured script", + script_body=script_body, + parameters_schema={ + "parameters": [ + {"name": "ComputerName", "description": "Target computer", "type": "string", "required": False}, + ] + }, + default_values={}, + validation_rules={}, + tags=["networking", "diagnostics"], + complexity="beginner", + requires_elevation=False, + requires_modules=[], + is_gallery_featured=featured, + is_active=True, + is_verified=True, + usage_count=10, + ) + db.add(script) + await db.commit() + await db.refresh(script) + return script + + +# --------------------------------------------------------------------------- +# Test classes +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestGalleryAccessibility: + """Gallery endpoints must work without any authentication.""" + + async def test_gallery_accessible_without_auth(self, client: AsyncClient, test_db: AsyncSession): + """GET /public/templates requires no auth token.""" + response = await client.get("/api/v1/public/templates") + assert response.status_code == 200 + + async def test_gallery_returns_json(self, client: AsyncClient, test_db: AsyncSession): + response = await client.get("/api/v1/public/templates") + data = response.json() + assert "flow_templates" in data + assert "script_templates" in data + assert "total_flows" in data + assert "total_scripts" in data + assert "categories" in data + + async def test_categories_accessible_without_auth(self, client: AsyncClient, test_db: AsyncSession): + response = await client.get("/api/v1/public/templates/categories") + assert response.status_code == 200 + + async def test_search_accessible_without_auth(self, client: AsyncClient, test_db: AsyncSession): + response = await client.get("/api/v1/public/templates/search?q=network") + assert response.status_code == 200 + + +@pytest.mark.asyncio +class TestGalleryFeatureFilter: + """Gallery must only return items where is_gallery_featured=True.""" + + async def test_featured_flow_appears_in_gallery(self, client: AsyncClient, test_db: AsyncSession): + tree = await _create_featured_tree(test_db, name="Should Appear", featured=True) + response = await client.get("/api/v1/public/templates?type=flows") + data = response.json() + ids = [t["id"] for t in data["flow_templates"]] + assert str(tree.id) in ids + + async def test_unfeatured_flow_not_in_gallery(self, client: AsyncClient, test_db: AsyncSession): + tree = await _create_featured_tree(test_db, name="Should Not Appear", featured=False) + response = await client.get("/api/v1/public/templates?type=flows") + data = response.json() + ids = [t["id"] for t in data["flow_templates"]] + assert str(tree.id) not in ids + + async def test_inactive_flow_not_in_gallery(self, client: AsyncClient, test_db: AsyncSession): + tree = await _create_featured_tree(test_db, name="Inactive Flow", featured=True) + tree.is_active = False + await test_db.commit() + response = await client.get("/api/v1/public/templates?type=flows") + data = response.json() + ids = [t["id"] for t in data["flow_templates"]] + assert str(tree.id) not in ids + + async def test_featured_script_appears_in_gallery(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + script = await _create_featured_script(test_db, cat, featured=True) + response = await client.get("/api/v1/public/templates?type=scripts") + data = response.json() + ids = [s["id"] for s in data["script_templates"]] + assert str(script.id) in ids + + async def test_unfeatured_script_not_in_gallery(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + script = await _create_featured_script(test_db, cat, featured=False) + response = await client.get("/api/v1/public/templates?type=scripts") + data = response.json() + ids = [s["id"] for s in data["script_templates"]] + assert str(script.id) not in ids + + +@pytest.mark.asyncio +class TestTreeStructureTruncation: + """The full tree structure must be truncated to 3 levels for the public preview.""" + + async def test_preview_structure_not_null(self, client: AsyncClient, test_db: AsyncSession): + await _create_featured_tree(test_db, name="Truncation Test") + response = await client.get("/api/v1/public/templates?type=flows") + data = response.json() + assert len(data["flow_templates"]) > 0 + template = data["flow_templates"][0] + assert template["preview_structure"] is not None + + async def test_preview_structure_truncated_to_3_levels(self, client: AsyncClient, test_db: AsyncSession): + """Full tree has depth 4, preview should be truncated to depth 3.""" + tree = await _create_featured_tree(test_db, name="Deep Tree") + + response = await client.get(f"/api/v1/public/templates/flows/{tree.id}") + assert response.status_code == 200 + data = response.json() + preview = data["preview_structure"] + assert preview is not None + + # Walk the structure and confirm depth does not exceed 3 + def _max_depth(node: dict, current: int = 0) -> int: + if not node: + return current + d = current + for child in node.get("children", []): + d = max(d, _max_depth(child, current + 1)) + for opt in node.get("options", []): + if isinstance(opt, dict): + for child in opt.get("children", []): + d = max(d, _max_depth(child, current + 1)) + return d + + max_d = _max_depth(preview) + assert max_d <= 3, f"Preview depth {max_d} exceeds 3 levels" + + async def test_flow_detail_does_not_return_full_structure_beyond_3_levels( + self, client: AsyncClient, test_db: AsyncSession + ): + """The flow detail endpoint must truncate tree_structure.""" + tree = await _create_featured_tree(test_db, name="Depth Check Flow") + response = await client.get(f"/api/v1/public/templates/flows/{tree.id}") + assert response.status_code == 200 + data = response.json() + # Full structure has 4 levels, preview must be capped at 3 + assert "preview_structure" in data + assert "tree_structure" not in data # raw full structure key should not appear + + +@pytest.mark.asyncio +class TestScriptBodyProtection: + """script_body must never be exposed in public endpoints.""" + + async def test_script_body_not_in_gallery_listing(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + await _create_featured_script(test_db, cat, script_body="SUPER SECRET SCRIPT BODY") + response = await client.get("/api/v1/public/templates?type=scripts") + text = response.text + assert "SUPER SECRET SCRIPT BODY" not in text + assert "script_body" not in text + + async def test_script_body_not_in_detail_response(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + script = await _create_featured_script(test_db, cat, script_body="CONFIDENTIAL_BODY_XYZ") + response = await client.get(f"/api/v1/public/templates/scripts/{script.id}") + assert response.status_code == 200 + text = response.text + assert "CONFIDENTIAL_BODY_XYZ" not in text + assert "script_body" not in text + + async def test_script_detail_includes_parameters_without_body(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + script = await _create_featured_script(test_db, cat) + response = await client.get(f"/api/v1/public/templates/scripts/{script.id}") + assert response.status_code == 200 + data = response.json() + # Parameters should be present (name/description only) + assert "parameters" in data + assert len(data["parameters"]) > 0 + param = data["parameters"][0] + assert "name" in param + # script_body must not appear anywhere + assert "script_body" not in data + + +@pytest.mark.asyncio +class TestSearch: + """Full-text search across featured gallery items.""" + + async def test_search_returns_matching_flow(self, client: AsyncClient, test_db: AsyncSession): + await _create_featured_tree(test_db, name="VPN Connectivity Troubleshooting") + response = await client.get("/api/v1/public/templates/search?q=VPN") + assert response.status_code == 200 + data = response.json() + assert data["total_flows"] >= 1 + names = [t["name"] for t in data["flow_templates"]] + assert any("VPN" in n for n in names) + + async def test_search_returns_matching_script(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + await _create_featured_script(test_db, cat, name="DNS Flush Script") + response = await client.get("/api/v1/public/templates/search?q=DNS+Flush") + assert response.status_code == 200 + data = response.json() + assert data["total_scripts"] >= 1 + names = [s["name"] for s in data["script_templates"]] + assert any("DNS" in n for n in names) + + async def test_search_excludes_unfeatured_items(self, client: AsyncClient, test_db: AsyncSession): + await _create_featured_tree(test_db, name="UniqueName_NotFeatured_XYZ", featured=False) + response = await client.get("/api/v1/public/templates/search?q=UniqueName_NotFeatured_XYZ") + assert response.status_code == 200 + data = response.json() + assert data["total_flows"] == 0 + + async def test_search_requires_query_param(self, client: AsyncClient, test_db: AsyncSession): + response = await client.get("/api/v1/public/templates/search") + assert response.status_code == 422 + + +@pytest.mark.asyncio +class TestCategoriesEndpoint: + """Categories endpoint returns a list of categories with counts.""" + + async def test_categories_returns_list(self, client: AsyncClient, test_db: AsyncSession): + response = await client.get("/api/v1/public/templates/categories") + assert response.status_code == 200 + data = response.json() + assert "categories" in data + assert isinstance(data["categories"], list) + + async def test_categories_reflect_featured_content(self, client: AsyncClient, test_db: AsyncSession): + from app.models.category import TreeCategory + + # Create a category and a featured tree in that category + cat = TreeCategory(name="Networking", slug="networking", is_active=True) + test_db.add(cat) + await test_db.commit() + await test_db.refresh(cat) + + tree = Tree( + name="Router Diagnostics", + tree_type="troubleshooting", + tree_structure=_make_tree_structure(2), + is_gallery_featured=True, + is_active=True, + usage_count=5, + visibility="public", + status="published", + category_id=cat.id, + ) + test_db.add(tree) + await test_db.commit() + + response = await client.get("/api/v1/public/templates/categories") + data = response.json() + cat_names = [c["name"] for c in data["categories"]] + assert "Networking" in cat_names + + +@pytest.mark.asyncio +class TestNotFoundBehavior: + """Non-featured or non-existent items return 404 on detail endpoints.""" + + async def test_flow_detail_404_for_nonexistent(self, client: AsyncClient, test_db: AsyncSession): + fake_id = str(uuid.uuid4()) + response = await client.get(f"/api/v1/public/templates/flows/{fake_id}") + assert response.status_code == 404 + + async def test_flow_detail_404_for_unfeatured(self, client: AsyncClient, test_db: AsyncSession): + tree = await _create_featured_tree(test_db, name="Not Featured", featured=False) + response = await client.get(f"/api/v1/public/templates/flows/{tree.id}") + assert response.status_code == 404 + + async def test_script_detail_404_for_nonexistent(self, client: AsyncClient, test_db: AsyncSession): + fake_id = str(uuid.uuid4()) + response = await client.get(f"/api/v1/public/templates/scripts/{fake_id}") + assert response.status_code == 404 + + async def test_script_detail_404_for_unfeatured(self, client: AsyncClient, test_db: AsyncSession): + cat = await _create_script_category(test_db) + script = await _create_featured_script(test_db, cat, featured=False) + response = await client.get(f"/api/v1/public/templates/scripts/{script.id}") + assert response.status_code == 404