Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions apps/api/app/services/documents/lifecycle_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,15 @@ def _document_chunk_asset_url(
return None

try:
if not result_storage.verify_raw_exists(
job_id=job_id,
relative_path=file_path,
):
logger.warning(
f"Skipping asset URL for missing chunk artifact: "
f"job_id={job_id}, file_path={file_path}"
)
return None
return result_storage.generate_artifact_url(
job_id=job_id,
artifact_ref=file_path,
Expand Down
3 changes: 3 additions & 0 deletions apps/api/tests/contract/test_demo_documents_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,9 @@ def normalize_artifact_ref(self, artifact_ref: str | None) -> str | None:
return normalized
return None

def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
return relative_path in self.raw_files_by_job_id.get(job_id, set())

def generate_artifact_url(
self,
*,
Expand Down
24 changes: 24 additions & 0 deletions apps/api/tests/contract/test_documents_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -533,6 +533,22 @@ def _upload_page_citation_asset(*, job_id: str, artifact_ref: str) -> None:
asset_path.unlink(missing_ok=True)


def _upload_chunk_asset(*, job_id: str, artifact_ref: str) -> None:
from shared.services.storage.result_storage import JobResultStorage

suffix = Path(artifact_ref).suffix or ".bin"
asset_path = Path("/tmp") / f"knowhere-contract-chunk-asset-{uuid4().hex}{suffix}"
asset_path.write_bytes(b"<table><tr><td>contract chunk asset</td></tr></table>")
try:
JobResultStorage().upload_raw_file(
job_id=job_id,
relative_path=artifact_ref,
local_file_path=str(asset_path),
)
finally:
asset_path.unlink(missing_ok=True)


@pytest.mark.asyncio
async def test_should_list_only_the_authenticated_users_documents_for_the_effective_namespace(
developer_api_client_factory: Callable[
Expand Down Expand Up @@ -992,6 +1008,10 @@ async def test_should_include_media_asset_urls_in_document_chunk_list_when_reque
},
],
)
_upload_chunk_asset(
job_id=revision["job_id"],
artifact_ref="tables/table-1.html",
)
response = await api_client.get(
f"/api/v1/documents/{document_id}/chunks",
params={
Expand Down Expand Up @@ -1167,6 +1187,10 @@ async def test_should_return_one_document_chunk_by_document_chunk_id(
}
],
)
_upload_chunk_asset(
job_id=revision["job_id"],
artifact_ref="images/figure-1.png",
)
response = await api_client.get(
f"/api/v1/documents/{document_id}/chunks/{chunk_id}",
params={"include_asset_urls": "true"},
Expand Down
3 changes: 3 additions & 0 deletions apps/api/tests/contract/test_retrieval_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,9 @@ def normalize_artifact_ref(self, artifact_ref: str | None) -> str | None:
return None
return normalized

def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
return True

def fake_get_result_storage() -> FakeResultStorage:
return FakeResultStorage()

Expand Down
177 changes: 177 additions & 0 deletions apps/api/tests/unit/test_document_chunk_asset_url.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,177 @@
"""Unit tests for document chunk asset URL generation."""

from __future__ import annotations

import sys
from pathlib import Path
from types import ModuleType
from typing import Any
from unittest.mock import patch

import pytest
from tests.support.import_environment import (
configure_import_environment,
ensure_import_paths,
)

# Defer importing apps/api `app` until test bodies run. Module-level imports
# would cache API's `app` package and break apps/worker contract collection in
# the same pytest process (both packages are named `app`).
configure_import_environment()
ensure_import_paths()

_API_ROOT = str(Path(__file__).resolve().parents[2])


def _is_api_app_module(module: ModuleType | None) -> bool:
if module is None:
return False
module_file = getattr(module, "__file__", None)
if isinstance(module_file, str) and module_file.startswith(_API_ROOT):
return True
module_paths = getattr(module, "__path__", ())
try:
return any(str(path).startswith(_API_ROOT) for path in module_paths)
except KeyError:
return False


def _drop_api_app_modules() -> None:
for module_name in sorted(sys.modules, key=len, reverse=True):
if module_name != "app" and not module_name.startswith("app."):
continue
if _is_api_app_module(sys.modules.get(module_name)):
sys.modules.pop(module_name, None)


def _prioritize_api_import_root() -> None:
"""Keep apps/api ahead of apps/worker for the shared `app` package name."""
ensure_import_paths()
if _API_ROOT in sys.path:
sys.path.remove(_API_ROOT)
sys.path.insert(0, _API_ROOT)


@pytest.fixture(autouse=True)
def _clear_api_app_modules_after_unit_test():
"""Avoid leaving API's `app` package cached for later worker contract tests."""
yield
_drop_api_app_modules()


class FakeResultStorage:
def __init__(self, *, exists: bool = True) -> None:
self.exists = exists
self.checked: list[str] = []

def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
self.checked.append(f"{job_id}/{relative_path}")
return self.exists

def generate_artifact_url(
self,
*,
job_id: str,
artifact_ref: str,
expires_in: int = 3600,
) -> str | None:
return f"https://assets.example.test/{job_id}/{artifact_ref}"


def _document_chunk_asset_url(**kwargs: Any) -> str | None:
_prioritize_api_import_root()
_drop_api_app_modules()
from app.services.documents.lifecycle_service import (
_document_chunk_asset_url,
)

return _document_chunk_asset_url(**kwargs)


def test_asset_url_generated_when_raw_file_exists() -> None:
storage = FakeResultStorage(exists=True)

url = _document_chunk_asset_url(
chunk_type="table",
job_id="job-1",
file_path="tables/table-1 Test.html",
include_asset_urls=True,
result_storage=storage,
)

assert url == "https://assets.example.test/job-1/tables/table-1 Test.html"
assert storage.checked == ["job-1/tables/table-1 Test.html"]


def test_asset_url_skipped_when_raw_file_missing() -> None:
storage = FakeResultStorage(exists=False)

url = _document_chunk_asset_url(
chunk_type="table",
job_id="job-1",
file_path="tables/table-1 Test.html",
include_asset_urls=True,
result_storage=storage,
)

assert url is None
assert storage.checked == ["job-1/tables/table-1 Test.html"]


def test_asset_url_falls_back_on_verification_error() -> None:
class ExplodingStorage:
def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
raise RuntimeError("storage unavailable")

def generate_artifact_url(
self,
*,
job_id: str,
artifact_ref: str,
expires_in: int = 3600,
) -> str | None:
return f"https://assets.example.test/{job_id}/{artifact_ref}"

with patch(
"app.services.documents.lifecycle_service.logger.warning"
) as mock_warning:
url = _document_chunk_asset_url(
chunk_type="image",
job_id="job-1",
file_path="images/page-1.png",
include_asset_urls=True,
result_storage=ExplodingStorage(),
)

assert url is None
assert mock_warning.call_count == 1


def test_asset_url_skipped_for_non_media_chunk_types() -> None:
storage = FakeResultStorage(exists=True)

url = _document_chunk_asset_url(
chunk_type="text",
job_id="job-1",
file_path="tables/table-1 Test.html",
include_asset_urls=True,
result_storage=storage,
)

assert url is None
assert storage.checked == []


def test_asset_url_skipped_when_include_asset_urls_false() -> None:
storage = FakeResultStorage(exists=True)

url = _document_chunk_asset_url(
chunk_type="table",
job_id="job-1",
file_path="tables/table-1 Test.html",
include_asset_urls=False,
result_storage=storage,
)

assert url is None
assert storage.checked == []
Loading