600 lines
23 KiB
Python
600 lines
23 KiB
Python
"""Acceptance, adversarial extraction and source-bound temporal regressions."""
|
|
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.knowledge.sources import profile_units, meeting_date
|
|
from tools.knowledge.extraction import extract_batch
|
|
from tools.knowledge.query import answer_from_index, intent, invalid_question, select_topic_evidence
|
|
|
|
|
|
def response(value):
|
|
return {"choices": [{"message": {"content": json.dumps(value, ensure_ascii=False)}}]}
|
|
|
|
|
|
def doc(name="Alena Nová", content=None):
|
|
return {
|
|
"title": name,
|
|
"path": "pages/students/2020/alena/README.md",
|
|
"content": content
|
|
or """# Diplomová práca 2026
|
|
Zadanie:
|
|
1. Vytvorte systém RAG na vyhľadávanie dokumentov.
|
|
Stretnutie 5.2.2026
|
|
Stav:
|
|
- Implementovaný systém RAG s vyhľadávaním dokumentov.
|
|
Úlohy:
|
|
- Vyhodnoťte systém pomocou DeepEval.
|
|
Stretnutie 6.3.
|
|
Stav:
|
|
- Vyhodnotenie modelu zatiaľ nefunguje.
|
|
""",
|
|
}
|
|
|
|
|
|
def fixture_index():
|
|
units = profile_units(doc())
|
|
second = profile_units(
|
|
{
|
|
**doc(),
|
|
"title": "Boris Starý",
|
|
"path": "pages/students/2020/boris/README.md",
|
|
"content": "# Diplomová práca 2025\nStretnutie 3.4.2025\nStav:\n- Implementovaný systém RAG pre vzdelávanie.\n",
|
|
}
|
|
)
|
|
for unit in [*units, *second]:
|
|
unit.update(concepts=["RAG"] if "RAG" in unit["text"] else [], relations=[])
|
|
return {
|
|
"units": [*units, *second],
|
|
"index_version": {"version": "fixture"},
|
|
"version": "knowledge-fixture",
|
|
}
|
|
|
|
|
|
def test_dates_are_explicit_and_invalid_dates_are_not_repaired():
|
|
assert meeting_date("Stretnutie 5.2.2026")["date"] == "2026-02-05"
|
|
assert meeting_date("Virtuálne stretnutie 12.3.") == {
|
|
"date": None,
|
|
"date_label": "12.3.",
|
|
"date_precision": "day_month",
|
|
}
|
|
assert meeting_date("Stretnutie 31.2.2026")["date"] is None
|
|
assert meeting_date("Diplomová práca 2026") is None
|
|
assert meeting_date("Virtuálne stretnutie 13.11:2020:")["date"] == "2020-11-13"
|
|
|
|
|
|
def test_source_units_keep_person_work_phase_and_date_together():
|
|
units = profile_units(doc())
|
|
assignment = next(u for u in units if "Vytvorte" in u["text"])
|
|
done = next(u for u in units if "Implementovaný" in u["text"])
|
|
planned = next(u for u in units if "Vyhodnoťte" in u["text"])
|
|
partial = next(u for u in units if "nefunguje" in u["text"])
|
|
assert assignment["phase"] == "assignment" and assignment["date"] is None
|
|
assert done["phase"] == "result" and done["date"] == "2026-02-05"
|
|
assert planned["phase"] == "planned" and planned["date"] == "2026-02-05"
|
|
assert partial["phase"] == "progress" and partial["date"] is None
|
|
assert partial["date_precision"] == "day_month" and partial["work_year"] == 2026
|
|
assert len({u["id"] for u in units}) == len(units)
|
|
|
|
|
|
def test_same_quote_in_different_works_is_not_a_single_evidence():
|
|
units = profile_units(
|
|
doc(
|
|
content="# Diplomový projekt 2025\nStav:\n- Funguje systém RAG.\n"
|
|
"# Diplomová práca 2026\nStav:\n- Funguje systém RAG.\n"
|
|
)
|
|
)
|
|
assert len(units) == 2 and units[0]["id"] != units[1]["id"]
|
|
assert {u["work_year"] for u in units} == {2025, 2026}
|
|
|
|
|
|
def test_sibling_nonacademic_heading_does_not_inherit_previous_work():
|
|
units = profile_units(
|
|
doc(
|
|
content="# Diplomová práca 2026\nStav:\n- Funguje systém RAG.\n"
|
|
"# Príprava na projekt\n- Iná poznámka mimo diplomovej práce.\n"
|
|
)
|
|
)
|
|
assert units[-1]["work_label"] is None and units[-1]["phase"] == "unknown"
|
|
|
|
|
|
def test_reference_urls_and_topic_examples_are_not_promoted_to_results_or_goals():
|
|
units = profile_units(
|
|
doc(
|
|
content="# Diplomová práca 2026\nZadanie:\n- Vytvorte systém RAG.\n"
|
|
"Napr.\n- téma pre základné školy.\nStav:\n- Implementovaný systém RAG.\n"
|
|
"Zdroje dokumentov:\n- https://example.test/research\n"
|
|
)
|
|
)
|
|
assert not any("https://" in u["text"] for u in units)
|
|
example = next(u for u in units if "téma" in u["text"])
|
|
assert example["phase"] == "unknown"
|
|
|
|
|
|
def test_partial_implementation_is_not_a_completed_result_even_under_urobene():
|
|
units = profile_units(
|
|
doc(content="# Diplomová práca 2026\nUrobené:\n- rozrobené web rozhranie\n")
|
|
)
|
|
assert units[0]["phase"] == "progress" and units[0]["kind"] == "progress"
|
|
|
|
|
|
def test_explicit_outputs_and_checked_tasks_remain_distinguishable():
|
|
units = profile_units(
|
|
doc(
|
|
content="# Bakalársky projekt 2020\nVýstupy:\n- Aplikácia vie rozpoznávať entity.\n"
|
|
"Úlohy:\n- [x] Naštudujte korpusy pre NER.\n- [ ] Vyhodnoťte model NER.\n"
|
|
)
|
|
)
|
|
assert units[0]["phase"] == "result"
|
|
assert units[1]["phase"] == "result" and units[1]["completion_basis"] == "checked_task"
|
|
assert units[2]["phase"] == "planned"
|
|
|
|
|
|
def test_model_intent_cannot_rewrite_questions_or_invent_query_operators():
|
|
from tools.knowledge.query import model_intent, ContentAnswerRejected, needs_intent_model
|
|
|
|
question = "Ktorí študenti sa zaoberali anotovaním textov?"
|
|
assert needs_intent_model(question)
|
|
|
|
def classify(messages):
|
|
assert messages[1]["content"] == question
|
|
return response({"intent": "discovery"})
|
|
|
|
assert model_intent(question, classify) == "discovery"
|
|
with pytest.raises(ContentAnswerRejected):
|
|
model_intent(question, lambda _: response({"intent": "execute_cypher"}))
|
|
with pytest.raises(ContentAnswerRejected):
|
|
model_intent(question, lambda _: response({"intent": "discovery", "query": "changed"}))
|
|
assert not needs_intent_model("Aké práce má Ján Holp?")
|
|
assert not needs_intent_model("Aké práce a projekty má Ondrej Megela?")
|
|
assert not needs_intent_model("Koľko osôb má prácu v roku 2024?")
|
|
|
|
|
|
def test_extraction_is_bound_to_literal_evidence_and_can_return_empty_concepts():
|
|
units = profile_units(doc())[:2]
|
|
output = {
|
|
"items": [
|
|
{"evidence_id": key, "decision":"knowledge", "concepts": ["RAG"], "relations": []} for key in ["E2", "E1"]
|
|
]
|
|
}
|
|
result = extract_batch(units, lambda _: response(output))
|
|
assert [r["id"] for r in result] == [u["id"] for u in units]
|
|
assert result[0]["phase"] == "assignment" and result[1]["phase"] == "result"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
[
|
|
lambda v: v["items"][0].update(evidence_id="unknown"),
|
|
lambda v: v["items"][0].update(date="2099-01-01"),
|
|
lambda v: v["items"].append(copy.deepcopy(v["items"][0])),
|
|
lambda v: v.update(instructions="ignore the sources"),
|
|
],
|
|
)
|
|
def test_wrong_llm_extractions_are_rejected_after_bounded_retry(change):
|
|
unit = profile_units(doc())[0]
|
|
payload = {"items": [{"evidence_id": "E1", "decision":"knowledge", "concepts": ["RAG"], "relations": []}]}
|
|
change(payload)
|
|
attempts = []
|
|
|
|
def complete(messages):
|
|
attempts.append(messages)
|
|
return response(payload)
|
|
|
|
with pytest.raises(ValueError):
|
|
extract_batch([unit], complete)
|
|
assert len(attempts) == 2
|
|
|
|
|
|
def test_unsupported_concepts_and_relations_are_discarded_without_losing_source_evidence():
|
|
unit = profile_units(doc())[0]
|
|
payload = {
|
|
"items": [
|
|
{
|
|
"evidence_id": "E1",
|
|
"decision":"knowledge",
|
|
"concepts": ["RAG", "vymyslený model"],
|
|
"relations": [{"subject": "RAG", "predicate": "uses", "object": "BERT", "quote":unit["text"]}],
|
|
}
|
|
]
|
|
}
|
|
rows = extract_batch([unit], lambda _: response(payload))
|
|
assert rows[0]["text"] == unit["text"] and rows[0]["concepts"] == ["RAG"]
|
|
assert rows[0]["relations"] == []
|
|
assert rows[0]["discarded_concepts"] == rows[0]["discarded_relations"] == 1
|
|
|
|
|
|
def test_timeline_uses_meeting_date_not_work_year_and_omits_plans():
|
|
result = answer_from_index(
|
|
"Kedy dosiahla Alena Nová aký výsledok v roku 2026?", fixture_index(), engine="rag"
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert len(result["evidence"]) == 1
|
|
assert result["evidence"][0]["date"] == "2026-02-05"
|
|
assert "DeepEval" not in result["answer"]
|
|
assert all(r["person"] == "Alena Nová" for r in result["evidence"])
|
|
|
|
|
|
def test_period_is_applied_to_same_evidence_not_another_person_or_work_year():
|
|
result = answer_from_index(
|
|
"Aké témy sa riešili od roku 2025 do roku 2025?", fixture_index(), engine="graphrag"
|
|
)
|
|
assert result["status"] == "ok" and {r["person"] for r in result["evidence"]} == {"Boris Starý"}
|
|
assert all(r["date"] == "2025-04-03" for r in result["evidence"])
|
|
|
|
|
|
def test_unknown_person_and_unknown_topic_do_not_expand_into_global_results():
|
|
result = answer_from_index("Ako pokročil Xaver Neexistujúci?", fixture_index(), engine="rag")
|
|
assert result["status"] == "no_answer" and not result["source_urls"]
|
|
result = answer_from_index(
|
|
"Kto sa venoval kvantovej teleportácii?", fixture_index(), engine="rag"
|
|
)
|
|
assert result["status"] == "no_answer" and not result["evidence"]
|
|
|
|
|
|
def test_similarity_includes_both_sides_of_its_evidence():
|
|
result = answer_from_index(
|
|
"Kto pracoval na podobnej téme ako Alena Nová?", fixture_index(), engine="graphrag"
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert {r["person"] for r in result["evidence"]} == {"Alena Nová", "Boris Starý"}
|
|
assert len(result["source_urls"]) == 2
|
|
assert "nie dôkaz spolupráce" in result["answer"]
|
|
|
|
|
|
def test_shared_interface_or_webhook_alone_does_not_prove_topic_similarity():
|
|
index = fixture_index()
|
|
for row in index["units"]:
|
|
row["concepts"] = ["webhook", "rozhranie"]
|
|
result = answer_from_index(
|
|
"Kto pracoval na podobnej téme ako Alena Nová?", index, engine="graphrag"
|
|
)
|
|
assert result["status"] == "no_answer" and not result["evidence"]
|
|
|
|
|
|
def test_shared_graph_concepts_are_relevance_checked_before_delivery():
|
|
selected = []
|
|
|
|
def choose(messages):
|
|
payload = json.loads(messages[1]["content"])
|
|
assert payload["reference_evidence"] and payload["evidence"]
|
|
selected.append(True)
|
|
return response({"evidence_ids": ["R1"]})
|
|
|
|
result = answer_from_index(
|
|
"Kto pracoval na podobnej téme ako Alena Nová?",
|
|
fixture_index(),
|
|
engine="graphrag",
|
|
complete=choose,
|
|
)
|
|
assert selected and result["status"] == "ok"
|
|
assert {u["person"] for u in result["evidence"]} == {"Alena Nová", "Boris Starý"}
|
|
|
|
|
|
def test_year_constraint_is_not_dropped_while_resolving_a_typo():
|
|
result = answer_from_index(
|
|
"Ako pokročila Alena Novvá pri diplomovej práci v roku 1999?", fixture_index(), engine="rag"
|
|
)
|
|
assert result["status"] == "no_answer" and not result["evidence"]
|
|
|
|
|
|
def test_metadata_filters_are_never_silently_dropped():
|
|
result = answer_from_index(
|
|
"Kto pracoval na podobnej téme ako Alena Nová s tagom NLP?", fixture_index(), engine="rag"
|
|
)
|
|
assert result["status"] == "ambiguous" and result["reason"] == "unsupported_content_join"
|
|
|
|
|
|
def test_topic_model_cannot_add_unprovided_evidence():
|
|
candidates = fixture_index()["units"][:1]
|
|
with pytest.raises(ValueError):
|
|
select_topic_evidence(
|
|
"Kto vytvoril chatbota?", candidates, lambda _: response({"evidence_ids": ["invented"]})
|
|
)
|
|
|
|
|
|
def test_selection_accepts_fenced_json_and_empty_selection_but_never_unbound_lists():
|
|
candidates = fixture_index()["units"][:1]
|
|
fenced = {"choices": [{"message": {"content": '```json\n{"evidence_ids":["R1"]}\n```'}}]}
|
|
assert select_topic_evidence("Kto pracoval na RAG?", candidates, lambda _: fenced) == candidates
|
|
assert (
|
|
select_topic_evidence("Kto pracoval na teleportácii?", candidates, lambda _: response([]))
|
|
== []
|
|
)
|
|
with pytest.raises(ValueError):
|
|
select_topic_evidence("Kto pracoval na RAG?", candidates, lambda _: response(["R1"]))
|
|
|
|
|
|
def test_work_context_cannot_join_another_person_or_another_work():
|
|
from tools.knowledge.sources import with_work_context
|
|
|
|
index = fixture_index()
|
|
topic = {**index["units"][0], "id": "topic", "kind": "topic", "text": "Názov: Agent pre RAG."}
|
|
unrelated = {
|
|
**topic,
|
|
"id": "unrelated",
|
|
"work_label": "Diplomový projekt 2025",
|
|
"text": "Názov: Iná práca.",
|
|
}
|
|
other = {
|
|
**topic,
|
|
"id": "other",
|
|
"person": "Boris Starý",
|
|
"source_path": "other",
|
|
"text": "Názov: Cudzia práca.",
|
|
}
|
|
rows = with_work_context([*index["units"], topic, unrelated, other])
|
|
unit = next(r for r in rows if r["id"] == index["units"][1]["id"])
|
|
assert [r["id"] for r in unit["context_quotes"]] == ["topic"]
|
|
|
|
|
|
def test_annotation_of_another_target_or_an_analogy_is_not_named_entity_annotation():
|
|
from tools.knowledge.query import activity_supported
|
|
|
|
question = "Kto sa venoval anotácii pomenovaných entít?"
|
|
assert not activity_supported(
|
|
question, {"text": "Vytvorené anotácie do databázy, cca 1700 jednotiek."}
|
|
)
|
|
assert not activity_supported(
|
|
question, {"text": "Anotácia urážlivých častí. Dáta anotujete podobne ako named entities."}
|
|
)
|
|
assert not activity_supported(
|
|
question, {"text": "Rozpoznávanie pomenovaných entít pomocou BERT."}
|
|
)
|
|
assert activity_supported(
|
|
question, {"text": "Rozbehaná slovenská anotačná schéma pre pomenované entity."}
|
|
)
|
|
assert activity_supported(
|
|
question,
|
|
{
|
|
"text": "Vytvorené anotácie do databázy.",
|
|
"context_quotes": [{"text": "Názov: Anotácia NER."}],
|
|
},
|
|
)
|
|
|
|
|
|
def test_explicit_dates_can_resolve_a_semester_without_inventing_boundaries():
|
|
result = answer_from_index(
|
|
"Aké témy sa riešili v zimnom semestri od 1.1.2026 do 1.3.2026?",
|
|
fixture_index(),
|
|
engine="rag",
|
|
mode="period",
|
|
)
|
|
assert result["status"] == "ok" and all(r["date"] == "2026-02-05" for r in result["evidence"])
|
|
|
|
|
|
@pytest.mark.parametrize("query", ["CRZP", "HDFS systémy", "NLP projekty"])
|
|
def test_real_acronyms_are_not_keyboard_gibberish(query):
|
|
assert not invalid_question(query)
|
|
|
|
|
|
def test_discovery_selects_recorded_implementation_before_a_bare_topic_link():
|
|
index = fixture_index()
|
|
rows = [
|
|
{
|
|
**index["units"][0],
|
|
"id": "link",
|
|
"phase": "unknown",
|
|
"kind": "context",
|
|
"text": "- [Chatbot](/topics/chatbot)",
|
|
},
|
|
{
|
|
**index["units"][0],
|
|
"id": "done",
|
|
"phase": "result",
|
|
"kind": "result",
|
|
"text": "- Chatbot funguje pre agendu mesta.",
|
|
},
|
|
]
|
|
|
|
def select(messages):
|
|
supplied = json.loads(messages[1]["content"])["evidence"]
|
|
assert supplied[0]["text"] == "- Chatbot funguje pre agendu mesta."
|
|
return response({"evidence_ids": ["R1"]})
|
|
|
|
result = answer_from_index(
|
|
"Kto vytvoril chatbota?", {**index, "units": rows}, engine="rag", complete=select
|
|
)
|
|
assert result["status"] == "ok" and result["evidence"][0]["id"] == "done"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"exception,reason",
|
|
[
|
|
(
|
|
__import__(
|
|
"tools.knowledge.query", fromlist=["ContentModelUnavailable"]
|
|
).ContentModelUnavailable,
|
|
"model_unavailable",
|
|
),
|
|
(
|
|
__import__(
|
|
"tools.knowledge.query", fromlist=["ContentAnswerRejected"]
|
|
).ContentAnswerRejected,
|
|
"answer_rejected",
|
|
),
|
|
(RuntimeError, "content_index_unavailable"),
|
|
],
|
|
)
|
|
def test_content_failures_keep_model_rejection_and_index_errors_distinct(
|
|
monkeypatch, exception, reason
|
|
):
|
|
import tools.knowledge.query as query
|
|
from tools.answers.answer_delivery import deliver_answer
|
|
|
|
def failed(*a, **kw):
|
|
raise exception("private diagnostic must not escape")
|
|
|
|
monkeypatch.setattr(query, "deliver_content", failed)
|
|
result = deliver_answer("Ako pokročil Ján Holp?")
|
|
assert result["reason"] == reason and not result["accepted_for_delivery"]
|
|
assert "private diagnostic" not in result["answer"]
|
|
|
|
|
|
@pytest.mark.parametrize("query", ["xdzf", "asdfgh", "bcdfrt", "x y z"])
|
|
def test_unintelligible_input_requests_clarification(query):
|
|
assert invalid_question(query)
|
|
|
|
|
|
def test_intent_does_not_intercept_existing_metadata_questions():
|
|
assert intent("Aký je názov diplomovej práce Jána Holpa a z ktorého roku je?") is None
|
|
assert intent("Aké spoločné kľúčové slová majú Ján Pták a Ján Malinovský?") is None
|
|
|
|
|
|
def test_real_markdown_acceptance_sources_are_independent_of_extraction_parser():
|
|
from tools.rag.common import ZPWIKI_ROOT
|
|
|
|
expectations = [
|
|
("2016/jan_holp", "funguje databáza Cassandra"),
|
|
("2018/ondrej_megela", "Dotrenovany t5 model na generovanie otazok."),
|
|
("2022/jan_malinovsky", "Extrahovaný text pomocou pypdf a docx."),
|
|
("2021/martin_sarissky", "Problém s kvalitou vyhľadávania."),
|
|
]
|
|
if not ZPWIKI_ROOT.is_dir():
|
|
pytest.skip("Local wiki is not installed")
|
|
for path, quote in expectations:
|
|
raw = (ZPWIKI_ROOT / "pages/students" / path / "README.md").read_text()
|
|
assert quote in raw
|
|
|
|
|
|
def test_stale_extraction_is_not_used(tmp_path):
|
|
from tools.knowledge.store import load_index
|
|
|
|
folder = tmp_path / "knowledge"
|
|
folder.mkdir()
|
|
(folder / "index.json").write_text(json.dumps({"index_version": {"version": "old"}}))
|
|
with pytest.raises(RuntimeError):
|
|
load_index(tmp_path, {"version": "new"})
|
|
|
|
|
|
def test_shared_embedding_cache_can_be_reused_without_writes(tmp_path):
|
|
import numpy as np
|
|
from tools.rag.embedding_cache import EmbeddingCache
|
|
|
|
path = tmp_path / "cache.sqlite"
|
|
writable = EmbeddingCache(path)
|
|
writable.save({"fixture": np.array([1.0, 0.0], dtype=np.float32)})
|
|
before = path.read_bytes()
|
|
readonly = EmbeddingCache(path, readonly=True)
|
|
assert readonly.load(["fixture"])["fixture"].tolist() == [1.0, 0.0]
|
|
with pytest.raises(RuntimeError):
|
|
readonly.save({"other": np.array([0.0, 1.0])})
|
|
assert path.read_bytes() == before
|
|
|
|
|
|
def test_build_is_cached_and_source_changes_cannot_activate_it(tmp_path):
|
|
from tools.knowledge.build import build
|
|
from tools.sync.index_state import durable_json
|
|
from tools.sync.sync_indexes import source_files
|
|
import hashlib
|
|
|
|
wiki = tmp_path / "wiki"
|
|
data = tmp_path / "data"
|
|
data.mkdir()
|
|
path = wiki / "pages/students/2020/alena/README.md"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text("---\ntitle: Alena Nová\npublished: true\n---\n" + doc()["content"])
|
|
version = {
|
|
"version": "fixture",
|
|
"source_sha256": hashlib.sha256(
|
|
json.dumps(source_files(wiki), sort_keys=True).encode()
|
|
).hexdigest(),
|
|
}
|
|
durable_json(data / "index_version.json", version)
|
|
calls = []
|
|
|
|
def extract(messages):
|
|
payload = json.loads(messages[1]["content"])
|
|
calls.append(1)
|
|
return response(
|
|
{
|
|
"items": [
|
|
{"evidence_id": r["evidence_id"], "decision":"irrelevant", "reason":"fixture deliberately selects no research concepts", "concepts": [], "relations": []}
|
|
for r in payload
|
|
]
|
|
}
|
|
)
|
|
|
|
first = build(wiki, data, extract, {"model": "fixture"}, graph=False, vectors=False)
|
|
assert first["profiles"] == 1 and first["llm_batches"] == 1
|
|
activated = (data / "knowledge/index.json").read_bytes()
|
|
second = build(wiki, data, extract, {"model": "fixture"}, graph=False, vectors=False)
|
|
assert second["llm_batches"] == 0 and second["cached_batches"] == 1 and len(calls) == 1
|
|
assert activated == (data / "knowledge/index.json").read_bytes()
|
|
path.write_text(path.read_text() + "\nNové podklady.\n")
|
|
with pytest.raises(RuntimeError):
|
|
build(wiki, data, extract, {"model": "fixture"}, graph=False, vectors=False)
|
|
assert activated == (data / "knowledge/index.json").read_bytes()
|
|
|
|
|
|
def test_rebuild_delegates_content_to_coordinated_publication(tmp_path, monkeypatch):
|
|
import tools.rag.rebuild_index as rebuild
|
|
import tools.sync.sync_indexes as sync
|
|
from tools.knowledge import build as knowledge
|
|
|
|
expected = {"counts": {"documents": 1}, "knowledge": {"cached_batches": 1}}
|
|
monkeypatch.setattr(sync, "synchronize_indexes", lambda *a, **kw: expected)
|
|
def forbidden(*args):
|
|
pytest.fail("Content must not be refreshed after publishing primary indexes")
|
|
monkeypatch.setattr(knowledge, "refresh_existing", forbidden)
|
|
result = rebuild.rebuild_index(wiki_root=tmp_path / "wiki", data_dir=tmp_path / "data")
|
|
assert result["knowledge"] == expected["knowledge"]
|
|
|
|
|
|
def test_explicit_work_year_is_not_misused_as_an_event_date():
|
|
index = fixture_index()
|
|
row = next(r for r in index["units"] if r["phase"] == "result" and r["person"] == "Alena Nová")
|
|
row["date"] = "2025-12-01"
|
|
result = answer_from_index(
|
|
"Kedy dosiahla Alena Nová výsledok pri diplomovej práci z roku 2026?", index, engine="rag"
|
|
)
|
|
assert result["status"] == "ok" and result["evidence"][0]["date"] == "2025-12-01"
|
|
|
|
|
|
def test_reversed_time_interval_requires_clarification():
|
|
result = answer_from_index(
|
|
"Aké témy sa riešili od roku 2026 do roku 2020?", fixture_index(), engine="rag"
|
|
)
|
|
assert result["status"] == "ambiguous" and result["reason"] == "invalid_period"
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
__import__("os").getenv("RUN_NEO4J_TESTS") != "1", reason="Requires local Neo4j"
|
|
)
|
|
def test_live_content_graph_roundtrip_is_idempotent_and_separate(monkeypatch):
|
|
import uuid
|
|
from tools.knowledge import store
|
|
from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings
|
|
|
|
scope = "knowledge-test-" + uuid.uuid4().hex
|
|
monkeypatch.setattr(store, "SCOPE", scope)
|
|
index = fixture_index()
|
|
try:
|
|
store.write_graph(index)
|
|
first = store.read_graph(index)
|
|
store.write_graph(index)
|
|
again = store.read_graph(index)
|
|
assert len(first) == len(again) == len(index["units"])
|
|
assert {r["id"]: r["text"] for r in first} == {r["id"]: r["text"] for r in index["units"]}
|
|
assert any(r.get("date") == "2026-02-05" for r in again)
|
|
with pytest.raises(RuntimeError):
|
|
store.read_graph({**index, "version": "wrong-version"})
|
|
broken = copy.deepcopy(index)
|
|
broken["units"][0]["source_path"] = {"not": "a Neo4j property"}
|
|
from neo4j.exceptions import Neo4jError
|
|
|
|
with pytest.raises(Neo4jError):
|
|
store.write_graph(broken)
|
|
assert {r["id"] for r in store.read_graph(index)} == {r["id"] for r in index["units"]}
|
|
finally:
|
|
settings = get_neo4j_settings()
|
|
with (
|
|
create_neo4j_driver(settings) as driver,
|
|
driver.session(database=settings.database) as session,
|
|
):
|
|
session.run("MATCH (n {graph_scope:$scope}) DETACH DELETE n", scope=scope).consume()
|
|
session.run("MATCH (n:KnowledgeState {scope:$scope}) DELETE n", scope=scope).consume()
|