Function _run

tests/parser/test_parse_native_lightrag_e2e.py:103–183

()

    therefore identical chunk_args_hash on chunk-0."""

    async def _run():
        input_dir = tmp_path / "input"
        input_dir.mkdir()
        monkeypatch.setenv("INPUT_DIR", str(input_dir))

        source_path = input_dir / "stable.docx"
        source_path.write_bytes(b"fake docx bytes")

        # Stub extract_docx_blocks at the adapter so the upstream DOCX
        # parser is never invoked. The adapter still does all the
        # LightRAG-specific writing — that is what we want under test.
        stable_blocks = [
            _block(
                "Title\nFirst paragraph body.\nSecond paragraph body.",
                heading="Title",
                level=1,
            ),
        ]

        def _stub_extract(file_path, drawing_context=None, **kwargs):
            return [dict(b) for b in stable_blocks]

        monkeypatch.setattr(
            "lightrag.parser.docx.parse_document.extract_docx_blocks",
            _stub_extract,
        )

        rag = _MiniRag(tmp_path / "work")

        # ---- First parse ----
        # parse_native archives the source after writing, so re-create it
        # before the second parse for a fair comparison.
        result1 = await _parse_via_registry(
            rag,
            "native",
            "doc-stable",
            str(source_path),
            {"parse_format": FULL_DOCS_FORMAT_PENDING_PARSE, "content": ""},
        )
        merged1 = result1["content"]
        assert merged1, "first parse produced empty merged_text"

        # ---- Second parse ----
        # Restore the source file (archive moved it), reset the in-memory
        # full_docs row, and remove the parsed_dir so the writer rewrites
        # both meta (with a fresh parse_time) and content lines.
        source_path.write_bytes(b"fake docx bytes")
        rag.full_docs.data.clear()
        parsed_artifact_dir = input_dir / PARSED_DIR_NAME / f"{source_path.name}.parsed"
        if parsed_artifact_dir.exists():
            import shutil

            shutil.rmtree(parsed_artifact_dir)

        result2 = await _parse_via_registry(
            rag,
            "native",
            "doc-stable",