diff --git a/packages/shared-python/shared/tests/test_asset_inline.py b/packages/shared-python/shared/tests/test_asset_inline.py index 75b09c04..7fd44335 100644 --- a/packages/shared-python/shared/tests/test_asset_inline.py +++ b/packages/shared-python/shared/tests/test_asset_inline.py @@ -2,6 +2,15 @@ from __future__ import annotations +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + import pytest from shared.services.retrieval.hydration.asset_inline import ( @@ -161,6 +170,84 @@ async def hydrate_connected_rows(**_kwargs: object) -> list[dict[str, object]]: assert "资产说明" in assembled[0]["content"] +@pytest.mark.asyncio +async def test_assemble_inlines_connected_image_without_placeholder() -> None: + """#206: connect_to image must be inlined, not dropped from both surfaces.""" + rows = [ + { + "chunk_id": "text-1", + "chunk_type": "text", + "content": "The process is illustrated below.", + "chunk_metadata": { + "connect_to": [ + { + "target": "image-1", + "relation": "embeds", + "ref": "[images/flow.png]", + } + ] + }, + }, + { + "chunk_id": "image-1", + "chunk_type": "image", + "content": "Flowchart of the ingestion pipeline.", + "file_path": "images/flow.png", + "asset_url": "https://assets.example.com/job-synth/images/flow.png", + }, + ] + assembled = await assemble_retrieval_results( + rows=rows, + exclude_document_ids=[], + exclude_sections=[], + ) + assert [row["chunk_id"] for row in assembled] == ["text-1"] + content = assembled[0]["content"] + assert "[images/" not in content + assert "[Image: https://assets.example.com/job-synth/images/flow.png]" in content + assert "Flowchart of the ingestion pipeline." in content + assert "The process is illustrated below." in content + + +@pytest.mark.asyncio +async def test_assemble_inserts_image_at_placeholder() -> None: + rows = [ + { + "chunk_id": "text-1", + "chunk_type": "text", + "content": "see [images/a.png] end", + "chunk_metadata": { + "connect_to": [ + { + "target": "img-1", + "relation": "embeds", + "ref": "[images/a.png]", + } + ] + }, + }, + { + "chunk_id": "img-1", + "chunk_type": "image", + "content": "chart summary", + "file_path": "images/a.png", + }, + ] + assembled = await assemble_retrieval_results( + rows=rows, + exclude_document_ids=[], + exclude_sections=[], + ) + assert len(assembled) == 1 + content = assembled[0]["content"] + assert "[images/" not in content + assert content.index("see") < content.index("[Image:") + assert content.index("[Image:") < content.index("end") + assert "[Image: images/a.png]" in content + assert "chart summary" in content + assert content.count("[Image:") == 1 + + def test_node_unit_span_inlines_section_assets() -> None: provider = KnowhereProvider( doc_id="doc-1",