Merge pull request #67 from j-sperling/feat/query-type-source-tiering

Add query-type-aware source tiering and scoring
This commit is contained in:
Matt Van Horn
2026-03-14 07:30:06 -07:00
committed by GitHub
11 changed files with 819 additions and 113 deletions
+197
View File
@@ -0,0 +1,197 @@
"""Tests for Brave Search module, including LLM Context endpoint."""
import sys
import os
import unittest
# Ensure scripts/ is on path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
from lib.brave_search import (
_normalize_results,
_normalize_llm_context,
_days_between,
_brave_freshness,
_parse_brave_date,
EXCLUDED_DOMAINS,
)
class TestDaysBetween(unittest.TestCase):
def test_same_day(self):
self.assertEqual(_days_between("2026-03-01", "2026-03-01"), 1)
def test_one_week(self):
self.assertEqual(_days_between("2026-03-01", "2026-03-08"), 7)
def test_invalid_dates(self):
self.assertEqual(_days_between("bad", "dates"), 30)
class TestBraveFreshness(unittest.TestCase):
def test_one_day(self):
self.assertEqual(_brave_freshness(1), "pd")
def test_one_week(self):
self.assertEqual(_brave_freshness(7), "pw")
def test_one_month(self):
self.assertEqual(_brave_freshness(31), "pm")
def test_longer_returns_range(self):
result = _brave_freshness(60)
self.assertIn("to", result)
def test_none(self):
self.assertIsNone(_brave_freshness(None))
class TestParseBraveDate(unittest.TestCase):
def test_hours_ago(self):
result = _parse_brave_date("3 hours ago", None)
self.assertIsNotNone(result)
self.assertRegex(result, r"\d{4}-\d{2}-\d{2}")
def test_days_ago(self):
result = _parse_brave_date("5 days ago", None)
self.assertIsNotNone(result)
def test_weeks_ago(self):
result = _parse_brave_date("2 weeks ago", None)
self.assertIsNotNone(result)
def test_iso_date(self):
self.assertEqual(_parse_brave_date("2026-03-10T12:00:00", None), "2026-03-10")
def test_none(self):
self.assertIsNone(_parse_brave_date(None, None))
class TestNormalizeResults(unittest.TestCase):
def test_merges_news_and_web(self):
response = {
"news": {"results": [
{"url": "https://news.example.com/a", "title": "News A", "description": "News desc"},
]},
"web": {"results": [
{"url": "https://blog.example.com/b", "title": "Blog B", "description": "Blog desc"},
]},
}
items = _normalize_results(response, "2026-03-01", "2026-03-10")
self.assertEqual(len(items), 2)
self.assertEqual(items[0]["title"], "News A")
self.assertEqual(items[1]["title"], "Blog B")
def test_excludes_reddit_and_x(self):
response = {
"web": {"results": [
{"url": "https://www.reddit.com/r/test/123", "title": "Reddit", "description": "text"},
{"url": "https://x.com/user/status/1", "title": "X post", "description": "text"},
{"url": "https://example.com/ok", "title": "OK", "description": "text"},
]},
}
items = _normalize_results(response, "2026-03-01", "2026-03-10")
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["title"], "OK")
def test_default_relevance(self):
response = {"web": {"results": [
{"url": "https://a.com", "title": "A", "description": "desc"},
]}}
items = _normalize_results(response, "2026-03-01", "2026-03-10")
self.assertEqual(items[0]["relevance"], 0.6)
class TestNormalizeLlmContext(unittest.TestCase):
def _make_response(self, generic=None, sources=None):
return {
"grounding": {"generic": generic or []},
"sources": sources or {},
}
def test_basic_result(self):
resp = self._make_response(
generic=[{
"url": "https://docs.example.com/page",
"title": "Example Page",
"snippets": ["First chunk of text.", "Second chunk of text."],
}],
sources={
"https://docs.example.com/page": {
"title": "Example Page",
"hostname": "docs.example.com",
"age": ["2026-03-05", "5 days ago"],
}
},
)
items = _normalize_llm_context(resp)
self.assertEqual(len(items), 1)
item = items[0]
self.assertEqual(item["title"], "Example Page")
self.assertEqual(item["url"], "https://docs.example.com/page")
self.assertIn("First chunk", item["snippet"])
self.assertIn("Second chunk", item["snippet"])
self.assertEqual(item["date"], "2026-03-05")
self.assertEqual(item["date_confidence"], "med")
self.assertEqual(item["relevance"], 0.7)
self.assertEqual(item["source_domain"], "docs.example.com")
def test_excludes_reddit(self):
resp = self._make_response(
generic=[
{"url": "https://www.reddit.com/r/test", "title": "Reddit", "snippets": ["text"]},
{"url": "https://example.com", "title": "OK", "snippets": ["text"]},
],
)
items = _normalize_llm_context(resp)
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["title"], "OK")
def test_empty_grounding(self):
resp = self._make_response()
items = _normalize_llm_context(resp)
self.assertEqual(items, [])
def test_snippet_truncation(self):
long_snippet = "x" * 2000
resp = self._make_response(
generic=[{"url": "https://a.com", "title": "A", "snippets": [long_snippet]}],
)
items = _normalize_llm_context(resp)
self.assertLessEqual(len(items[0]["snippet"]), 1500)
def test_no_date_gives_low_confidence(self):
resp = self._make_response(
generic=[{"url": "https://a.com", "title": "A", "snippets": ["text"]}],
sources={"https://a.com": {"hostname": "a.com", "age": None}},
)
items = _normalize_llm_context(resp)
self.assertIsNone(items[0]["date"])
self.assertEqual(items[0]["date_confidence"], "low")
def test_multiple_age_entries_picks_first_valid(self):
resp = self._make_response(
generic=[{"url": "https://a.com", "title": "A", "snippets": ["text"]}],
sources={"https://a.com": {
"hostname": "a.com",
"age": ["Monday, March 10, 2026", "2026-03-10", "1 day ago"],
}},
)
items = _normalize_llm_context(resp)
self.assertEqual(items[0]["date"], "2026-03-10")
def test_ids_are_sequential(self):
resp = self._make_response(
generic=[
{"url": "https://a.com", "title": "A", "snippets": ["a"]},
{"url": "https://b.com", "title": "B", "snippets": ["b"]},
{"url": "https://c.com", "title": "C", "snippets": ["c"]},
],
)
items = _normalize_llm_context(resp)
ids = [item["id"] for item in items]
self.assertEqual(ids, ["W1", "W2", "W3"])
if __name__ == "__main__":
unittest.main()
+95 -29
View File
@@ -28,21 +28,47 @@ class TestParseVersion(unittest.TestCase):
self.assertIsNone(result)
class TestIsMainlineOpenAIModel(unittest.TestCase):
def test_gpt5_is_mainline(self):
class TestIsSearchCapableModel(unittest.TestCase):
def test_gpt5_is_capable(self):
self.assertTrue(models.is_search_capable_model("gpt-5"))
def test_gpt52_is_capable(self):
self.assertTrue(models.is_search_capable_model("gpt-5.2"))
def test_gpt5_mini_is_capable(self):
self.assertTrue(models.is_search_capable_model("gpt-5-mini"))
def test_gpt41_mini_is_capable(self):
self.assertTrue(models.is_search_capable_model("gpt-4.1-mini"))
def test_gpt4o_is_capable(self):
self.assertTrue(models.is_search_capable_model("gpt-4o"))
def test_gpt4o_mini_not_capable(self):
"""gpt-4o-mini does not support web_search with domain filtering."""
self.assertFalse(models.is_search_capable_model("gpt-4o-mini"))
def test_nano_not_capable(self):
"""nano models don't support web_search."""
self.assertFalse(models.is_search_capable_model("gpt-4.1-nano"))
self.assertFalse(models.is_search_capable_model("gpt-5-nano"))
def test_gpt4_not_capable(self):
self.assertFalse(models.is_search_capable_model("gpt-4"))
def test_codex_not_capable(self):
self.assertFalse(models.is_search_capable_model("gpt-5.1-codex"))
def test_backward_compat_alias(self):
"""is_mainline_openai_model still works as alias."""
self.assertTrue(models.is_mainline_openai_model("gpt-5"))
def test_gpt52_is_mainline(self):
self.assertTrue(models.is_mainline_openai_model("gpt-5.2"))
def test_gpt5_mini_is_not_mainline(self):
self.assertFalse(models.is_mainline_openai_model("gpt-5-mini"))
def test_gpt4_is_not_mainline(self):
self.assertFalse(models.is_mainline_openai_model("gpt-4"))
class TestSelectOpenAIModel(unittest.TestCase):
def setUp(self):
from lib import cache
cache.MODEL_CACHE_FILE.unlink(missing_ok=True)
def test_pinned_policy(self):
result = models.select_openai_model(
"fake-key",
@@ -51,20 +77,8 @@ class TestSelectOpenAIModel(unittest.TestCase):
)
self.assertEqual(result, "gpt-5.1")
def test_auto_with_mock_models(self):
mock_models = [
{"id": "gpt-5.2", "created": 1704067200},
{"id": "gpt-5.1", "created": 1701388800},
{"id": "gpt-5", "created": 1698710400},
]
result = models.select_openai_model(
"fake-key",
policy="auto",
mock_models=mock_models
)
self.assertEqual(result, "gpt-5.2")
def test_auto_filters_variants(self):
def test_prefers_mini_over_mainline(self):
"""Mini models should be preferred for cost-efficiency."""
mock_models = [
{"id": "gpt-5.2", "created": 1704067200},
{"id": "gpt-5-mini", "created": 1704067200},
@@ -75,8 +89,50 @@ class TestSelectOpenAIModel(unittest.TestCase):
policy="auto",
mock_models=mock_models
)
self.assertEqual(result, "gpt-5-mini")
def test_prefers_newer_generation_mini(self):
"""gpt-5-mini should beat gpt-4.1-mini (newer generation)."""
mock_models = [
{"id": "gpt-4.1-mini", "created": 1701388800},
{"id": "gpt-5-mini", "created": 1704067200},
{"id": "gpt-4.1", "created": 1698710400},
]
result = models.select_openai_model(
"fake-key",
policy="auto",
mock_models=mock_models
)
self.assertEqual(result, "gpt-5-mini")
def test_falls_back_to_mainline_when_no_mini(self):
"""Without mini models, mainline models are selected."""
mock_models = [
{"id": "gpt-5.2", "created": 1704067200},
{"id": "gpt-4.1", "created": 1698710400},
]
result = models.select_openai_model(
"fake-key",
policy="auto",
mock_models=mock_models
)
self.assertEqual(result, "gpt-5.2")
def test_filters_unsupported_variants(self):
"""Nano, codex, preview models should be excluded."""
mock_models = [
{"id": "gpt-5-nano", "created": 1704067200},
{"id": "gpt-5.1-codex", "created": 1704067200},
{"id": "gpt-4o-mini", "created": 1704067200},
{"id": "gpt-4.1-mini", "created": 1698710400},
]
result = models.select_openai_model(
"fake-key",
policy="auto",
mock_models=mock_models
)
self.assertEqual(result, "gpt-4.1-mini")
class TestSelectXAIModel(unittest.TestCase):
def test_latest_policy(self):
@@ -106,6 +162,10 @@ class TestSelectXAIModel(unittest.TestCase):
class TestGetModels(unittest.TestCase):
def setUp(self):
from lib import cache
cache.MODEL_CACHE_FILE.unlink(missing_ok=True)
def test_no_keys_returns_none(self):
config = {}
result = models.get_models(config)
@@ -114,9 +174,12 @@ class TestGetModels(unittest.TestCase):
def test_openai_key_only(self):
config = {"OPENAI_API_KEY": "sk-test"}
mock_models = [{"id": "gpt-5.2", "created": 1704067200}]
mock_models = [
{"id": "gpt-5.2", "created": 1704067200},
{"id": "gpt-5-mini", "created": 1704067200},
]
result = models.get_models(config, mock_openai_models=mock_models)
self.assertEqual(result["openai"], "gpt-5.2")
self.assertEqual(result["openai"], "gpt-5-mini")
self.assertIsNone(result["xai"])
def test_both_keys(self):
@@ -124,10 +187,13 @@ class TestGetModels(unittest.TestCase):
"OPENAI_API_KEY": "sk-test",
"XAI_API_KEY": "xai-test",
}
mock_openai = [{"id": "gpt-5.2", "created": 1704067200}]
mock_openai = [
{"id": "gpt-5.2", "created": 1704067200},
{"id": "gpt-5-mini", "created": 1704067200},
]
mock_xai = [{"id": "grok-4-1-fast-non-reasoning", "created": 1704067200}]
result = models.get_models(config, mock_openai, mock_xai)
self.assertEqual(result["openai"], "gpt-5.2")
self.assertEqual(result["openai"], "gpt-5-mini")
self.assertEqual(result["xai"], "grok-4-1-fast-non-reasoning")
+10 -5
View File
@@ -64,13 +64,18 @@ class TestIsModelAccessError(unittest.TestCase):
class TestModelFallbackOrder(unittest.TestCase):
"""Tests for MODEL_FALLBACK_ORDER constant."""
def test_contains_gpt4o(self):
"""Fallback list should include gpt-4o."""
def test_mini_first(self):
"""Mini models should come first (cost-efficient for structured extraction)."""
self.assertEqual(MODEL_FALLBACK_ORDER[0], "gpt-5-mini")
def test_contains_mainline_fallbacks(self):
"""Fallback list should include mainline models as last resort."""
self.assertIn("gpt-4.1", MODEL_FALLBACK_ORDER)
self.assertIn("gpt-4o", MODEL_FALLBACK_ORDER)
def test_gpt41_is_first(self):
"""gpt-4.1 should be the first fallback option."""
self.assertEqual(MODEL_FALLBACK_ORDER[0], "gpt-4.1")
def test_no_gpt4o_mini(self):
"""gpt-4o-mini should NOT be in fallback (no domain filtering support)."""
self.assertNotIn("gpt-4o-mini", MODEL_FALLBACK_ORDER)
if __name__ == "__main__":
+156
View File
@@ -0,0 +1,156 @@
"""Tests for query type detection and source tiering."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
from lib.query_type import (
detect_query_type,
is_source_enabled,
WEBSEARCH_PENALTY_BY_TYPE,
TIEBREAKER_BY_TYPE,
SOURCE_TIERS,
)
class TestDetectQueryType(unittest.TestCase):
def test_product_queries(self):
self.assertEqual(detect_query_type("cursor IDE pricing"), "product")
self.assertEqual(detect_query_type("is Claude Pro worth the cost"), "product")
self.assertEqual(detect_query_type("best free tier LLM API"), "product")
def test_concept_queries(self):
self.assertEqual(detect_query_type("what is WebTransport"), "concept")
self.assertEqual(detect_query_type("explain React Server Components"), "concept")
self.assertEqual(detect_query_type("how does MCP protocol work"), "concept")
def test_opinion_queries(self):
self.assertEqual(detect_query_type("is cursor worth it"), "opinion")
self.assertEqual(detect_query_type("thoughts on Claude Code"), "opinion")
self.assertEqual(detect_query_type("should i switch to Neovim"), "opinion")
def test_howto_queries(self):
self.assertEqual(detect_query_type("how to deploy on Vercel"), "how_to")
self.assertEqual(detect_query_type("tutorial for building MCP servers"), "how_to")
self.assertEqual(detect_query_type("step by step Kubernetes setup"), "how_to")
self.assertEqual(detect_query_type("nano banana pro prompting"), "how_to")
self.assertEqual(detect_query_type("remotion animations for Claude Code"), "how_to")
def test_comparison_queries(self):
self.assertEqual(detect_query_type("cursor vs windsurf"), "comparison")
self.assertEqual(detect_query_type("Claude compared to GPT-5"), "comparison")
self.assertEqual(detect_query_type("difference between React and Vue"), "comparison")
def test_breaking_news_queries(self):
self.assertEqual(detect_query_type("latest AI funding rounds"), "breaking_news")
self.assertEqual(detect_query_type("OpenAI just announced GPT-6"), "breaking_news")
def test_prediction_queries(self):
self.assertEqual(detect_query_type("odds of Fed rate cut"), "prediction")
self.assertEqual(detect_query_type("predict the next recession"), "prediction")
self.assertEqual(detect_query_type("election outcome 2028"), "prediction")
def test_default_is_breaking_news(self):
self.assertEqual(detect_query_type("tariffs"), "breaking_news")
self.assertEqual(detect_query_type("AI agents"), "breaking_news")
def test_comparison_beats_product(self):
"""Comparison is more specific than product."""
self.assertEqual(detect_query_type("cursor vs windsurf pricing"), "comparison")
def test_howto_beats_concept(self):
"""How-to is more specific than concept."""
self.assertEqual(detect_query_type("how to explain transformers"), "how_to")
def test_will_alone_not_prediction(self):
"""Bare 'will' should not trigger prediction classification."""
self.assertNotEqual(detect_query_type("Will React 19 support concurrent mode"), "prediction")
def test_or_for_not_comparison(self):
"""'or X for Y' should not trigger comparison classification."""
self.assertNotEqual(detect_query_type("best tools or libraries for Python"), "comparison")
class TestIsSourceEnabled(unittest.TestCase):
def test_truthsocial_always_opt_in(self):
for qt in ["product", "concept", "opinion", "breaking_news", "prediction"]:
self.assertFalse(is_source_enabled("truthsocial", qt))
self.assertTrue(is_source_enabled("truthsocial", "breaking_news", explicitly_requested=True))
def test_tier1_sources_enabled(self):
self.assertTrue(is_source_enabled("reddit", "product"))
self.assertTrue(is_source_enabled("youtube", "how_to"))
self.assertTrue(is_source_enabled("polymarket", "prediction"))
self.assertTrue(is_source_enabled("x", "breaking_news"))
def test_tier2_sources_enabled(self):
self.assertTrue(is_source_enabled("web", "product"))
self.assertTrue(is_source_enabled("bluesky", "opinion"))
self.assertTrue(is_source_enabled("x", "how_to"))
self.assertTrue(is_source_enabled("youtube", "breaking_news"))
self.assertTrue(is_source_enabled("hn", "prediction"))
def test_tier3_sources_disabled_by_default(self):
self.assertFalse(is_source_enabled("instagram", "concept"))
self.assertFalse(is_source_enabled("tiktok", "comparison"))
self.assertFalse(is_source_enabled("bluesky", "product"))
def test_explicit_request_overrides_tier(self):
self.assertTrue(is_source_enabled("instagram", "concept", explicitly_requested=True))
self.assertTrue(is_source_enabled("tiktok", "comparison", explicitly_requested=True))
class TestWebSearchPenalty(unittest.TestCase):
def test_concept_has_zero_penalty(self):
self.assertEqual(WEBSEARCH_PENALTY_BY_TYPE["concept"], 0)
def test_product_has_full_penalty(self):
self.assertEqual(WEBSEARCH_PENALTY_BY_TYPE["product"], 15)
def test_howto_has_reduced_penalty(self):
self.assertLess(WEBSEARCH_PENALTY_BY_TYPE["how_to"], WEBSEARCH_PENALTY_BY_TYPE["product"])
def test_all_query_types_have_penalty(self):
for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]:
self.assertIn(qt, WEBSEARCH_PENALTY_BY_TYPE)
class TestTiebreakerPriority(unittest.TestCase):
def test_youtube_highest_for_howto(self):
self.assertEqual(TIEBREAKER_BY_TYPE["how_to"]["youtube"], 0)
def test_x_highest_for_breaking_news(self):
self.assertEqual(TIEBREAKER_BY_TYPE["breaking_news"]["x"], 0)
def test_polymarket_highest_for_prediction(self):
self.assertEqual(TIEBREAKER_BY_TYPE["prediction"]["polymarket"], 0)
def test_hn_highest_for_concept(self):
self.assertEqual(TIEBREAKER_BY_TYPE["concept"]["hn"], 0)
def test_all_query_types_have_tiebreakers(self):
for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]:
self.assertIn(qt, TIEBREAKER_BY_TYPE)
class TestSourceTiers(unittest.TestCase):
def test_all_query_types_have_tiers(self):
for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]:
self.assertIn(qt, SOURCE_TIERS)
self.assertIn("tier1", SOURCE_TIERS[qt])
self.assertIn("tier2", SOURCE_TIERS[qt])
def test_truthsocial_not_in_any_tier(self):
for qt, tiers in SOURCE_TIERS.items():
self.assertNotIn("truthsocial", tiers["tier1"], f"truthsocial in tier1 for {qt}")
self.assertNotIn("truthsocial", tiers["tier2"], f"truthsocial in tier2 for {qt}")
if __name__ == "__main__":
unittest.main()