feat(wiki): convert existing source to wiki + human page edit endpoint

Unit 5 of F-Wiki (D20-D23). convert_source_to_wiki task reuses reingest's storage file-load + parser to materialize files->wiki_pages (one page/file), skips/reports non-text, re-embeds per page, and flips kind=wiki + exposure=agentic_tool only when pages were created. POST /wiki/convert (explicit, write-authz; blank source enables inline, fileful enqueues the task; rejects mid-ingest). PUT /wiki/page for human edits (write-authz, optimistic version -> 409, re-embed). PATCH /config preserves kind (kind changes only via convert).
This commit is contained in:
Alex committed 2026-06-22 20:16:18 +01:00
1 parent 5110189143
commit 0c9e0313bd
8 files changed
+1133 -25

No files matched your search

+245 -1
View File
@@ -10,7 +10,12 @@ from pydantic import ValidationError
from application.agents.tools.path_utils import validate_tool_path
from application.api import api
from application.api.user.tasks import reembed_wiki_page, reingest_source_task, sync_source
from application.api.user.tasks import (
convert_source_to_wiki,
reembed_wiki_page,
reingest_source_task,
sync_source,
)
from application.api.user.team_sharing import (
can_access,
effective_write_owner,
@@ -23,6 +28,7 @@ from application.storage.db.repositories.ingest_chunk_progress import (
)
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import (
WikiPageConflict,
WikiPagesRepository,
_content_hash,
rebuild_wiki_directory_structure,
@@ -624,6 +630,26 @@ class SourceConfigResource(Resource):
# Ingest-time fields (config.chunking) only take effect after a
# re-ingest (D8); compare against the current config to decide.
current_config = SourceConfig.parse(doc.get("config"))
# kind changes route exclusively through /wiki/convert. An
# explicit kind in the body that differs from the persisted
# kind is rejected (either direction); otherwise the current
# kind is preserved so a partial edit omitting it (the model
# defaults to "classic") never demotes a wiki.
if "kind" in body and body["kind"] != current_config.kind:
return make_response(
jsonify(
{
"success": False,
"message": (
"Change wiki mode via "
"POST /api/sources/<id>/wiki/convert, "
"not the config endpoint."
),
}
),
400,
)
new_config.kind = current_config.kind
requires_reingest = (
new_config.chunking.model_dump()
!= current_config.chunking.model_dump()
@@ -670,6 +696,7 @@ def _wiki_page_node(page):
"token_count": page.get("token_count") or 0,
"version": page.get("version"),
"embed_status": page.get("embed_status"),
"updated_by": page.get("updated_by"),
"updated_at": (
page["updated_at"].isoformat()
if page.get("updated_at") is not None
@@ -679,6 +706,18 @@ def _wiki_page_node(page):
}
_wiki_page_edit_model = sources_ns.model(
"WikiPageEditModel",
{
"path": fields.String(required=True, description="Page path"),
"content": fields.String(required=True, description="Markdown content"),
"expected_version": fields.Integer(
description="Optimistic-lock version from the last read"
),
},
)
@sources_ns.route("/sources/wiki")
class CreateWikiSource(Resource):
create_wiki_model = api.model(
@@ -838,3 +877,208 @@ class WikiPage(Resource):
node = _wiki_page_node(page)
node["content"] = page.get("content") or ""
return make_response(jsonify({"success": True, "page": node}), 200)
@api.expect(_wiki_page_edit_model)
@api.doc(
description="Create or overwrite a wiki page (human edit). Write access "
"required; a stale expected_version returns 409."
)
def put(self, source_id):
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
user = decoded_token.get("sub")
data = request.get_json(silent=True) or {}
missing_fields = check_required_fields(data, ["path", "content"])
if missing_fields:
return missing_fields
path = validate_tool_path(data["path"])
if not path or path == "/" or path.endswith("/"):
return make_response(
jsonify({"success": False, "message": "Invalid page path"}), 400
)
content = data["content"]
expected_version = data.get("expected_version")
try:
with db_session() as conn:
owner = effective_write_owner(conn, "source", source_id, user)
if not owner:
return make_response(
jsonify(
{"success": False, "message": "Source not accessible"}
),
403,
)
doc = SourcesRepository(conn).get_any(source_id, owner)
if doc is None:
return make_response(
jsonify({"success": False, "message": "Source not found"}),
404,
)
resolved_source_id = str(doc["id"])
try:
page = WikiPagesRepository(conn).upsert(
resolved_source_id,
path,
content,
updated_by=user,
expected_version=expected_version,
)
except WikiPageConflict:
return make_response(
jsonify(
{
"success": False,
"message": (
"Page changed since you loaded it. "
"Reload and reapply your edit."
),
}
),
409,
)
rebuild_wiki_directory_structure(conn, resolved_source_id, owner)
except Exception as err:
current_app.logger.error(
f"Error editing wiki page for {source_id}: {err}", exc_info=True
)
return make_response(jsonify({"success": False}), 400)
try:
reembed_wiki_page.delay(
resolved_source_id,
path,
_content_hash(content),
user=owner,
idempotency_key=(
f"reembed-wiki:{resolved_source_id}:{path}:"
f"{_content_hash(content)}"
),
)
except Exception as err:
current_app.logger.error(
f"Error enqueuing wiki re-embed for {source_id}: {err}",
exc_info=True,
)
return make_response(
jsonify({"success": True, "page": _wiki_page_node(page)}), 200
)
def _source_is_blank(doc):
"""True when a source has no ingested files to convert into pages."""
structure = doc.get("directory_structure") or {}
if isinstance(structure, str):
try:
structure = json.loads(structure)
except Exception:
structure = {}
return not bool(structure)
def _enable_wiki_inline(conn, doc, owner):
"""Flip a blank source to wiki mode + agentic exposure with no task."""
cfg = SourceConfig.parse(doc.get("config"))
SourcesRepository(conn).update(
str(doc["id"]), owner, {"config": cfg.wiki_enabled()}
)
@sources_ns.route("/sources/<string:source_id>/wiki/convert")
class ConvertSourceToWiki(Resource):
@api.doc(
description="Convert an ingested source into a wiki. A blank source is "
"enabled inline; a source with files runs a conversion task (poll its "
"status for the per-file summary). Write access required (owner/editor)."
)
def post(self, source_id):
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
user = decoded_token.get("sub")
try:
with db_session() as conn:
owner = effective_write_owner(conn, "source", source_id, user)
if not owner:
return make_response(
jsonify(
{"success": False, "message": "Source not accessible"}
),
403,
)
doc = SourcesRepository(conn).get_any(source_id, owner)
if doc is None:
return make_response(
jsonify({"success": False, "message": "Source not found"}),
404,
)
resolved_source_id = str(doc["id"])
# A mid-ingest source has an incomplete directory structure, so
# it could be mis-detected as blank and wrongly enabled inline.
# Reject while an embed is still in flight (mirrors the
# "processing" ingest status the sources list derives).
progress = IngestChunkProgressRepository(conn).get_progress(
resolved_source_id
)
if (
progress
and progress.get("status") != "stalled"
and (progress.get("embedded_chunks") or 0)
< (progress.get("total_chunks") or 0)
):
return make_response(
jsonify(
{
"success": False,
"message": (
"Source is still ingesting; wait for it to "
"finish before converting to a wiki."
),
}
),
409,
)
cfg = SourceConfig.parse(doc.get("config"))
if cfg.kind == "wiki":
return make_response(
jsonify(
{
"success": True,
"converted": False,
"enabled": True,
}
),
200,
)
if _source_is_blank(doc):
_enable_wiki_inline(conn, doc, owner)
return make_response(
jsonify(
{
"success": True,
"converted": False,
"enabled": True,
}
),
200,
)
except Exception as err:
current_app.logger.error(
f"Error preparing wiki conversion for {source_id}: {err}",
exc_info=True,
)
return make_response(jsonify({"success": False}), 400)
try:
task = convert_source_to_wiki.delay(
source_id=resolved_source_id,
user=owner,
idempotency_key=f"convert-wiki:{resolved_source_id}",
)
except Exception as err:
current_app.logger.error(
f"Error starting wiki conversion for {source_id}: {err}",
exc_info=True,
)
return make_response(jsonify({"success": False}), 400)
return make_response(
jsonify({"success": True, "task_id": task.id}), 200
)
+9
View File
@@ -128,6 +128,15 @@ def reembed_wiki_page(
return resp
@celery.task(**DURABLE_TASK)
@with_idempotency(task_name="convert_source_to_wiki")
def convert_source_to_wiki(self, source_id, user, idempotency_key=None):
from application.worker import convert_source_to_wiki_worker
resp = convert_source_to_wiki_worker(self, source_id, user)
return resp
# Beat-driven dispatch tasks default to ``acks_late=False``: a SIGKILL
# of a beat tick is harmless to redeliver only if the dispatch itself is
# idempotent. We keep these early-ACK so the broker doesn't replay a
+13
View File
@@ -117,6 +117,19 @@ class SourceConfig(BaseModel):
chunking: ChunkingConfig = ChunkingConfig()
retrieval: RetrievalConfig = RetrievalConfig()
def wiki_enabled(self) -> dict:
"""Return a config dict flipped to wiki mode + browse-as-you-go exposure.
Sets ``kind="wiki"`` and defaults ``retrieval.exposure`` to
``agentic_tool`` (a wiki is navigated, not bulk-prefetched), preserving
any non-default exposure the source already carries.
"""
new_config = self.model_dump()
new_config["kind"] = "wiki"
if self.retrieval.exposure == "prefetch":
new_config["retrieval"]["exposure"] = "agentic_tool"
return new_config
@classmethod
def parse(cls, raw: Optional[dict]) -> "SourceConfig":
"""Lenient read: return all-defaults for ``{}``/``None``.
+167 -24
View File
@@ -39,7 +39,11 @@ from application.storage.db.repositories.ingest_chunk_progress import (
IngestChunkProgressRepository,
)
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import WikiPagesRepository
from application.storage.db.repositories.wiki_pages import (
WikiPagesRepository,
_content_hash,
rebuild_wiki_directory_structure,
)
from application.storage.db.session import db_readonly, db_session
from application.storage.db.source_config import SourceConfig
from application.storage.storage_creator import StorageCreator
@@ -174,6 +178,25 @@ def _apply_display_names_to_structure(structure, file_name_map, prefix=""):
return structure
def _download_source_files_to_dir(storage, source_file_path, temp_dir):
"""Mirror a source's stored files into ``temp_dir``, preserving structure."""
if not storage.is_directory(source_file_path):
return
for storage_file_path in storage.list_files(source_file_path):
if storage.is_directory(storage_file_path):
continue
rel_path = os.path.relpath(storage_file_path, source_file_path)
local_file_path = os.path.join(temp_dir, rel_path)
os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
try:
file_data = storage.get_file(storage_file_path)
with open(local_file_path, "wb") as f:
f.write(file_data.read())
except Exception as e:
logging.error(f"Error downloading file {storage_file_path}: {e}")
continue
# Define a function to generate a random string of a given length.
@@ -734,29 +757,7 @@ def reingest_source_worker(self, source_id, user):
)
with tempfile.TemporaryDirectory() as temp_dir:
# Download all files from storage to temp directory, preserving directory structure
if storage.is_directory(source_file_path):
files_list = storage.list_files(source_file_path)
for storage_file_path in files_list:
if storage.is_directory(storage_file_path):
continue
rel_path = os.path.relpath(storage_file_path, source_file_path)
local_file_path = os.path.join(temp_dir, rel_path)
os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
# Download file
try:
file_data = storage.get_file(storage_file_path)
with open(local_file_path, "wb") as f:
f.write(file_data.read())
except Exception as e:
logging.error(
f"Error downloading file {storage_file_path}: {e}"
)
continue
_download_source_files_to_dir(storage, source_file_path, temp_dir)
reader = SimpleDirectoryReader(
input_dir=temp_dir,
@@ -2058,3 +2059,145 @@ def reembed_wiki_page_worker(self, source_id, path, content_hash, user):
raise
return {"status": "embedded", "added": added, "deleted": deleted}
def _wiki_page_path_from_rel(rel_path):
"""Build a validated leading-slash wiki page path, or None if invalid.
Runs the derived path through ``validate_tool_path`` so a stored filename
carrying traversal (``..``) never lands as a ``wiki_pages.path``.
"""
from application.agents.tools.path_utils import validate_tool_path
raw = "/" + rel_path.replace(os.sep, "/").lstrip("/")
return validate_tool_path(raw)
def convert_source_to_wiki_worker(self, source_id, user):
"""Convert an ingested source into a wiki by materializing files as pages.
Re-parses the source's stored originals (the same storage file-load +
``SimpleDirectoryReader`` machinery reingest uses), creating one wiki page
per file with extractable text. Non-text/binary/unsupported files are
skipped and reported. Embedding is enqueued per page (async); the source's
``config.kind`` flips to ``wiki`` only after at least one page is
materialized — a source with no extractable pages keeps its original kind
and directory structure.
Args:
self: Celery task instance.
source_id: ID of the source to convert.
user: Owner identifier used to load the source row and author pages.
Returns:
A summary dict ``{"status": "converted", "pages_created", "skipped"}``,
``{"status": "no_pages", ...}`` when nothing was extractable, or
``{"status": "already_wiki"}`` when the source is already a wiki.
"""
from application.api.user.tasks import reembed_wiki_page
source_id = str(source_id)
with db_readonly() as conn:
source = SourcesRepository(conn).get_any(source_id, user)
if not source:
raise ValueError(f"Source {source_id} not found or access denied")
source_id = str(source["id"])
cfg = SourceConfig.parse(source.get("config"))
if cfg.kind == "wiki":
return {"status": "already_wiki", "pages_created": 0, "skipped": []}
storage = StorageCreator.get_storage()
source_file_path = source.get("file_path", "")
file_name_map = _normalize_file_name_map(source.get("file_name_map"))
created_pages: list[tuple[str, str]] = []
skipped: list[dict] = []
stored_files: set[str] = set()
if storage.is_directory(source_file_path):
for stored in storage.list_files(source_file_path):
if storage.is_directory(stored):
continue
stored_files.add(os.path.relpath(stored, source_file_path))
with tempfile.TemporaryDirectory() as temp_dir:
_download_source_files_to_dir(storage, source_file_path, temp_dir)
reader = SimpleDirectoryReader(
input_dir=temp_dir,
recursive=True,
required_exts=list(SUPPORTED_SOURCE_EXTENSIONS),
exclude_hidden=True,
errors="ignore",
file_metadata=metadata_from_filename,
)
raw_docs = reader.load_data()
texts_by_rel: dict[str, list[str]] = {}
for doc in raw_docs:
extra_info = getattr(doc, "extra_info", None) or {}
rel_path = extra_info.get("source")
if not rel_path:
continue
text = getattr(doc, "text", None) or ""
if not text.strip():
continue
texts_by_rel.setdefault(rel_path, []).append(text)
with db_session() as conn:
repo = WikiPagesRepository(conn)
for rel_path in sorted(texts_by_rel):
page_path = _wiki_page_path_from_rel(rel_path)
if page_path is None:
skipped.append({"file": rel_path, "reason": "invalid path"})
continue
content = "\n\n".join(texts_by_rel[rel_path])
title = _get_display_name(file_name_map, rel_path) or os.path.basename(
rel_path
)
repo.upsert(
source_id, page_path, content, title=title, updated_by=user
)
created_pages.append((page_path, _content_hash(content)))
for rel_path in sorted(stored_files - set(texts_by_rel)):
ext = os.path.splitext(rel_path)[1].lower()
reason = (
"could not read/parse"
if ext in SUPPORTED_SOURCE_EXTENSIONS
else "no extractable text"
)
skipped.append({"file": rel_path, "reason": reason})
if created_pages:
with db_session() as conn:
rebuild_wiki_directory_structure(conn, source_id, user)
if not created_pages:
return {
"status": "no_pages",
"pages_created": 0,
"skipped": skipped,
}
for page_path, content_hash in created_pages:
reembed_wiki_page.delay(
source_id,
page_path,
content_hash,
user=user,
idempotency_key=f"reembed-wiki:{source_id}:{page_path}:{content_hash}",
)
with db_session() as conn:
SourcesRepository(conn).update(
source_id, user, {"config": cfg.wiki_enabled()}
)
return {
"status": "converted",
"pages_created": len(created_pages),
"skipped": skipped,
}
+105
View File
@@ -1029,3 +1029,108 @@ class TestSourceConfigResource:
# owner-scoped read sees it.
got = SourcesRepository(pg_conn).get_any(sid, owner)
assert got["config"]["retrieval"]["chunks"] == 9
def test_kind_flip_to_wiki_rejected(self, app, pg_conn):
# Flipping kind to wiki must route through /wiki/convert, not config.
from application.api.user.sources.routes import SourceConfigResource
from application.storage.db.repositories.sources import SourcesRepository
user = "u-cfg-wiki-flip"
src = _seed_source(pg_conn, user, name="cfg-wiki", type="file")
sid = str(src["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/sources/{sid}/config",
method="PATCH",
json={"kind": "wiki"},
):
from flask import request
request.decoded_token = {"sub": user}
response = SourceConfigResource().patch(sid)
assert response.status_code == 400
# The kind must NOT have silently flipped.
from application.storage.db.source_config import SourceConfig
got = SourcesRepository(pg_conn).get_any(sid, user)
assert SourceConfig.parse(got.get("config")).kind != "wiki"
def test_other_edits_work_on_wiki_source(self, app, pg_conn):
# A wiki source can still edit retrieval (kind stays wiki, no reject).
from application.api.user.sources.routes import SourceConfigResource
from application.storage.db.repositories.sources import SourcesRepository
user = "u-cfg-wiki-edit"
src = _seed_source(
pg_conn, user, name="wiki-cfg", type="wiki",
config={"kind": "wiki"},
)
sid = str(src["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/sources/{sid}/config",
method="PATCH",
json={"kind": "wiki", "retrieval": {"chunks": 4}},
):
from flask import request
request.decoded_token = {"sub": user}
response = SourceConfigResource().patch(sid)
assert response.status_code == 200
got = SourcesRepository(pg_conn).get_any(sid, user)
assert got["config"]["retrieval"]["chunks"] == 4
def test_partial_edit_preserves_wiki_kind(self, app, pg_conn):
# A partial edit that OMITS kind must not demote a wiki to classic
# (SourceConfig.kind defaults to "classic" on a full-replace write).
from application.api.user.sources.routes import SourceConfigResource
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.source_config import SourceConfig
user = "u-cfg-wiki-partial"
src = _seed_source(
pg_conn, user, name="wiki-partial", type="wiki",
config={"kind": "wiki"},
)
sid = str(src["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/sources/{sid}/config",
method="PATCH",
json={"retrieval": {"chunks": 6}},
):
from flask import request
request.decoded_token = {"sub": user}
response = SourceConfigResource().patch(sid)
assert response.status_code == 200
assert response.json["config"]["kind"] == "wiki"
got = SourcesRepository(pg_conn).get_any(sid, user)
assert SourceConfig.parse(got.get("config")).kind == "wiki"
assert got["config"]["retrieval"]["chunks"] == 6
def test_explicit_kind_demotion_from_wiki_rejected(self, app, pg_conn):
# Demoting wiki -> classic via config is rejected; use /wiki/convert.
from application.api.user.sources.routes import SourceConfigResource
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.source_config import SourceConfig
user = "u-cfg-wiki-demote"
src = _seed_source(
pg_conn, user, name="wiki-demote", type="wiki",
config={"kind": "wiki"},
)
sid = str(src["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/sources/{sid}/config",
method="PATCH",
json={"kind": "classic"},
):
from flask import request
request.decoded_token = {"sub": user}
response = SourceConfigResource().patch(sid)
assert response.status_code == 400
got = SourcesRepository(pg_conn).get_any(sid, user)
assert SourceConfig.parse(got.get("config")).kind == "wiki"
+273
View File
@@ -290,3 +290,276 @@ class TestWikiPage:
request.decoded_token = {"sub": stranger}
response = WikiPage().get(sid)
assert response.status_code == 404
class TestConvertSourceToWiki:
def test_returns_401_unauthenticated(self, app):
from application.api.user.sources.routes import ConvertSourceToWiki
with app.test_request_context(
"/api/sources/x/wiki/convert", method="POST"
):
from flask import request
request.decoded_token = None
response = ConvertSourceToWiki().post("x")
assert response.status_code == 401
def test_blank_source_enabled_inline_no_task(self, app, pg_conn):
from application.api.user.sources.routes import ConvertSourceToWiki
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.source_config import SourceConfig
user = "u-convert-blank"
src = SourcesRepository(pg_conn).create(
"blank", user_id=user, type="file", directory_structure={}
)
sid = str(src["id"])
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.convert_source_to_wiki.delay"
) as mock_convert, app.test_request_context(
f"/api/sources/{sid}/wiki/convert", method="POST"
):
from flask import request
request.decoded_token = {"sub": user}
response = ConvertSourceToWiki().post(sid)
assert response.status_code == 200
assert response.json["converted"] is False
assert response.json["enabled"] is True
mock_convert.assert_not_called()
got = SourcesRepository(pg_conn).get_any(sid, user)
cfg = SourceConfig.parse(got.get("config"))
assert cfg.kind == "wiki"
assert cfg.retrieval.exposure == "agentic_tool"
def test_fileful_source_enqueues_task(self, app, pg_conn):
from application.api.user.sources.routes import ConvertSourceToWiki
from application.storage.db.repositories.sources import SourcesRepository
user = "u-convert-files"
src = SourcesRepository(pg_conn).create(
"files", user_id=user, type="file",
directory_structure={"a.md": {"type": "text/markdown"}},
)
sid = str(src["id"])
fake_task = type("T", (), {"id": "task-xyz"})()
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.convert_source_to_wiki.delay",
return_value=fake_task,
) as mock_convert, app.test_request_context(
f"/api/sources/{sid}/wiki/convert", method="POST"
):
from flask import request
request.decoded_token = {"sub": user}
response = ConvertSourceToWiki().post(sid)
assert response.status_code == 200
assert response.json["task_id"] == "task-xyz"
mock_convert.assert_called_once()
assert mock_convert.call_args.kwargs["source_id"] == sid
assert mock_convert.call_args.kwargs["user"] == user
assert mock_convert.call_args.kwargs["idempotency_key"] == (
f"convert-wiki:{sid}"
)
def test_in_progress_ingest_rejected_409(self, app, pg_conn):
from application.api.user.sources.routes import ConvertSourceToWiki
from application.storage.db.repositories.ingest_chunk_progress import (
IngestChunkProgressRepository,
)
from application.storage.db.repositories.sources import SourcesRepository
user = "u-convert-ingesting"
src = SourcesRepository(pg_conn).create(
"ingesting", user_id=user, type="file",
directory_structure={"a.md": {"type": "text/markdown"}},
)
sid = str(src["id"])
# An active embed in flight (embedded 0 of 5) → "processing".
IngestChunkProgressRepository(pg_conn).init_progress(sid, 5)
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.convert_source_to_wiki.delay"
) as mock_convert, app.test_request_context(
f"/api/sources/{sid}/wiki/convert", method="POST"
):
from flask import request
request.decoded_token = {"sub": user}
response = ConvertSourceToWiki().post(sid)
assert response.status_code == 409
mock_convert.assert_not_called()
def test_viewer_rejected_403(self, app, pg_conn):
from application.api.user.sources.routes import ConvertSourceToWiki
from application.storage.db.repositories.sources import SourcesRepository
owner = "alice-convert"
viewer = "bob-convert-viewer"
src = SourcesRepository(pg_conn).create(
"files", user_id=owner, type="file",
directory_structure={"a.md": {"type": "text/markdown"}},
)
sid = str(src["id"])
_grant_team_access(pg_conn, owner, viewer, sid, "viewer")
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.convert_source_to_wiki.delay"
) as mock_convert, app.test_request_context(
f"/api/sources/{sid}/wiki/convert", method="POST"
):
from flask import request
request.decoded_token = {"sub": viewer}
response = ConvertSourceToWiki().post(sid)
assert response.status_code == 403
mock_convert.assert_not_called()
class TestWikiPageEdit:
def test_returns_401_unauthenticated(self, app):
from application.api.user.sources.routes import WikiPage
with app.test_request_context(
"/api/sources/x/wiki/page", method="PUT", json={}
):
from flask import request
request.decoded_token = None
response = WikiPage().put("x")
assert response.status_code == 401
def test_owner_writes_and_enqueues_reembed(self, app, pg_conn):
from application.api.user.sources.routes import WikiPage
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import WikiPagesRepository
user = "u-edit-owner"
src = SourcesRepository(pg_conn).create(
"wiki", user_id=user, type="wiki", config={"kind": "wiki"}
)
sid = str(src["id"])
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.reembed_wiki_page.delay"
) as mock_reembed, app.test_request_context(
f"/api/sources/{sid}/wiki/page",
method="PUT",
json={"path": "/notes.md", "content": "body"},
):
from flask import request
request.decoded_token = {"sub": user}
response = WikiPage().put(sid)
assert response.status_code == 200
assert response.json["page"]["path"] == "/notes.md"
assert response.json["page"]["version"] is not None
page = WikiPagesRepository(pg_conn).get_by_path(sid, "/notes.md")
assert page["content"] == "body"
mock_reembed.assert_called_once()
assert mock_reembed.call_args.args[1] == "/notes.md"
assert mock_reembed.call_args.kwargs["user"] == user
def test_team_editor_reembeds_as_owner(self, app, pg_conn):
from application.api.user.sources.routes import WikiPage
from application.storage.db.repositories.sources import SourcesRepository
owner = "alice-edit"
editor = "bob-edit-editor"
src = SourcesRepository(pg_conn).create(
"wiki", user_id=owner, type="wiki", config={"kind": "wiki"}
)
sid = str(src["id"])
_grant_team_access(pg_conn, owner, editor, sid, "editor")
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.reembed_wiki_page.delay"
) as mock_reembed, app.test_request_context(
f"/api/sources/{sid}/wiki/page",
method="PUT",
json={"path": "/e.md", "content": "edited"},
):
from flask import request
request.decoded_token = {"sub": editor}
response = WikiPage().put(sid)
assert response.status_code == 200
# Re-embed runs AS the owner so the owner-scoped worker load resolves.
assert mock_reembed.call_args.kwargs["user"] == owner
def test_stale_version_returns_409(self, app, pg_conn):
from application.api.user.sources.routes import WikiPage
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import WikiPagesRepository
user = "u-edit-conflict"
src = SourcesRepository(pg_conn).create(
"wiki", user_id=user, type="wiki", config={"kind": "wiki"}
)
sid = str(src["id"])
WikiPagesRepository(pg_conn).upsert(sid, "/c.md", "v1", updated_by=user)
# Bump to version 2 so a stale expected_version=1 loses the race.
WikiPagesRepository(pg_conn).upsert(sid, "/c.md", "v2", updated_by=user)
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.reembed_wiki_page.delay"
) as mock_reembed, app.test_request_context(
f"/api/sources/{sid}/wiki/page",
method="PUT",
json={"path": "/c.md", "content": "v3", "expected_version": 1},
):
from flask import request
request.decoded_token = {"sub": user}
response = WikiPage().put(sid)
assert response.status_code == 409
mock_reembed.assert_not_called()
def test_traversal_path_returns_400(self, app, pg_conn):
from application.api.user.sources.routes import WikiPage
from application.storage.db.repositories.sources import SourcesRepository
user = "u-edit-traversal"
src = SourcesRepository(pg_conn).create(
"wiki", user_id=user, type="wiki", config={"kind": "wiki"}
)
sid = str(src["id"])
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.reembed_wiki_page.delay"
), app.test_request_context(
f"/api/sources/{sid}/wiki/page",
method="PUT",
json={"path": "/../secret.md", "content": "x"},
):
from flask import request
request.decoded_token = {"sub": user}
response = WikiPage().put(sid)
assert response.status_code == 400
def test_viewer_rejected_403(self, app, pg_conn):
from application.api.user.sources.routes import WikiPage
from application.storage.db.repositories.sources import SourcesRepository
owner = "alice-edit-viewer"
viewer = "bob-edit-viewer"
src = SourcesRepository(pg_conn).create(
"wiki", user_id=owner, type="wiki", config={"kind": "wiki"}
)
sid = str(src["id"])
_grant_team_access(pg_conn, owner, viewer, sid, "viewer")
with _patch_db(pg_conn), patch(
"application.api.user.sources.routes.reembed_wiki_page.delay"
) as mock_reembed, app.test_request_context(
f"/api/sources/{sid}/wiki/page",
method="PUT",
json={"path": "/v.md", "content": "x"},
):
from flask import request
request.decoded_token = {"sub": viewer}
response = WikiPage().put(sid)
assert response.status_code == 403
mock_reembed.assert_not_called()
+16
View File
@@ -86,6 +86,20 @@ class TestReingestSourceTask:
assert result == {"status": "ok"}
class TestConvertSourceToWikiTask:
@pytest.mark.unit
@patch("application.worker.convert_source_to_wiki_worker")
def test_calls_convert_worker(self, mock_worker):
from application.api.user.tasks import convert_source_to_wiki
mock_worker.return_value = {"status": "converted"}
result = convert_source_to_wiki("source123", "user1")
mock_worker.assert_called_once_with(ANY, "source123", "user1")
assert result == {"status": "converted"}
class TestScheduleSyncsTask:
@pytest.mark.unit
@patch("application.api.user.tasks.sync_worker")
@@ -286,6 +300,8 @@ class TestDurableTaskRetryPolicy:
"store_attachment",
"process_agent_webhook",
"ingest_connector_task",
"reembed_wiki_page",
"convert_source_to_wiki",
],
)
def test_task_has_retry_config(self, task_name):
+305
View File
@@ -0,0 +1,305 @@
"""Tests for ``application.worker.convert_source_to_wiki_worker``.
The worker re-parses a source's stored originals into wiki pages and enqueues
a per-page re-embed. Storage, ``SimpleDirectoryReader``, and the re-embed task
are mocked so no real downloads or embeddings run; the ``sources`` /
``wiki_pages`` rows are real so the assertions read them back.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from application.parser.schema.base import Document
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.wiki_pages import WikiPagesRepository
def _seed_source(pg_conn, config=None, file_path="inputs/alice/doc-set"):
src = SourcesRepository(pg_conn).create(
"doc-set",
user_id="alice",
type="local",
retriever="classic",
file_path=file_path,
config=config or {},
directory_structure={"a.md": {"type": "text/markdown", "size_bytes": 1}},
)
return str(src["id"])
def _patch_storage(monkeypatch, stored_files):
"""Storage whose directory holds ``stored_files`` (rel paths)."""
from application import worker
base = "inputs/alice/doc-set"
storage = MagicMock(name="storage")
storage.is_directory.side_effect = lambda p: p == base
storage.list_files.return_value = [f"{base}/{f}" for f in stored_files]
storage.get_file.return_value = MagicMock(read=lambda: b"x")
monkeypatch.setattr(worker.StorageCreator, "get_storage", lambda: storage)
return storage
def _patch_reader(monkeypatch, docs):
from application import worker
reader = MagicMock(name="reader")
reader.load_data.return_value = docs
monkeypatch.setattr(worker, "SimpleDirectoryReader", lambda *a, **kw: reader)
return reader
@pytest.mark.unit
class TestConvertSourceToWikiWorker:
def test_one_page_per_text_file(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["guide.md", "notes/intro.md", "logo.png"])
_patch_reader(
monkeypatch,
[
Document(text="guide body", extra_info={"source": "guide.md"}),
Document(
text="intro body", extra_info={"source": "notes/intro.md"}
),
],
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["status"] == "converted"
assert result["pages_created"] == 2
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
by_path = {p["path"]: p for p in pages}
assert set(by_path) == {"/guide.md", "/notes/intro.md"}
assert by_path["/guide.md"]["content"] == "guide body"
assert by_path["/guide.md"]["title"] == "guide.md"
assert by_path["/notes/intro.md"]["title"] == "intro.md"
def test_binaries_skipped_and_reported(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["guide.md", "logo.png", "scan.pdf"])
_patch_reader(
monkeypatch,
[Document(text="guide body", extra_info={"source": "guide.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["pages_created"] == 1
skipped_files = {s["file"] for s in result["skipped"]}
assert skipped_files == {"logo.png", "scan.pdf"}
for entry in result["skipped"]:
assert entry["reason"]
def test_reembed_enqueued_per_page_with_per_page_key(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["a.md", "b.md"])
_patch_reader(
monkeypatch,
[
Document(text="aaa", extra_info={"source": "a.md"}),
Document(text="bbb", extra_info={"source": "b.md"}),
],
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
assert delay.call_count == 2
keys = {c.kwargs["idempotency_key"] for c in delay.call_args_list}
paths = {c.args[1] for c in delay.call_args_list}
assert paths == {"/a.md", "/b.md"}
for call in delay.call_args_list:
assert call.args[0] == source_id
assert call.kwargs["user"] == "alice"
assert call.kwargs["idempotency_key"].startswith(
f"reembed-wiki:{source_id}:"
)
assert len(keys) == 2
def test_sets_kind_wiki_and_agentic_exposure(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
from application.storage.db.source_config import SourceConfig
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["a.md"])
_patch_reader(
monkeypatch,
[Document(text="aaa", extra_info={"source": "a.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice")
cfg = SourceConfig.parse(refreshed.get("config"))
assert cfg.kind == "wiki"
assert cfg.retrieval.exposure == "agentic_tool"
def test_preserves_non_default_exposure(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
from application.storage.db.source_config import SourceConfig
source_id = _seed_source(
pg_conn, config={"retrieval": {"exposure": "agentic_tool"}}
)
_patch_storage(monkeypatch, ["a.md"])
_patch_reader(
monkeypatch,
[Document(text="aaa", extra_info={"source": "a.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
worker.convert_source_to_wiki_worker(task_self, source_id, "alice")
refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice")
cfg = SourceConfig.parse(refreshed.get("config"))
assert cfg.kind == "wiki"
assert cfg.retrieval.exposure == "agentic_tool"
def test_no_pages_leaves_kind_and_structure(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
from application.storage.db.source_config import SourceConfig
original_structure = {"a.md": {"type": "text/markdown", "size_bytes": 1}}
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["logo.png", "scan.pdf"])
_patch_reader(monkeypatch, [])
rebuild = MagicMock(name="rebuild")
monkeypatch.setattr(
worker, "rebuild_wiki_directory_structure", rebuild
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["status"] == "no_pages"
assert result["pages_created"] == 0
assert result["skipped"]
rebuild.assert_not_called()
delay.assert_not_called()
refreshed = SourcesRepository(pg_conn).get_any(source_id, "alice")
assert SourceConfig.parse(refreshed.get("config")).kind != "wiki"
assert refreshed.get("directory_structure") == original_structure
def test_traversal_filename_skipped(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
_patch_storage(monkeypatch, ["ok.md", "../evil.md"])
_patch_reader(
monkeypatch,
[
Document(text="good", extra_info={"source": "ok.md"}),
Document(text="bad", extra_info={"source": "../evil.md"}),
],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["pages_created"] == 1
pages = WikiPagesRepository(pg_conn).list_for_source(source_id)
assert {p["path"] for p in pages} == {"/ok.md"}
skipped = {(s["file"], s["reason"]) for s in result["skipped"]}
assert ("../evil.md", "invalid path") in skipped
def test_parse_failure_reason_distinct_from_binary(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn)
# broken.md has a supported extension but produced no doc (parse/read
# failure); blob.bin is an unsupported/binary file.
_patch_storage(monkeypatch, ["guide.md", "broken.md", "blob.bin"])
_patch_reader(
monkeypatch,
[Document(text="guide body", extra_info={"source": "guide.md"})],
)
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", MagicMock()
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
reasons = {s["file"]: s["reason"] for s in result["skipped"]}
assert reasons["broken.md"] == "could not read/parse"
assert reasons["blob.bin"] == "no extractable text"
def test_already_wiki_returns_early(
self, pg_conn, patch_worker_db, task_self, monkeypatch
):
from application import worker
source_id = _seed_source(pg_conn, config={"kind": "wiki"})
reader = MagicMock(name="reader")
monkeypatch.setattr(
worker, "SimpleDirectoryReader", lambda *a, **kw: reader
)
delay = MagicMock(name="reembed_delay")
monkeypatch.setattr(
"application.api.user.tasks.reembed_wiki_page.delay", delay
)
result = worker.convert_source_to_wiki_worker(
task_self, source_id, "alice"
)
assert result["status"] == "already_wiki"
reader.load_data.assert_not_called()
delay.assert_not_called()