154 lines
5.8 KiB
Python
154 lines
5.8 KiB
Python
import pytest
|
|
from pydantic import ValidationError
|
|
from deep_research.schemas import Scope, Angle, Search, SearchResult, Extract, Claim, Verdict, Report
|
|
from deep_research.core import norm_url, Deduper, rank_claims, tally, build_synth_blocks, build_stats
|
|
from deep_research.config import Config
|
|
|
|
|
|
def test_scope_requires_min_3_angles():
|
|
with pytest.raises(ValidationError):
|
|
Scope(question="q", summary="s", angles=[Angle(label="a", query="x")])
|
|
|
|
|
|
def test_scope_accepts_5_angles():
|
|
angles = [Angle(label=f"a{i}", query="x") for i in range(5)]
|
|
s = Scope(question="q", summary="s", angles=angles)
|
|
assert len(s.angles) == 5
|
|
|
|
|
|
def test_search_relevance_enum_enforced():
|
|
with pytest.raises(ValidationError):
|
|
SearchResult(url="http://a", title="t", relevance="bogus")
|
|
|
|
|
|
def test_json_schema_export_works():
|
|
schema = Scope.model_json_schema()
|
|
assert "angles" in schema["properties"]
|
|
assert schema["properties"]["angles"]["minItems"] == 3
|
|
assert schema["properties"]["angles"]["maxItems"] == 6
|
|
|
|
|
|
def test_norm_url_strips_www_and_trailing_slash():
|
|
assert norm_url("http://www.Example.com/Path/") == "example.com/path"
|
|
|
|
|
|
def test_norm_url_bare_host():
|
|
assert norm_url("https://example.com") == "example.com"
|
|
|
|
|
|
def test_norm_url_fallback_on_garbage():
|
|
assert norm_url("not a url") == "not a url"
|
|
|
|
|
|
def test_deduper_filters_exact_dupes():
|
|
d = Deduper(Config(MAX_FETCH=15))
|
|
r = [{"url": "http://a.com/x", "title": "t", "relevance": "high"}]
|
|
assert len(d.filter_novel("angle1", r)) == 1
|
|
# same normalized url from another angle -> dup
|
|
r2 = [{"url": "http://www.a.com/x/", "title": "t2", "relevance": "high"}]
|
|
assert d.filter_novel("angle2", r2) == []
|
|
assert len(d.dupes) == 1
|
|
|
|
|
|
def test_deduper_budget_drops_medium_low_when_slots_exhausted():
|
|
d = Deduper(Config(MAX_FETCH=1))
|
|
first = [{"url": "http://a.com/1", "title": "t", "relevance": "high"}]
|
|
d.filter_novel("a1", first) # consumes the only slot
|
|
more = [
|
|
{"url": "http://b.com/2", "title": "t", "relevance": "medium"},
|
|
{"url": "http://c.com/3", "title": "t", "relevance": "low"},
|
|
]
|
|
assert d.filter_novel("a2", more) == []
|
|
assert len(d.budget_dropped) == 2
|
|
|
|
|
|
def test_deduper_high_passes_even_when_slots_exhausted():
|
|
# 원본: high(rank 0)는 budget 조건(rank>=1)에 안 걸려 slot<=0 이어도 통과
|
|
d = Deduper(Config(MAX_FETCH=1))
|
|
d.filter_novel("a1", [{"url": "http://a.com/1", "title": "t", "relevance": "high"}])
|
|
high = [{"url": "http://d.com/4", "title": "t", "relevance": "high"}]
|
|
assert len(d.filter_novel("a2", high)) == 1
|
|
assert d.fetch_slots == -1
|
|
|
|
|
|
def _claim(imp, qual, name="c"):
|
|
return {"claim": name, "importance": imp, "sourceQuality": qual}
|
|
|
|
|
|
def test_rank_claims_orders_by_importance_then_quality():
|
|
claims = [
|
|
_claim("tangential", "primary", "t-prim"),
|
|
_claim("central", "blog", "c-blog"),
|
|
_claim("central", "primary", "c-prim"),
|
|
]
|
|
ranked = rank_claims(claims, max_verify=25)
|
|
assert [c["claim"] for c in ranked] == ["c-prim", "c-blog", "t-prim"]
|
|
|
|
|
|
def test_rank_claims_truncates_to_max():
|
|
claims = [_claim("central", "primary", f"c{i}") for i in range(30)]
|
|
assert len(rank_claims(claims, max_verify=25)) == 25
|
|
|
|
|
|
def test_tally_survives_2_valid_0_refute():
|
|
v = [{"refuted": False}, {"refuted": False}, {"refuted": False}]
|
|
t = tally(v, votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is True and t["refutedVotes"] == 0
|
|
|
|
|
|
def test_tally_killed_2_refute():
|
|
v = [{"refuted": True}, {"refuted": True}, {"refuted": False}]
|
|
t = tally(v, votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is False and t["refutedVotes"] == 2
|
|
|
|
|
|
def test_tally_all_abstain_does_not_survive():
|
|
# ⚠️ 거짓 생존 차단: all-None -> refuted=0 이지만 valid<2 라 미생존
|
|
t = tally([None, None, None], votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is False and t["abstained"] == 3
|
|
|
|
|
|
def test_tally_one_valid_two_abstain_does_not_survive():
|
|
t = tally([{"refuted": False}, None, None], votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is False # valid(1) < 2
|
|
|
|
|
|
def test_tally_boundary_one_refute_two_valid_survives():
|
|
t = tally([{"refuted": True}, {"refuted": False}], votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is True # 1 refute < 2 required, quorum met
|
|
|
|
|
|
def test_tally_boundary_two_refute_two_valid_killed():
|
|
t = tally([{"refuted": True}, {"refuted": True}], votes_per_claim=3, refutations_required=2)
|
|
assert t["survives"] is False # 2 refute == 2 required, strict < fails
|
|
|
|
|
|
def test_build_synth_blocks_includes_vote_and_source():
|
|
confirmed = [{
|
|
"claim": "X causes Y", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
|
|
"valid": [{"refuted": False, "confidence": "high", "evidence": "e"}],
|
|
"refutedVotes": 0,
|
|
}]
|
|
killed = [{
|
|
"claim": "Z", "sourceUrl": "http://b", "valid": [{"refuted": True}], "refutedVotes": 1,
|
|
}]
|
|
block, killed_block = build_synth_blocks(confirmed, killed)
|
|
assert "X causes Y" in block and "1-0" in block
|
|
assert "Refuted claims" in killed_block and "Z" in killed_block
|
|
|
|
|
|
def test_build_synth_blocks_no_killed():
|
|
block, killed_block = build_synth_blocks([{
|
|
"claim": "X", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
|
|
"valid": [{"refuted": False, "confidence": "high", "evidence": "e"}], "refutedVotes": 0,
|
|
}], [])
|
|
assert killed_block == ""
|
|
|
|
|
|
def test_build_stats_agent_calls_formula():
|
|
s = build_stats(angles=5, sources=12, claims=20, voted=18, confirmed=10, killed=8,
|
|
after_synth=6, dupes=3, budget_dropped=2, votes_per_claim=3)
|
|
# 1 + angles + sources + voted*votes_per_claim + 1
|
|
assert s["agentCalls"] == 1 + 5 + 12 + 18 * 3 + 1
|
|
assert s["confirmed"] == 10 and s["afterSynthesis"] == 6
|