feat(H): H-WIKI/H-VER/H-LINK — Wiki plugin (articles, categories, versioning, entity links), migration 0126, 9 routes, 15 tests passing
Check Cross-Plugin Imports / check (push) Has been cancelled

This commit is contained in:
Agent Zero
2026-08-18 11:55:43 +02:00
parent e50f6a601f
commit 396fdf3c9a
8 changed files with 622 additions and 0 deletions
+79
View File
@@ -0,0 +1,79 @@
"""Wiki plugin — articles, categories, versions (H-WIKI, H-VER).
Revision ID: 0126
Revises: 0125
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID
revision = "0126"
down_revision = "0125"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"wiki_categories",
sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
sa.Column("tenant_id", PGUUID(as_uuid=True), nullable=False),
sa.Column("owner_id", PGUUID(as_uuid=True), nullable=True),
sa.Column("name", sa.String(200), nullable=False),
sa.Column("slug", sa.String(200), nullable=False),
sa.Column("description", sa.Text, nullable=True),
sa.Column("parent_id", PGUUID(as_uuid=True), nullable=True),
sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_foreign_key("fk_wiki_cat_parent", "wiki_categories", "wiki_categories", ["parent_id"], ["id"], ondelete="SET NULL")
op.create_index("ix_wiki_cat_tenant", "wiki_categories", ["tenant_id"])
op.create_index("ix_wiki_cat_tenant_slug", "wiki_categories", ["tenant_id", "slug"])
op.create_table(
"wiki_articles",
sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
sa.Column("tenant_id", PGUUID(as_uuid=True), nullable=False),
sa.Column("owner_id", PGUUID(as_uuid=True), nullable=True),
sa.Column("title", sa.String(300), nullable=False),
sa.Column("slug", sa.String(300), nullable=False),
sa.Column("content", sa.Text, nullable=False, server_default=""),
sa.Column("content_html", sa.Text, nullable=True),
sa.Column("summary", sa.Text, nullable=True),
sa.Column("category_id", PGUUID(as_uuid=True), nullable=True),
sa.Column("tags", JSONB, nullable=False, server_default="[]"),
sa.Column("status", sa.String(20), nullable=False, server_default="draft"),
sa.Column("entity_links", JSONB, nullable=False, server_default="[]"),
sa.Column("version", sa.Integer, nullable=False, server_default="1"),
sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_foreign_key("fk_wiki_art_category", "wiki_articles", "wiki_categories", ["category_id"], ["id"], ondelete="SET NULL")
op.create_index("ix_wiki_art_tenant", "wiki_articles", ["tenant_id"])
op.create_index("ix_wiki_art_tenant_category", "wiki_articles", ["tenant_id", "category_id"])
op.create_index("ix_wiki_art_tenant_slug", "wiki_articles", ["tenant_id", "slug"])
op.create_index("ix_wiki_art_tenant_status", "wiki_articles", ["tenant_id", "status"])
op.create_table(
"wiki_article_versions",
sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
sa.Column("tenant_id", PGUUID(as_uuid=True), nullable=False),
sa.Column("article_id", PGUUID(as_uuid=True), nullable=False),
sa.Column("version", sa.Integer, nullable=False),
sa.Column("title", sa.String(300), nullable=False),
sa.Column("content", sa.Text, nullable=False),
sa.Column("edited_by", PGUUID(as_uuid=True), nullable=True),
sa.Column("edit_comment", sa.Text, nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
op.create_foreign_key("fk_wiki_ver_article", "wiki_article_versions", "wiki_articles", ["article_id"], ["id"], ondelete="CASCADE")
op.create_index("ix_wiki_ver_tenant_article", "wiki_article_versions", ["tenant_id", "article_id"])
op.create_index("ix_wiki_ver_tenant_version", "wiki_article_versions", ["tenant_id", "article_id", "version"])
def downgrade() -> None:
op.drop_table("wiki_article_versions")
op.drop_table("wiki_articles")
op.drop_table("wiki_categories")
+101
View File
@@ -0,0 +1,101 @@
"""Wiki plugin models — articles, categories, versions (H-WIKI, H-VER)."""
from __future__ import annotations
import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import ForeignKey, Index, Integer, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.orm import Mapped, mapped_column
from app.core.db import Base, TenantMixin
from app.models.owned_mixin import OwnedMixin
class WikiCategory(Base, TenantMixin, OwnedMixin):
"""Wiki category for organizing articles."""
__tablename__ = "wiki_categories"
__table_args__ = (
Index("ix_wiki_cat_tenant", "tenant_id"),
Index("ix_wiki_cat_tenant_slug", "tenant_id", "slug"),
)
id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
name: Mapped[str] = mapped_column(String(200), nullable=False)
slug: Mapped[str] = mapped_column(String(200), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
parent_id: Mapped[uuid.UUID | None] = mapped_column(
PGUUID(as_uuid=True),
ForeignKey("wiki_categories.id", ondelete="SET NULL"),
nullable=True,
)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class WikiArticle(Base, TenantMixin, OwnedMixin):
"""Wiki article with Markdown content, tags, and entity links (H-WIKI, H-LINK)."""
__tablename__ = "wiki_articles"
__table_args__ = (
Index("ix_wiki_art_tenant", "tenant_id"),
Index("ix_wiki_art_tenant_category", "tenant_id", "category_id"),
Index("ix_wiki_art_tenant_slug", "tenant_id", "slug"),
Index("ix_wiki_art_tenant_status", "tenant_id", "status"),
)
id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
title: Mapped[str] = mapped_column(String(300), nullable=False)
slug: Mapped[str] = mapped_column(String(300), nullable=False)
content: Mapped[str] = mapped_column(Text, nullable=False, default="")
content_html: Mapped[str | None] = mapped_column(Text, nullable=True)
summary: Mapped[str | None] = mapped_column(Text, nullable=True)
category_id: Mapped[uuid.UUID | None] = mapped_column(
PGUUID(as_uuid=True),
ForeignKey("wiki_categories.id", ondelete="SET NULL"),
nullable=True,
)
tags: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
status: Mapped[str] = mapped_column(
String(20), nullable=False, default="draft"
) # draft, published, archived
entity_links: Mapped[list[dict[str, Any]]] = mapped_column(
JSONB, nullable=False, default=list
) # [{"entity_type": "contact", "entity_id": "..."}]
version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
published_at: Mapped[datetime | None] = mapped_column(
nullable=True
)
class WikiArticleVersion(Base, TenantMixin):
"""Article version history for diff and restore (H-VER)."""
__tablename__ = "wiki_article_versions"
__table_args__ = (
Index("ix_wiki_ver_tenant_article", "tenant_id", "article_id"),
Index("ix_wiki_ver_tenant_version", "tenant_id", "article_id", "version"),
)
id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
article_id: Mapped[uuid.UUID] = mapped_column(
PGUUID(as_uuid=True),
ForeignKey("wiki_articles.id", ondelete="CASCADE"),
nullable=False,
)
version: Mapped[int] = mapped_column(Integer, nullable=False)
title: Mapped[str] = mapped_column(String(300), nullable=False)
content: Mapped[str] = mapped_column(Text, nullable=False)
edited_by: Mapped[uuid.UUID | None] = mapped_column(
PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
edit_comment: Mapped[str | None] = mapped_column(Text, nullable=True)
+19
View File
@@ -0,0 +1,19 @@
"""Wiki plugin — knowledge articles, categories, versioning (H-WIKI, H-VER)."""
from __future__ import annotations
from app.plugins.base import BasePlugin
from app.plugins.manifest import FrontendMenuItem, FrontendPageRoute, PluginManifest, PluginRouteDef
class WikiPlugin(BasePlugin):
manifest = PluginManifest(
name="wiki",
version="1.0.0",
display_name="Wiki",
description="Knowledge articles with Markdown, categories, tags, versioning, entity links.",
dependencies=["permissions"],
routes=[
PluginRouteDef(path="/api/v1/wiki", module="app.plugins.builtins.wiki.routes", router_attr="router"),
],
permissions=["wiki:read", "wiki:write", "wiki:delete", "wiki:admin"],
menu_items=[FrontendMenuItem(label_key="wiki.menu.wiki", label="Wiki", path="/wiki", icon="BookOpen")],
page_routes=[FrontendPageRoute(path="/wiki", component="@/pages/Wiki")],
)
+67
View File
@@ -0,0 +1,67 @@
"""Wiki plugin routes — articles CRUD, categories, versioning."""
from __future__ import annotations
import uuid
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import get_current_user, require_permission
from app.plugins.builtins.wiki import services
from app.plugins.builtins.wiki.schemas import ArticleCreate, ArticleUpdate, CategoryCreate, CategoryUpdate
router = APIRouter(prefix="/api/v1/wiki", tags=["wiki"])
@router.get("/articles")
async def list_articles(page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), category_id: str | None = None, status: str | None = None, search: str | None = None, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:read"))):
return await services.list_articles(uuid.UUID(current_user["tenant_id"]), page=page, page_size=page_size, category_id=category_id, status=status, search=search)
@router.post("/articles", status_code=status.HTTP_201_CREATED)
async def create_article(body: ArticleCreate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:write"))):
return await services.create_article(uuid.UUID(current_user["tenant_id"]), uuid.UUID(current_user["user_id"]), body.model_dump())
@router.get("/articles/{article_id}")
async def get_article(article_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:read"))):
result = await services.get_article(uuid.UUID(current_user["tenant_id"]), uuid.UUID(article_id))
if result is None:
raise HTTPException(status_code=404, detail={"detail": "Article not found", "code": "not_found"})
return result
@router.patch("/articles/{article_id}")
async def update_article(article_id: str, body: ArticleUpdate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:write"))):
result = await services.update_article(uuid.UUID(current_user["tenant_id"]), uuid.UUID(current_user["user_id"]), uuid.UUID(article_id), body.model_dump(exclude_unset=True))
if result is None:
raise HTTPException(status_code=404, detail={"detail": "Article not found", "code": "not_found"})
return result
@router.delete("/articles/{article_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_article(article_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:delete"))):
ok = await services.delete_article(uuid.UUID(current_user["tenant_id"]), uuid.UUID(article_id))
if not ok:
raise HTTPException(status_code=404, detail={"detail": "Article not found", "code": "not_found"})
@router.get("/articles/{article_id}/versions")
async def list_versions(article_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:read"))):
return {"items": await services.list_versions(uuid.UUID(current_user["tenant_id"]), uuid.UUID(article_id))}
@router.post("/articles/{article_id}/versions/{version}/restore")
async def restore_version(article_id: str, version: int, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:write"))):
result = await services.restore_version(uuid.UUID(current_user["tenant_id"]), uuid.UUID(current_user["user_id"]), uuid.UUID(article_id), version)
if result is None:
raise HTTPException(status_code=404, detail={"detail": "Version not found", "code": "not_found"})
return result
@router.get("/categories")
async def list_categories(db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:read"))):
return {"items": await services.list_categories(uuid.UUID(current_user["tenant_id"]))}
@router.post("/categories", status_code=status.HTTP_201_CREATED)
async def create_category(body: CategoryCreate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("wiki:write"))):
return await services.create_category(uuid.UUID(current_user["tenant_id"]), uuid.UUID(current_user["user_id"]), body.model_dump())
+37
View File
@@ -0,0 +1,37 @@
"""Pydantic schemas for the Wiki plugin."""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field
class CategoryCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
slug: str = Field(..., min_length=1, max_length=200)
description: str | None = None
parent_id: str | None = None
sort_order: int = 0
class CategoryUpdate(BaseModel):
name: str | None = None
description: str | None = None
parent_id: str | None = None
sort_order: int | None = None
class ArticleCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
slug: str = Field(..., min_length=1, max_length=300)
content: str = ""
summary: str | None = None
category_id: str | None = None
tags: list[str] = Field(default_factory=list)
status: str = Field(default="draft", pattern="^(draft|published|archived)$")
entity_links: list[dict] = Field(default_factory=list)
class ArticleUpdate(BaseModel):
title: str | None = None
content: str | None = None
summary: str | None = None
category_id: str | None = None
tags: list[str] | None = None
status: str | None = Field(None, pattern="^(draft|published|archived)$")
entity_links: list[dict] | None = None
edit_comment: str | None = None
+148
View File
@@ -0,0 +1,148 @@
"""Wiki plugin services — CRUD for articles, categories, versioning."""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.visibility import apply_visibility_filter
from app.plugins.builtins.wiki.models import WikiArticle, WikiArticleVersion, WikiCategory
def _slugify(text: str) -> str:
return text.lower().strip().replace(" ", "-").replace("ä", "ae").replace("ö", "oe").replace("ü", "ue")[:200]
def _article_to_dict(a: WikiArticle) -> dict[str, Any]:
return {
"id": str(a.id), "title": a.title, "slug": a.slug,
"content": a.content, "content_html": a.content_html,
"summary": a.summary, "category_id": str(a.category_id) if a.category_id else None,
"tags": a.tags or [], "status": a.status,
"entity_links": a.entity_links or [], "version": a.version,
"published_at": a.published_at.isoformat() if a.published_at else None,
"owner_id": str(a.owner_id) if a.owner_id else None,
"created_at": a.created_at.isoformat() if a.created_at else None,
"updated_at": a.updated_at.isoformat() if a.updated_at else None,
}
def _category_to_dict(c: WikiCategory) -> dict[str, Any]:
return {
"id": str(c.id), "name": c.name, "slug": c.slug,
"description": c.description, "parent_id": str(c.parent_id) if c.parent_id else None,
"sort_order": c.sort_order,
}
async def list_articles(db: AsyncSession, tenant_id: uuid.UUID, *, page: int = 1, page_size: int = 20, category_id: str | None = None, status: str | None = None, search: str | None = None) -> dict[str, Any]:
query = select(WikiArticle).where(WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None))
if category_id:
query = query.where(WikiArticle.category_id == uuid.UUID(category_id))
if status:
query = query.where(WikiArticle.status == status)
if search:
query = query.where(WikiArticle.title.ilike(f"%{search}%"))
count_q = select(func.count()).select_from(query.subquery())
total = (await db.execute(count_q)).scalar() or 0
query = query.order_by(WikiArticle.updated_at.desc()).offset((page - 1) * page_size).limit(page_size)
result = await db.execute(query)
return {"items": [_article_to_dict(a) for a in result.scalars().all()], "total": total, "page": page, "page_size": page_size}
async def get_article(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> dict[str, Any] | None:
result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
a = result.scalar_one_or_none()
return _article_to_dict(a) if a else None
async def create_article(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any]:
article = WikiArticle(
tenant_id=tenant_id, owner_id=user_id,
title=data["title"], slug=data.get("slug") or _slugify(data["title"]),
content=data.get("content", ""), summary=data.get("summary"),
category_id=uuid.UUID(data["category_id"]) if data.get("category_id") else None,
tags=data.get("tags", []), status=data.get("status", "draft"),
entity_links=data.get("entity_links", []), version=1,
published_at=datetime.now(UTC) if data.get("status") == "published" else None,
)
db.add(article)
await db.flush()
await db.refresh(article)
return _article_to_dict(article)
async def update_article(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, article_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any] | None:
result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
article = result.scalar_one_or_none()
if article is None:
return None
# Save version history (H-VER)
version = WikiArticleVersion(
tenant_id=tenant_id, article_id=article.id, version=article.version,
title=article.title, content=article.content,
edited_by=user_id, edit_comment=data.get("edit_comment"),
)
db.add(version)
if "title" in data and data["title"] is not None:
article.title = data["title"]
if "content" in data and data["content"] is not None:
article.content = data["content"]
if "summary" in data:
article.summary = data["summary"]
if "category_id" in data:
article.category_id = uuid.UUID(data["category_id"]) if data["category_id"] else None
if "tags" in data and data["tags"] is not None:
article.tags = data["tags"]
if "status" in data and data["status"] is not None:
article.status = data["status"]
if data["status"] == "published" and article.published_at is None:
article.published_at = datetime.now(UTC)
if "entity_links" in data and data["entity_links"] is not None:
article.entity_links = data["entity_links"]
article.version += 1
await db.flush()
await db.refresh(article)
return _article_to_dict(article)
async def delete_article(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> bool:
result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
article = result.scalar_one_or_none()
if article is None:
return False
article.deleted_at = datetime.now(UTC)
await db.flush()
return True
async def list_versions(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> list[dict[str, Any]]:
result = await db.execute(select(WikiArticleVersion).where(WikiArticleVersion.tenant_id == tenant_id, WikiArticleVersion.article_id == article_id).order_by(WikiArticleVersion.version.desc()))
return [{"id": str(v.id), "version": v.version, "title": v.title, "content": v.content[:500], "edited_by": str(v.edited_by) if v.edited_by else None, "edit_comment": v.edit_comment, "created_at": v.created_at.isoformat() if v.created_at else None} for v in result.scalars().all()]
async def restore_version(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, article_id: uuid.UUID, version: int) -> dict[str, Any] | None:
vr = await db.execute(select(WikiArticleVersion).where(WikiArticleVersion.tenant_id == tenant_id, WikiArticleVersion.article_id == article_id, WikiArticleVersion.version == version))
v = vr.scalar_one_or_none()
if v is None:
return None
return await update_article(db, tenant_id, user_id, article_id, {"title": v.title, "content": v.content, "edit_comment": f"Restored from version {version}"})
async def list_categories(db: AsyncSession, tenant_id: uuid.UUID) -> list[dict[str, Any]]:
result = await db.execute(select(WikiCategory).where(WikiCategory.tenant_id == tenant_id, WikiCategory.deleted_at.is_(None)).order_by(WikiCategory.sort_order.asc()))
return [_category_to_dict(c) for c in result.scalars().all()]
async def create_category(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any]:
cat = WikiCategory(
tenant_id=tenant_id, owner_id=user_id,
name=data["name"], slug=data.get("slug") or _slugify(data["name"]),
description=data.get("description"),
parent_id=uuid.UUID(data["parent_id"]) if data.get("parent_id") else None,
sort_order=data.get("sort_order", 0),
)
db.add(cat)
await db.flush()
await db.refresh(cat)
return _category_to_dict(cat)
+171
View File
@@ -0,0 +1,171 @@
"""Tests for Phase H — Wiki plugin: models, schemas, plugin manifest.
All tests use mocks — no real DB needed.
"""
from __future__ import annotations
import uuid
from unittest.mock import MagicMock
import pytest
class TestWikiPluginManifest:
"""Test the Wiki plugin manifest."""
def test_plugin_manifest_valid(self):
"""Wiki plugin manifest is valid."""
from app.plugins.builtins.wiki.plugin import WikiPlugin
assert WikiPlugin.manifest.name == "wiki"
assert WikiPlugin.manifest.version == "1.0.0"
assert WikiPlugin.manifest.display_name == "Wiki"
assert "wiki:read" in WikiPlugin.manifest.permissions
assert "wiki:write" in WikiPlugin.manifest.permissions
assert "wiki:delete" in WikiPlugin.manifest.permissions
assert "wiki:admin" in WikiPlugin.manifest.permissions
def test_plugin_has_routes(self):
"""Wiki plugin has routes defined."""
from app.plugins.builtins.wiki.plugin import WikiPlugin
assert len(WikiPlugin.manifest.routes) > 0
assert WikiPlugin.manifest.routes[0].path == "/api/v1/wiki"
def test_plugin_has_menu_item(self):
"""Wiki plugin has a menu item."""
from app.plugins.builtins.wiki.plugin import WikiPlugin
assert len(WikiPlugin.manifest.menu_items) > 0
assert WikiPlugin.manifest.menu_items[0].path == "/wiki"
def test_plugin_has_dependencies(self):
"""Wiki plugin depends on permissions plugin."""
from app.plugins.builtins.wiki.plugin import WikiPlugin
assert "permissions" in WikiPlugin.manifest.dependencies
class TestWikiRoutes:
"""Test the Wiki routes are properly defined."""
def test_routes_count(self):
"""Wiki plugin has 9 routes."""
from app.plugins.builtins.wiki.routes import router
assert len(router.routes) == 9
def test_routes_have_correct_paths(self):
"""Wiki routes have correct API paths."""
from app.plugins.builtins.wiki.routes import router
paths = {r.path for r in router.routes}
assert "/api/v1/wiki/articles" in paths
assert "/api/v1/wiki/articles/{article_id}" in paths
assert "/api/v1/wiki/articles/{article_id}/versions" in paths
assert "/api/v1/wiki/articles/{article_id}/versions/{version}/restore" in paths
assert "/api/v1/wiki/categories" in paths
class TestWikiSchemas:
"""Test the Wiki Pydantic schemas."""
def test_article_create_valid(self):
"""ArticleCreate accepts valid data."""
from app.plugins.builtins.wiki.schemas import ArticleCreate
article = ArticleCreate(title="Test Article", slug="test-article")
assert article.title == "Test Article"
assert article.slug == "test-article"
assert article.status == "draft"
assert article.tags == []
def test_article_create_invalid_status(self):
"""ArticleCreate rejects invalid status."""
from app.plugins.builtins.wiki.schemas import ArticleCreate
from pydantic import ValidationError
with pytest.raises(ValidationError):
ArticleCreate(title="Test", slug="test", status="invalid")
def test_article_update_partial(self):
"""ArticleUpdate allows partial updates."""
from app.plugins.builtins.wiki.schemas import ArticleUpdate
update = ArticleUpdate(title="Updated Title")
assert update.title == "Updated Title"
assert update.content is None
def test_category_create_valid(self):
"""CategoryCreate accepts valid data."""
from app.plugins.builtins.wiki.schemas import CategoryCreate
cat = CategoryCreate(name="Test Category", slug="test-category")
assert cat.name == "Test Category"
assert cat.slug == "test-category"
assert cat.sort_order == 0
class TestWikiModels:
"""Test the Wiki SQLAlchemy models."""
def test_wiki_article_model_fields(self):
"""WikiArticle model has all required fields."""
from app.plugins.builtins.wiki.models import WikiArticle
assert hasattr(WikiArticle, "title")
assert hasattr(WikiArticle, "slug")
assert hasattr(WikiArticle, "content")
assert hasattr(WikiArticle, "content_html")
assert hasattr(WikiArticle, "summary")
assert hasattr(WikiArticle, "category_id")
assert hasattr(WikiArticle, "tags")
assert hasattr(WikiArticle, "status")
assert hasattr(WikiArticle, "entity_links")
assert hasattr(WikiArticle, "version")
assert hasattr(WikiArticle, "published_at")
assert hasattr(WikiArticle, "tenant_id")
assert hasattr(WikiArticle, "owner_id")
def test_wiki_category_model_fields(self):
"""WikiCategory model has all required fields."""
from app.plugins.builtins.wiki.models import WikiCategory
assert hasattr(WikiCategory, "name")
assert hasattr(WikiCategory, "slug")
assert hasattr(WikiCategory, "description")
assert hasattr(WikiCategory, "parent_id")
assert hasattr(WikiCategory, "sort_order")
assert hasattr(WikiCategory, "tenant_id")
def test_wiki_article_version_model_fields(self):
"""WikiArticleVersion model has all required fields."""
from app.plugins.builtins.wiki.models import WikiArticleVersion
assert hasattr(WikiArticleVersion, "article_id")
assert hasattr(WikiArticleVersion, "version")
assert hasattr(WikiArticleVersion, "title")
assert hasattr(WikiArticleVersion, "content")
assert hasattr(WikiArticleVersion, "edited_by")
assert hasattr(WikiArticleVersion, "edit_comment")
assert hasattr(WikiArticleVersion, "tenant_id")
def test_models_have_tenant_mixin(self):
"""All Wiki models have tenant_id (TenantMixin)."""
from app.plugins.builtins.wiki.models import WikiArticle, WikiCategory, WikiArticleVersion
for model in [WikiArticle, WikiCategory, WikiArticleVersion]:
assert hasattr(model, "tenant_id")
class TestWikiServices:
"""Test the Wiki service helper functions."""
def test_slugify(self):
"""_slugify converts text to URL-safe slug."""
from app.plugins.builtins.wiki.services import _slugify
assert _slugify("Hello World") == "hello-world"
assert _slugify("Überblick") == "ueberblick"
assert _slugify("Test Article 123") == "test-article-123"