YoVinchen
ecdb10c37a
Some checks failed
verify / verify (push) Has been cancelled
Separate local parsing from model indexing, bind review decisions to immutable manifests, persist vectors behind active profiles, and expose retrieval, chat, evaluation, and document workflows through the React workbench. Constraint: Live Bailian authentication currently fails for all three configured capabilities Rejected: Direct upload-to-embedding flow | bypasses local review and manifest binding Confidence: high Scope-risk: broad Directive: Keep private-data deployment blocked until authentication, RBAC, and separate database roles land Tested: make verify; fresh and replay Docker document smoke; worker recovery smoke; frozen synthetic evaluation; migration 0003-0004 roundtrip Not-tested: Successful live Bailian calls, OCR, real multi-user authorization
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass, field, replace
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from app.core.config import Settings
|
|
from app.persistence.job_queue import BackgroundJob, JobLease
|
|
from app.services.indexing import DocumentIndexingService, IndexingResult
|
|
from app.workers.indexing_jobs import (
|
|
InvalidIndexingJobError,
|
|
build_embed_document_handler,
|
|
build_indexing_handlers,
|
|
)
|
|
|
|
NOW = datetime(2026, 7, 13, 8, 0, tzinfo=UTC)
|
|
JOB_ID = uuid.UUID("10000000-0000-0000-0000-000000000001")
|
|
DOCUMENT_VERSION_ID = uuid.UUID("20000000-0000-0000-0000-000000000002")
|
|
LEASE_TOKEN = uuid.UUID("30000000-0000-0000-0000-000000000003")
|
|
|
|
|
|
def _job() -> BackgroundJob:
|
|
lease = JobLease(JOB_ID, "embedding-worker-a", LEASE_TOKEN)
|
|
return BackgroundJob(
|
|
id=JOB_ID,
|
|
job_type="EMBED_DOCUMENT",
|
|
required_capability="embedding",
|
|
resource_type="document_version",
|
|
resource_id=DOCUMENT_VERSION_ID,
|
|
idempotency_key="embed-document:version:profile",
|
|
payload={"document_version_id": str(DOCUMENT_VERSION_ID)},
|
|
stage="EMBEDDING",
|
|
progress=20,
|
|
priority=0,
|
|
attempt=1,
|
|
max_attempts=3,
|
|
run_after=NOW,
|
|
lease_until=NOW + timedelta(seconds=60),
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
lease=lease,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class SpyIndexingService:
|
|
calls: list[tuple[JobLease, uuid.UUID, uuid.UUID]] = field(default_factory=list)
|
|
|
|
async def index_document_version(
|
|
self,
|
|
*,
|
|
lease: JobLease,
|
|
document_version_id: uuid.UUID,
|
|
trace_id: uuid.UUID,
|
|
) -> IndexingResult:
|
|
self.calls.append((lease, document_version_id, trace_id))
|
|
return IndexingResult(
|
|
document_version_id=document_version_id,
|
|
profile_hash="a" * 64,
|
|
expected_count=1,
|
|
ready_count=1,
|
|
cache_hit_count=0,
|
|
newly_embedded_count=1,
|
|
provider_call_count=1,
|
|
activated=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handler_validates_payload_resource_and_passes_exact_lease_and_job_trace() -> None:
|
|
service = SpyIndexingService()
|
|
handler = build_embed_document_handler(cast(DocumentIndexingService, service))
|
|
job = _job()
|
|
|
|
await handler(job)
|
|
|
|
assert service.calls == [(job.lease, DOCUMENT_VERSION_ID, JOB_ID)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"job",
|
|
[
|
|
replace(_job(), job_type="PARSE_DOCUMENT"),
|
|
replace(_job(), required_capability="document_parse"),
|
|
replace(_job(), resource_type="document"),
|
|
replace(_job(), resource_id=uuid.uuid4()),
|
|
replace(_job(), payload={}),
|
|
replace(_job(), payload={"document_version_id": "not-a-uuid"}),
|
|
replace(
|
|
_job(),
|
|
lease=JobLease(uuid.uuid4(), "embedding-worker-a", LEASE_TOKEN),
|
|
),
|
|
],
|
|
)
|
|
async def test_invalid_job_envelope_never_reaches_service(job: BackgroundJob) -> None:
|
|
service = SpyIndexingService()
|
|
handler = build_embed_document_handler(cast(DocumentIndexingService, service))
|
|
|
|
with pytest.raises(InvalidIndexingJobError) as captured:
|
|
await handler(job)
|
|
|
|
assert service.calls == []
|
|
assert str(DOCUMENT_VERSION_ID) not in str(captured.value)
|
|
assert "not-a-uuid" not in str(captured.value)
|
|
|
|
|
|
def test_production_handler_registration_requires_worker_gateway_identity(tmp_path: Path) -> None:
|
|
password = tmp_path / "postgres-password"
|
|
password.write_text("synthetic-test-password", encoding="utf-8")
|
|
worker_settings = Settings(
|
|
postgres_password_file=password,
|
|
model_gateway_caller="worker",
|
|
)
|
|
|
|
handlers = build_indexing_handlers(worker_settings)
|
|
|
|
assert set(handlers) == {"EMBED_DOCUMENT"}
|
|
assert callable(handlers["EMBED_DOCUMENT"])
|
|
|
|
api_settings = Settings(
|
|
postgres_password_file=password,
|
|
model_gateway_caller="api",
|
|
)
|
|
with pytest.raises(ValueError, match="MODEL_GATEWAY_CALLER=worker"):
|
|
build_indexing_handlers(api_settings)
|