diff --git a/application/worker.py b/application/worker.py index 17962460..6f033951 100755 --- a/application/worker.py +++ b/application/worker.py @@ -12,7 +12,7 @@ import zipfile import uuid from collections import Counter -from urllib.parse import urljoin +from urllib.parse import urljoin, urlsplit import requests @@ -2012,7 +2012,9 @@ def reembed_wiki_page_worker(self, source_id, path, content_hash, user): source_id = str(source["id"]) cfg = SourceConfig.parse(source.get("config")) - store = VectorCreator.create_vectorstore(settings.VECTOR_STORE, source_id) + store = VectorCreator.create_vectorstore( + settings.VECTOR_STORE, source_id, settings.EMBEDDINGS_KEY + ) deleted = store.delete_chunks_by_source_path(path) with db_readonly() as conn: @@ -2073,16 +2075,98 @@ def _wiki_page_path_from_rel(rel_path): 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. +def _chunk_text(chunk) -> str: + if isinstance(chunk, dict): + return chunk.get("text") or chunk.get("page_content") or "" + return getattr(chunk, "text", None) or getattr(chunk, "page_content", "") or "" - 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. + +def _chunk_metadata(chunk) -> dict: + if isinstance(chunk, dict): + meta = chunk.get("metadata") + else: + meta = getattr(chunk, "metadata", None) + return meta if isinstance(meta, dict) else {} + + +def _chunk_doc_id(chunk): + if isinstance(chunk, dict): + return chunk.get("doc_id") + return getattr(chunk, "doc_id", None) + + +def _url_to_virtual_path(value: str) -> str: + parts = urlsplit(value) + path = parts.path.strip("/") + if path: + return path + if parts.netloc: + return parts.netloc + return value.replace("://", "/") + + +def _chunk_page_path(metadata) -> str: + for key in ("file_path", "file_name", "filename", "title", "source"): + value = metadata.get(key) + if value: + value = str(value) + if "://" in value: + value = _url_to_virtual_path(value) + return value + return "" + + +def _chunk_order_hint(metadata): + for key in ("chunk", "chunk_index", "index", "start", "start_index", "offset"): + value = metadata.get(key) + if isinstance(value, (int, float)): + return value + if isinstance(value, str): + try: + return float(value) + except ValueError: + continue + return None + + +MIN_CHUNK_OVERLAP_TRIM = 32 + + +def _join_chunk_texts(texts: list[str]) -> str: + """Concatenate chunk texts, trimming an exact overlapping boundary. + + Only an exact suffix/prefix overlap of at least ``MIN_CHUNK_OVERLAP_TRIM`` + characters between consecutive chunks is removed (the common + ``chunk_overlap`` case); shorter, possibly legitimate repeats are kept. + """ + parts: list[str] = [] + for text in texts: + if parts: + prev = parts[-1] + limit = min(len(prev), len(text)) + overlap = 0 + for size in range(limit, MIN_CHUNK_OVERLAP_TRIM - 1, -1): + if prev.endswith(text[:size]): + overlap = size + break + if overlap: + text = text[overlap:] + if text: + parts.append(text) + return "\n\n".join(parts) + + +def convert_source_to_wiki_worker(self, source_id, user): + """Convert an ingested source into a wiki by reassembling pages from chunks. + + Groups the source's existing vector-store chunks by their page path + (``metadata.source``), concatenating each group into one wiki page. This + works uniformly for every source type, including crawler/remote/connector + sources that hold no stored originals. Chunks with a missing or invalid + path 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 usable chunks keeps its original kind and + directory structure. Args: self: Celery task instance. @@ -2091,10 +2175,11 @@ def convert_source_to_wiki_worker(self, source_id, user): Returns: A summary dict ``{"status": "converted", "pages_created", "skipped"}``, - ``{"status": "no_pages", ...}`` when nothing was extractable, or + ``{"status": "no_pages", ...}`` when nothing was reassembled, or ``{"status": "already_wiki"}`` when the source is already a wiki. """ from application.api.user.tasks import reembed_wiki_page + from application.vectorstore.vector_creator import VectorCreator source_id = str(source_id) @@ -2108,77 +2193,61 @@ def convert_source_to_wiki_worker(self, source_id, user): 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)) + store = VectorCreator.create_vectorstore( + settings.VECTOR_STORE, source_id, settings.EMBEDDINGS_KEY + ) - with tempfile.TemporaryDirectory() as temp_dir: - _download_source_files_to_dir(storage, source_file_path, temp_dir) + grouped: dict[str, list[tuple[object, str]]] = {} + original_doc_ids: list[object] = [] + for chunk in store.get_chunks() or []: + metadata = _chunk_metadata(chunk) + rel_path = _chunk_page_path(metadata) + text = _chunk_text(chunk) + if not text.strip(): + continue + if not rel_path: + skipped.append({"file": "", "reason": "missing path"}) + continue + if _wiki_page_path_from_rel(rel_path) is None: + skipped.append({"file": rel_path, "reason": "invalid path"}) + continue + doc_id = _chunk_doc_id(chunk) + if doc_id is not None: + original_doc_ids.append(doc_id) + grouped.setdefault(rel_path, []).append((_chunk_order_hint(metadata), text)) - 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 + with db_session() as conn: + repo = WikiPagesRepository(conn) + for rel_path in sorted(grouped): + page_path = _wiki_page_path_from_rel(rel_path) + entries = grouped[rel_path] + if any(hint is not None for hint, _ in entries): + entries = sorted( + enumerate(entries), + key=lambda item: ( + item[1][0] if item[1][0] is not None else float("inf"), + item[0], + ), ) - repo.upsert( - source_id, - page_path, - content, - title=title, - updated_by=user, - updated_via="agent", - ) - 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" + entries = [entry for _, entry in entries] + content = _join_chunk_texts([text for _, text in entries]) + title = _get_display_name(file_name_map, rel_path) or os.path.basename( + rel_path ) - skipped.append({"file": rel_path, "reason": reason}) - - if created_pages: - with db_session() as conn: - rebuild_wiki_directory_structure(conn, source_id, user) + repo.upsert( + source_id, + page_path, + content, + title=title, + updated_by=user, + updated_via="agent", + ) + created_pages.append((page_path, _content_hash(content))) if not created_pages: return { @@ -2187,6 +2256,15 @@ def convert_source_to_wiki_worker(self, source_id, user): "skipped": skipped, } + with db_session() as conn: + rebuild_wiki_directory_structure(conn, source_id, user) + + for doc_id in original_doc_ids: + try: + store.delete_chunk(doc_id) + except Exception as exc: + logging.error(f"Failed deleting original chunk {doc_id}: {exc}") + for page_path, content_hash in created_pages: reembed_wiki_page.delay( source_id, diff --git a/tests/worker/test_convert_source_to_wiki.py b/tests/worker/test_convert_source_to_wiki.py index 8347b55b..12e7de11 100644 --- a/tests/worker/test_convert_source_to_wiki.py +++ b/tests/worker/test_convert_source_to_wiki.py @@ -1,9 +1,10 @@ """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. +The worker reassembles wiki pages from a source's existing vector-store +chunks (grouped by ``metadata.source``) and enqueues a per-page re-embed. +``VectorCreator.create_vectorstore`` and the re-embed task are mocked so no +real store access or embeddings run; the ``sources`` / ``wiki_pages`` rows +are real so the assertions read them back. """ from __future__ import annotations @@ -12,16 +13,15 @@ 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"): +def _seed_source(pg_conn, config=None, file_path=""): src = SourcesRepository(pg_conn).create( "doc-set", user_id="alice", - type="local", + type="crawler", retriever="classic", file_path=file_path, config=config or {}, @@ -30,54 +30,48 @@ def _seed_source(pg_conn, config=None, file_path="inputs/alice/doc-set"): 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 _chunk(text, source=None, doc_id="d", **extra_meta): + metadata = dict(extra_meta) + if source is not None: + metadata["source"] = source + return {"doc_id": doc_id, "text": text, "metadata": metadata} -def _patch_reader(monkeypatch, docs): - from application import worker +def _patch_store(monkeypatch, chunks): + store = MagicMock(name="vectorstore") + store.get_chunks.return_value = chunks + monkeypatch.setattr( + "application.vectorstore.vector_creator.VectorCreator.create_vectorstore", + lambda *a, **kw: store, + ) + return store - reader = MagicMock(name="reader") - reader.load_data.return_value = docs - monkeypatch.setattr(worker, "SimpleDirectoryReader", lambda *a, **kw: reader) - return reader + +def _patch_reembed(monkeypatch): + delay = MagicMock(name="reembed_delay") + monkeypatch.setattr("application.api.user.tasks.reembed_wiki_page.delay", delay) + return delay @pytest.mark.unit class TestConvertSourceToWikiWorker: - def test_one_page_per_text_file( + def test_two_pages_reassembled_from_chunks( 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( + _patch_store( monkeypatch, [ - Document(text="guide body", extra_info={"source": "guide.md"}), - Document( - text="intro body", extra_info={"source": "notes/intro.md"} - ), + _chunk("guide part one.", source="guide.md", doc_id="g1"), + _chunk("guide part two.", source="guide.md", doc_id="g2"), + _chunk("intro body.", source="notes/intro.md", doc_id="i1"), ], ) - delay = MagicMock(name="reembed_delay") - monkeypatch.setattr( - "application.api.user.tasks.reembed_wiki_page.delay", delay - ) + delay = _patch_reembed(monkeypatch) - result = worker.convert_source_to_wiki_worker( - task_self, source_id, "alice" - ) + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") assert result["status"] == "converted" assert result["pages_created"] == 2 @@ -85,83 +79,82 @@ class TestConvertSourceToWikiWorker: 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"]["content"] == "guide part one.\n\nguide part two." assert by_path["/guide.md"]["title"] == "guide.md" + assert by_path["/notes/intro.md"]["content"] == "intro body." 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"} + assert paths == {"/guide.md", "/notes/intro.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( + def test_original_chunks_deleted_after_convert( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + store = _patch_store( + monkeypatch, + [ + _chunk("guide part one.", source="guide.md", doc_id="g1"), + _chunk("guide part two.", source="guide.md", doc_id="g2"), + _chunk("intro body.", source="notes/intro.md", doc_id="i1"), + ], + ) + delay = _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["status"] == "converted" + deleted = {c.args[0] for c in store.delete_chunk.call_args_list} + assert deleted == {"g1", "g2", "i1"} + assert delay.call_count == 2 + + def test_chunk_without_doc_id_not_deleted( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + chunk = {"text": "body", "metadata": {"source": "a.md"}} + store = _patch_store(monkeypatch, [chunk]) + _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["status"] == "converted" + store.delete_chunk.assert_not_called() + + def test_no_pages_path_deletes_nothing( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + store = _patch_store(monkeypatch, []) + _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["status"] == "no_pages" + store.delete_chunk.assert_not_called() + + def test_kind_flipped_and_exposure_defaulted( 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() - ) + _patch_store(monkeypatch, [_chunk("body", source="a.md")]) + _patch_reembed(monkeypatch) worker.convert_source_to_wiki_worker(task_self, source_id, "alice") @@ -179,14 +172,8 @@ class TestConvertSourceToWikiWorker: 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() - ) + _patch_store(monkeypatch, [_chunk("body", source="a.md")]) + _patch_reembed(monkeypatch) worker.convert_source_to_wiki_worker(task_self, source_id, "alice") @@ -195,7 +182,71 @@ class TestConvertSourceToWikiWorker: assert cfg.kind == "wiki" assert cfg.retrieval.exposure == "agentic_tool" - def test_no_pages_leaves_kind_and_structure( + def test_chunk_order_hint_respected( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk("second", source="a.md", doc_id="a2", chunk_index=1), + _chunk("first", source="a.md", doc_id="a1", chunk_index=0), + ], + ) + _patch_reembed(monkeypatch) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + by_path = {p["path"]: p for p in pages} + assert by_path["/a.md"]["content"] == "first\n\nsecond" + + def test_short_overlap_not_trimmed( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk("The end.", source="a.md", doc_id="a1"), + _chunk("The end. More", source="a.md", doc_id="a2"), + ], + ) + _patch_reembed(monkeypatch) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + by_path = {p["path"]: p for p in pages} + assert by_path["/a.md"]["content"] == "The end.\n\nThe end. More" + + def test_long_overlap_trimmed_between_chunks( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + overlap = "x" * 40 + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk("head " + overlap, source="a.md", doc_id="a1"), + _chunk(overlap + " tail", source="a.md", doc_id="a2"), + ], + ) + _patch_reembed(monkeypatch) + + worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + by_path = {p["path"]: p for p in pages} + assert by_path["/a.md"]["content"] == "head " + overlap + "\n\n tail" + + def test_no_chunks_leaves_kind_and_structure( self, pg_conn, patch_worker_db, task_self, monkeypatch ): from application import worker @@ -203,24 +254,15 @@ class TestConvertSourceToWikiWorker: 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, []) + _patch_store(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 - ) + monkeypatch.setattr(worker, "rebuild_wiki_directory_structure", rebuild) + delay = _patch_reembed(monkeypatch) - result = worker.convert_source_to_wiki_worker( - task_self, source_id, "alice" - ) + 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() @@ -228,27 +270,44 @@ class TestConvertSourceToWikiWorker: assert SourceConfig.parse(refreshed.get("config")).kind != "wiki" assert refreshed.get("directory_structure") == original_structure - def test_traversal_filename_skipped( + def test_missing_path_chunk_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( + _patch_store( monkeypatch, [ - Document(text="good", extra_info={"source": "ok.md"}), - Document(text="bad", extra_info={"source": "../evil.md"}), + _chunk("good", source="ok.md"), + _chunk("orphan", source=None), ], ) - monkeypatch.setattr( - "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() - ) + _patch_reembed(monkeypatch) - result = worker.convert_source_to_wiki_worker( - task_self, source_id, "alice" + 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"} + assert any(s["reason"] == "missing path" for s in result["skipped"]) + + def test_invalid_path_chunk_skipped( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk("good", source="ok.md"), + _chunk("bad", source="../evil.md"), + ], ) + _patch_reembed(monkeypatch) + + 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) @@ -256,30 +315,91 @@ class TestConvertSourceToWikiWorker: 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( + def test_path_falls_back_to_filename_then_title( 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( + _patch_store( monkeypatch, - [Document(text="guide body", extra_info={"source": "guide.md"})], - ) - monkeypatch.setattr( - "application.api.user.tasks.reembed_wiki_page.delay", MagicMock() + [ + _chunk("from filename", source=None, filename="byname.md"), + _chunk("from title", source=None, title="bytitle.md"), + ], ) + _patch_reembed(monkeypatch) - result = worker.convert_source_to_wiki_worker( - task_self, source_id, "alice" - ) + 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" + assert result["pages_created"] == 2 + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + assert {p["path"] for p in pages} == {"/byname.md", "/bytitle.md"} + + def test_crawler_chunk_uses_file_path_not_url( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk( + "setup body", + source="https://docs.x.com/guides/setup", + file_path="guides/setup.md", + ), + ], + ) + _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["status"] == "converted" + assert result["skipped"] == [] + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + assert {p["path"] for p in pages} == {"/guides/setup.md"} + + def test_connector_chunks_kept_separate_by_file_name( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [ + _chunk("page a", source="confluence", file_name="Page A", doc_id="a"), + _chunk("page b", source="confluence", file_name="Page B", doc_id="b"), + ], + ) + _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["pages_created"] == 2 + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + assert {p["path"] for p in pages} == {"/Page A", "/Page B"} + + def test_url_only_chunk_normalized_not_skipped( + self, pg_conn, patch_worker_db, task_self, monkeypatch + ): + from application import worker + + source_id = _seed_source(pg_conn) + _patch_store( + monkeypatch, + [_chunk("body", source="https://x.com/a/b")], + ) + _patch_reembed(monkeypatch) + + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") + + assert result["status"] == "converted" + assert result["skipped"] == [] + pages = WikiPagesRepository(pg_conn).list_for_source(source_id) + assert {p["path"] for p in pages} == {"/a/b"} def test_already_wiki_returns_early( self, pg_conn, patch_worker_db, task_self, monkeypatch @@ -287,19 +407,11 @@ class TestConvertSourceToWikiWorker: 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 - ) + store = _patch_store(monkeypatch, [_chunk("body", source="a.md")]) + delay = _patch_reembed(monkeypatch) - result = worker.convert_source_to_wiki_worker( - task_self, source_id, "alice" - ) + result = worker.convert_source_to_wiki_worker(task_self, source_id, "alice") assert result["status"] == "already_wiki" - reader.load_data.assert_not_called() + store.get_chunks.assert_not_called() delay.assert_not_called()