307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""Synthetic coordinates/evidence test fusion semantics, not domain/model quality."""
|
|
|
|
import math
|
|
from dataclasses import replace
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select, text
|
|
from sqlalchemy.orm import Session
|
|
from test_indexing import build
|
|
from test_retrieval import add_item
|
|
from test_vectors import ToyProvider, text_version, vector_build, vector_publish
|
|
|
|
from app.models.retrieval import RetrievalRun
|
|
from app.schemas.hybrid import RankedCandidate
|
|
from app.services import vectors
|
|
from app.services.embedding import EmbeddingError
|
|
from app.services.embedding_runtime import get_embedding_provider
|
|
from app.services.hybrid import fuse
|
|
from app.services.indexing import publish_index
|
|
|
|
|
|
@pytest.fixture
|
|
def provider(client):
|
|
value = ToyProvider()
|
|
client.app.dependency_overrides[get_embedding_provider] = lambda: value
|
|
return value
|
|
|
|
|
|
def search(client, path="hybrid", **kwargs):
|
|
return client.post(f"/api/retrieval/{path}", json={"query": "query", **kwargs})
|
|
|
|
|
|
def publish_vectors(client):
|
|
text_ = text_version(client)
|
|
version = vector_build(client, text_)
|
|
assert vector_publish(client, version).status_code == 200
|
|
return text_, version
|
|
|
|
|
|
def test_rrf_hand_calculation_absent_leg_and_raw_score_independence():
|
|
a, b, c = uuid4(), uuid4(), uuid4()
|
|
lexical = [
|
|
RankedCandidate(chunk_id=a, rank=1, score=1000),
|
|
RankedCandidate(chunk_id=b, rank=2, score=1),
|
|
]
|
|
vector = [
|
|
RankedCandidate(chunk_id=b, rank=1, score=-0.9),
|
|
RankedCandidate(chunk_id=c, rank=2, score=-1),
|
|
]
|
|
scores = fuse(lexical, vector)
|
|
assert scores == {str(a): 1 / 61, str(b): 1 / 62 + 1 / 61, str(c): 1 / 62}
|
|
for candidate in lexical + vector:
|
|
candidate.score *= 10000
|
|
assert fuse(lexical, vector) == scores
|
|
assert fuse([], []) == {}
|
|
|
|
|
|
def test_same_chunk_scope_standalone_rankings_formula_and_history(client, provider):
|
|
items = [
|
|
add_item(client, content="alpha query extra words words"),
|
|
add_item(client, content="query"),
|
|
add_item(client, content="beta"),
|
|
]
|
|
add_item(client, status="pending", content="query alpha")
|
|
text_, version = publish_vectors(client)
|
|
response = search(client, top_k=3, candidate_k=3)
|
|
assert response.status_code == 200, response.text
|
|
result = response.json()
|
|
lexical = search(client, "chunks", top_k=3).json()
|
|
vector = search(client, "vector", top_k=3).json()
|
|
assert result["corpus_size"] == lexical["corpus_size"] == vector["corpus_size"] == 3
|
|
assert (
|
|
result["text_index_version_id"]
|
|
== lexical["text_index_version_id"]
|
|
== vector["text_index_version_id"]
|
|
== text_["id"]
|
|
)
|
|
assert result["vector_version_id"] == version["id"]
|
|
assert result["config"]["space"]["id"] == provider.spec.space_id
|
|
assert result["config"]["lexical"]["title_weight"] == 0
|
|
assert result["config"]["lexical"]["k1"] == 1.2
|
|
assert result["config"]["lexical"]["b"] == 0.75
|
|
assert result["config"]["chunking"] == text_["config"]
|
|
assert result["lexical_total_hits"] == 2 and result["vector_total_hits"] == 3
|
|
assert [c["chunk_id"] for c in result["lexical_candidates"]] == [
|
|
c["chunk_id"] for c in lexical["results"]
|
|
]
|
|
assert [c["chunk_id"] for c in result["vector_candidates"]] == [
|
|
c["chunk_id"] for c in vector["results"]
|
|
]
|
|
# Independently calculate positive-log-IDF BM25 on bodies of lengths 5, 1, 1.
|
|
expected = math.log(1 + 1.5 / 2.5) * 2.2 / (1 + 1.2 * (0.25 + 0.75 / (7 / 3)))
|
|
assert lexical["results"][0]["knowledge_id"] == items[1]["id"]
|
|
assert lexical["results"][0]["score"] == pytest.approx(expected)
|
|
for hit in result["results"]:
|
|
left = next(
|
|
(c for c in result["lexical_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
|
|
)
|
|
right = next(
|
|
(c for c in result["vector_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
|
|
)
|
|
assert hit["score"] == pytest.approx(
|
|
(1 / (60 + left["rank"]) if left else 0) + (1 / (60 + right["rank"]) if right else 0)
|
|
)
|
|
assert hit["score"] == hit["lexical_contribution"] + hit["vector_contribution"]
|
|
assert hit["source_url"] == "https://example.com/"
|
|
assert search(client, candidate_k=3, top_k=3).json()["results"] == result["results"]
|
|
for value in (result, lexical, vector):
|
|
assert client.get(f"/api/retrieval/runs/{value['run_id']}").json() == value
|
|
old = client.post("/api/retrieval", json={"query": "query"}).json()
|
|
assert old["mode"] == "keyword" and "chunk_id" not in old["results"][0]
|
|
history = client.get("/api/retrieval/runs?page_size=100").json()
|
|
assert {r["mode"] for r in history["items"]} == {
|
|
"keyword",
|
|
"vector",
|
|
"keyword_chunks",
|
|
"hybrid",
|
|
}
|
|
client.cookies.clear()
|
|
assert search(client).status_code == search(client, "chunks").status_code == 200
|
|
assert client.get("/api/retrieval/runs").status_code == 401
|
|
|
|
|
|
def test_candidate_cutoffs_disjoint_legs_and_deterministic_tie(client, provider):
|
|
first = add_item(client, content="alpha")
|
|
second = add_item(client, content="query")
|
|
publish_vectors(client)
|
|
small = search(client, candidate_k=1, top_k=1).json()
|
|
assert small["total_hits"] == 2 and len(small["results"]) == 1
|
|
assert small["results"][0]["knowledge_id"] == first["id"]
|
|
assert small["results"][0]["lexical_rank"] is None
|
|
assert small["results"][0]["lexical_score"] is None
|
|
assert small["results"][0]["lexical_contribution"] == 0
|
|
assert small["results"][0]["score"] == 1 / 61
|
|
full = search(client, candidate_k=2, top_k=2).json()
|
|
assert full["results"][0]["knowledge_id"] == second["id"]
|
|
assert full["results"][0]["lexical_rank"] == 1 and full["results"][0]["vector_rank"] == 2
|
|
|
|
|
|
def test_chunk_bodies_offsets_overlap_and_category_apply_before_both_rankings(client, provider):
|
|
item = add_item(client, title="uniquetitle", content="alpha query 文本。" * 30, category="交通")
|
|
add_item(client, content="query", category="其他")
|
|
publish_vectors(client)
|
|
result = search(client, category="交通", top_k=20, candidate_k=20).json()
|
|
assert result["corpus_size"] > 1 and result["total_hits"] == result["corpus_size"]
|
|
assert len({h["chunk_id"] for h in result["results"]}) == len(result["results"])
|
|
for hit in result["results"]:
|
|
assert hit["knowledge_id"] == item["id"]
|
|
assert hit["snippet"] == item["content"][hit["snippet_start"] : hit["snippet_end"]]
|
|
assert search(client, "chunks", query="uniquetitle").json()["status"] == "no_evidence"
|
|
empty = search(client, category="不存在分类").json()
|
|
assert empty["status"] == "no_evidence" and empty["results"] == []
|
|
assert empty["lexical_candidates"] == empty["vector_candidates"] == []
|
|
|
|
|
|
def test_empty_lexical_leg_and_empty_corpus_are_explicit(client, provider):
|
|
add_item(client, content="alpha")
|
|
publish_vectors(client)
|
|
for query in ["unmatchedword", "！！"]:
|
|
result = search(client, query=query).json()
|
|
assert result["lexical_candidates"] == [] and result["status"] == "ok"
|
|
assert result["results"][0]["score"] == 1 / 61
|
|
assert search(client, "chunks", query="！！").status_code == 422
|
|
|
|
|
|
def test_empty_published_corpus(client, provider):
|
|
publish_vectors(client)
|
|
for path in ["chunks", "hybrid"]:
|
|
result = search(client, path).json()
|
|
assert result["status"] == "no_evidence" and result["total_hits"] == 0
|
|
|
|
|
|
def test_keyword_chunks_need_no_runtime_or_vector_publication(client):
|
|
add_item(client, content="query")
|
|
assert search(client, "chunks").status_code == 409
|
|
text_version(client)
|
|
assert search(client, "chunks").status_code == 200
|
|
assert search(client).status_code == 503
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"candidate_k": 1, "top_k": 2},
|
|
{"candidate_k": 201},
|
|
{"candidate_k": 0},
|
|
{"top_k": 21},
|
|
{"top_k": 0},
|
|
{"mode": "vector"},
|
|
{"rrf_k": 10},
|
|
{"query": " "},
|
|
{"query": "x" * 501},
|
|
],
|
|
)
|
|
def test_request_budget_and_fixed_config_validation(client, provider, payload):
|
|
assert search(client, **payload).status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["catalog", "text", "vector"])
|
|
def test_query_encoding_releases_locks_and_rechecks_publication(
|
|
client, provider, migrated_database, change
|
|
):
|
|
item = add_item(client, content="alpha query")
|
|
text_, first = publish_vectors(client)
|
|
replacement_text = build(client)
|
|
replacement_vector = vector_build(client, text_)
|
|
|
|
def mutate(query):
|
|
assert query
|
|
with Session(migrated_database) as other:
|
|
other.execute(text("SET LOCAL lock_timeout='1s'"))
|
|
if change == "catalog":
|
|
state = vectors.locked_state(other)
|
|
state.catalog_revision += 1
|
|
other.execute(
|
|
text(
|
|
"UPDATE knowledge_items SET content='changed synthetic evidence' WHERE id=:id"
|
|
),
|
|
{"id": item["id"]},
|
|
)
|
|
other.commit()
|
|
elif change == "text":
|
|
publish_index(other, UUID(replacement_text["id"]), 1, first["created_by"])
|
|
else:
|
|
vectors.publish_vectors(
|
|
other, UUID(replacement_vector["id"]), 1, first["created_by"]
|
|
)
|
|
|
|
provider.callback = mutate
|
|
assert search(client).status_code == 409
|
|
with Session(migrated_database) as db:
|
|
assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["body", "chunks", "vectors", "space"])
|
|
def test_corrupt_or_incompatible_snapshot_fails_closed(client, provider, migrated_database, change):
|
|
add_item(client, content="alpha query")
|
|
publish_vectors(client)
|
|
if change == "space":
|
|
provider.spec = replace(provider.spec, revision="b" * 40)
|
|
else:
|
|
with migrated_database.begin() as connection:
|
|
connection.execute(
|
|
text(
|
|
{
|
|
"body": "UPDATE knowledge_items SET content='out of band change'",
|
|
"chunks": "UPDATE chunks SET text='tampered'",
|
|
"vectors": "DELETE FROM chunk_vectors",
|
|
}[change]
|
|
)
|
|
)
|
|
assert search(client).status_code == 409
|
|
with Session(migrated_database) as db:
|
|
assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("failure,status", [("token", 422), ("runtime", 503)])
|
|
def test_model_failure_has_no_fallback_or_partial_history(
|
|
client, provider, migrated_database, failure, status
|
|
):
|
|
add_item(client, content="alpha query")
|
|
publish_vectors(client)
|
|
|
|
def fail(*args, **kwargs):
|
|
if failure == "token":
|
|
raise EmbeddingError("token budget")
|
|
raise OSError("private-runtime-path")
|
|
|
|
provider.encode = fail
|
|
response = search(client)
|
|
assert response.status_code == status and "private-runtime-path" not in response.text
|
|
with Session(migrated_database) as db:
|
|
assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
|
|
|
|
|
|
def test_failed_history_insert_rolls_back_both_leg_result(client, provider, migrated_database):
|
|
add_item(client, content="alpha query")
|
|
publish_vectors(client)
|
|
with migrated_database.begin() as connection:
|
|
connection.execute(
|
|
text("ALTER TABLE retrieval_runs ADD CONSTRAINT test_no_runs CHECK (false) NOT VALID")
|
|
)
|
|
assert search(client).status_code == 409
|
|
with Session(migrated_database) as db:
|
|
assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
|
|
|
|
|
|
def test_stale_history_survives_evidence_change_and_capacity_is_explicit(
|
|
client, provider, migrated_database, monkeypatch
|
|
):
|
|
item = add_item(client, content="alpha query")
|
|
publish_vectors(client)
|
|
result = search(client).json()
|
|
monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0)
|
|
assert search(client).status_code == search(client, "chunks").status_code == 503
|
|
monkeypatch.undo()
|
|
assert (
|
|
client.patch(
|
|
f"/api/knowledge/{item['id']}",
|
|
json={"expected_version": item["version"], "content": "updated synthetic body"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert search(client).status_code == search(client, "chunks").status_code == 409
|
|
assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result
|