SignalMesh / tests /test_alias_resolution.py
KyklosIO's picture
Claude Opus 5
feat(ingest+resolve): classify sources, delegate feeds, resolve by metadata
68514f4
Raw History Blame Contribute Delete
4 kB
"""Wording must reach a frequency through the source's own metadata.
Baseline captured live on 2026-08-03 against the hosted Space: "typewriter" hit
oz.typewriter.posts, but "Robert", "Robert Messenger", "oztypewriter.blogspot.com"
and "typewriter history blog" all missed — every one of them a word that appears
in the signal's metadata but not in the frequency name.
Run: python3 -m pytest tests/test_alias_resolution.py
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from core.signal_registry import SignalRegistry
def _mesh():
SignalRegistry._instance = None
reg = SignalRegistry()
for title in ("The Ongoing Popularity of Portable Typewriters",
"THE UPPER AND LOWER CASE FOR (TYPEWRITTEN) LETTERS"):
reg.broadcast(
name="oz.typewriter.posts", source_type="feed_hydrator",
data={"title": title, "link": "https://oztypewriter.blogspot.com/x.html",
"feed": "oz.Typewriter — Robert Messenger", "summary": ""},
metadata={"feed_url": "https://oztypewriter.blogspot.com/feeds/posts/default"},
)
# Decoys: resolution must stay narrow, not answer everything with everything.
reg.broadcast(name="aws_blog", source_type="feed_hydrator",
data={"title": "New in S3", "feed": "AWS Blog"},
metadata={"feed_url": "https://aws.amazon.com/blogs/aws/feed/"})
reg.broadcast(name="krebs", source_type="feed_hydrator",
data={"title": "Phishing wave", "feed": "Krebs on Security"},
metadata={"feed_url": "https://krebsonsecurity.com/feed/"})
return reg
def _hits(reg, kw):
return {s["name"] for s in reg.tune_in([kw])}
def test_frequency_name_still_resolves():
reg = _mesh()
for kw in ("oz.typewriter.posts", "typewriter", "oz"):
assert "oz.typewriter.posts" in _hits(reg, kw), kw
def test_author_resolves():
"""'What has Robert posted lately?' — the case that motivated all of this."""
reg = _mesh()
for kw in ("Robert Messenger", "What has Robert posted lately",
"Robert Messenger's Typewriting Blog"):
assert "oz.typewriter.posts" in _hits(reg, kw), kw
def test_hostname_resolves():
reg = _mesh()
assert "oz.typewriter.posts" in _hits(reg, "oztypewriter.blogspot.com")
def test_article_title_resolves():
reg = _mesh()
assert "oz.typewriter.posts" in _hits(reg, "upper and lower case for typewritten letters")
def test_download_phrasing_resolves():
"""Wording is irrelevant: 'tell me' / 'update me' / 'download' all route the same."""
reg = _mesh()
for kw in ("download Robert's latest typewriter posts",
"update me on Robert Messenger",
"tell me about oztypewriter.blogspot.com"):
assert "oz.typewriter.posts" in _hits(reg, kw), kw
def test_resolution_stays_narrow():
"""An alias index that matches everything is worse than no alias index."""
reg = _mesh()
assert _hits(reg, "Krebs on Security") == {"krebs"}
assert "oz.typewriter.posts" not in _hits(reg, "aws_blog")
assert "krebs" not in _hits(reg, "Robert Messenger")
def test_honest_miss():
reg = _mesh()
assert _hits(reg, "quarterly revenue forecast") == set()
def test_tokens_split_dots_and_camel():
t = SignalRegistry._tokens
assert t("oz.Typewriter") == {"oz", "typewriter"}
assert t("oz.typewriter.posts") == {"oz", "typewriter", "posts"}
assert t("Robert Messenger's Typewriting Blog") == \
{"robert", "messenger", "s", "typewriting", "blog"}
def test_trail_keyed_by_meaning_not_wording():
"""Two phrasings of one ask share a trail instead of each paying to learn it."""
reg = _mesh()
assert reg._trail_key("Robert's latest posts") == \
reg._trail_key("latest posts by Robert")
if __name__ == "__main__":
import pytest
raise SystemExit(pytest.main([__file__, "-q"]))