Matt Van Horn
03848365dd
* refactor(x): extract shared leading_mentions() to query.py One definition of the leading-@mention parser, reusable by every X backend; bird_x._leading_mentions delegates to it. * feat(x): xquik FROM/ABOUT handle lanes, mentions, honest probe - search_handles (from:handle, topic for ranking only) and search_mentions (@handle, own-tweet dropped) on a shared _execute_search; unique item ids across lanes; anti-bare-generic query guard. - Populate mentioned_handles (leading @mentions) so the first-party interaction signal fires for xquik items. - probe_works/probe_reason: tri-state diagnose (funded / unpaid-402 / inconclusive); 402 is fatal on the real search path too, so an unpaid key surfaces an error instead of empty results. * feat(x): single X source with an ordered backend chain x_backend_chain() returns the available X backends in priority order (xai > bird > xurl > xquik); chain[0] is the default, the rest are failover backups. A LAST30DAYS_X_BACKEND pin forces one backend. get_x_source() becomes a thin wrapper. get_x_source_status probes xquik and reports it as the active X source when the others are absent. * feat(x): X failover + fold xquik in as a backend (one X source) The 'x' source now tries its backend chain in order, falling through to the next only when one returns nothing or errors; a sole-backend hard failure (e.g. unpaid xquik 402) raises so it surfaces honestly. xquik is no longer a separate parallel source — it's a backend of 'x', so all X items share the 'x' slug and inherit native X ranking/authorship/interaction. Phase 2 handle lanes run via the primary backend (bird or xquik). * docs: position xquik as the keyless X backend for cookie-less hosts * fix(x): run handle lanes via first handle-capable backend; docstring Address review: the Phase 2 from:/mentions lanes now use the first handle-capable backend in the chain (bird or xquik) rather than requiring it to be the topic primary — so a user with xAI primary + an xquik key still gets handle enrichment via xquik instead of skipping Phase 2. The topic source stays single-with-failover; these supplemental lanes are complementary. Also restore the leading_mentions docstring noun phrasing. --------- Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com>
488 lines
19 KiB
Python
488 lines
19 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from lib.xquik import (
|
|
DEPTH_CONFIG,
|
|
_parse_tweet,
|
|
_safe_int,
|
|
expand_xquik_queries,
|
|
parse_xquik_response,
|
|
search_xquik,
|
|
)
|
|
|
|
|
|
class TestExpandXquikQueries(unittest.TestCase):
|
|
def test_quick_returns_one_query(self):
|
|
queries = expand_xquik_queries("latest trends in AI agents", "quick")
|
|
self.assertEqual(len(queries), 1)
|
|
|
|
def test_default_returns_up_to_two_queries(self):
|
|
queries = expand_xquik_queries("multi-agent systems research", "default")
|
|
self.assertLessEqual(len(queries), 2)
|
|
self.assertGreaterEqual(len(queries), 1)
|
|
|
|
def test_deep_returns_up_to_three_queries(self):
|
|
queries = expand_xquik_queries("best AI coding assistants 2026", "deep")
|
|
self.assertLessEqual(len(queries), 3)
|
|
self.assertGreaterEqual(len(queries), 1)
|
|
|
|
def test_single_word_topic(self):
|
|
queries = expand_xquik_queries("Bitcoin", "quick")
|
|
self.assertEqual(len(queries), 1)
|
|
self.assertIn("bitcoin", queries[0].lower())
|
|
|
|
|
|
class TestParseTweet(unittest.TestCase):
|
|
def test_valid_tweet(self):
|
|
tweet = {
|
|
"id": "123456",
|
|
"text": "This is a test tweet about AI agents",
|
|
"createdAt": "2026-03-15T12:00:00Z",
|
|
"likeCount": 42,
|
|
"retweetCount": 10,
|
|
"replyCount": 5,
|
|
"quoteCount": 2,
|
|
"viewCount": 5000,
|
|
"bookmarkCount": 8,
|
|
"author": {"username": "testuser", "name": "Test User"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "AI agents")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(item["id"], "XQ1")
|
|
self.assertEqual(item["url"], "https://x.com/testuser/status/123456")
|
|
self.assertEqual(item["author_handle"], "testuser")
|
|
self.assertEqual(item["date"], "2026-03-15")
|
|
self.assertEqual(item["engagement"]["likes"], 42)
|
|
self.assertEqual(item["engagement"]["reposts"], 10)
|
|
self.assertEqual(item["engagement"]["replies"], 5)
|
|
self.assertEqual(item["engagement"]["quotes"], 2)
|
|
self.assertEqual(item["engagement"]["views"], 5000)
|
|
self.assertEqual(item["engagement"]["bookmarks"], 8)
|
|
self.assertGreater(item["relevance"], 0)
|
|
|
|
def test_missing_author_returns_none(self):
|
|
tweet = {"id": "123", "text": "test"}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNone(item)
|
|
|
|
def test_at_prefix_stripped(self):
|
|
tweet = {
|
|
"id": "456",
|
|
"text": "hello",
|
|
"author": {"username": "@someone"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "hello")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(item["author_handle"], "someone")
|
|
|
|
def test_zero_engagement_preserved(self):
|
|
tweet = {
|
|
"id": "789",
|
|
"text": "zero likes tweet",
|
|
"author": {"username": "user"},
|
|
"likeCount": 0,
|
|
"retweetCount": 0,
|
|
"replyCount": 0,
|
|
"quoteCount": 0,
|
|
"viewCount": 0,
|
|
"bookmarkCount": 0,
|
|
}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(item["engagement"]["likes"], 0)
|
|
self.assertEqual(item["engagement"]["reposts"], 0)
|
|
self.assertEqual(item["engagement"]["views"], 0)
|
|
|
|
def test_none_engagement_values(self):
|
|
tweet = {
|
|
"id": "101",
|
|
"text": "minimal tweet",
|
|
"author": {"username": "user"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertIsNone(item["engagement"]["likes"])
|
|
self.assertIsNone(item["engagement"]["views"])
|
|
|
|
def test_text_truncated_at_500(self):
|
|
tweet = {
|
|
"id": "102",
|
|
"text": "x" * 600,
|
|
"author": {"username": "user"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(len(item["text"]), 500)
|
|
|
|
def test_twitter_date_format(self):
|
|
tweet = {
|
|
"id": "103",
|
|
"text": "old format",
|
|
"createdAt": "Wed Jan 15 14:30:00 +0000 2026",
|
|
"author": {"username": "user"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(item["date"], "2026-01-15")
|
|
|
|
def test_invalid_date_graceful(self):
|
|
tweet = {
|
|
"id": "104",
|
|
"text": "bad date",
|
|
"createdAt": "not-a-date",
|
|
"author": {"username": "user"},
|
|
}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertIsNone(item["date"])
|
|
|
|
def test_empty_author_dict(self):
|
|
tweet = {"id": "105", "text": "test", "author": {}}
|
|
item = _parse_tweet(tweet, 0, "test")
|
|
self.assertIsNone(item)
|
|
|
|
def test_index_offset(self):
|
|
tweet = {
|
|
"id": "106",
|
|
"text": "test",
|
|
"author": {"username": "user"},
|
|
}
|
|
item = _parse_tweet(tweet, 4, "test")
|
|
self.assertIsNotNone(item)
|
|
self.assertEqual(item["id"], "XQ5")
|
|
|
|
|
|
class TestSafeInt(unittest.TestCase):
|
|
def test_int_passthrough(self):
|
|
self.assertEqual(_safe_int(42), 42)
|
|
|
|
def test_string_int(self):
|
|
self.assertEqual(_safe_int("100"), 100)
|
|
|
|
def test_none_returns_none(self):
|
|
self.assertIsNone(_safe_int(None))
|
|
|
|
def test_invalid_string(self):
|
|
self.assertIsNone(_safe_int("abc"))
|
|
|
|
def test_zero(self):
|
|
self.assertEqual(_safe_int(0), 0)
|
|
|
|
def test_float_truncates(self):
|
|
self.assertEqual(_safe_int(3.7), 3)
|
|
|
|
|
|
class TestParseXquikResponse(unittest.TestCase):
|
|
def test_extracts_items(self):
|
|
response = {"items": [{"id": "1"}, {"id": "2"}]}
|
|
items = parse_xquik_response(response)
|
|
self.assertEqual(len(items), 2)
|
|
|
|
def test_empty_response(self):
|
|
self.assertEqual(parse_xquik_response({}), [])
|
|
|
|
def test_error_response(self):
|
|
response = {"items": [], "error": "something went wrong"}
|
|
self.assertEqual(parse_xquik_response(response), [])
|
|
|
|
|
|
class TestSearchXquik(unittest.TestCase):
|
|
def test_no_token_returns_error(self):
|
|
result = search_xquik("test", "2026-01-01", "2026-03-01", token="")
|
|
self.assertEqual(result["items"], [])
|
|
self.assertIn("XQUIK_API_KEY", result["error"])
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_successful_search(self, mock_get):
|
|
mock_get.return_value = {
|
|
"tweets": [
|
|
{
|
|
"id": "111",
|
|
"text": "AI agents are amazing",
|
|
"createdAt": "2026-02-15T10:00:00Z",
|
|
"likeCount": 50,
|
|
"retweetCount": 12,
|
|
"replyCount": 3,
|
|
"quoteCount": 1,
|
|
"viewCount": 2000,
|
|
"bookmarkCount": 5,
|
|
"author": {"username": "aidev"},
|
|
},
|
|
],
|
|
"has_next_page": False,
|
|
}
|
|
result = search_xquik("AI agents", "2026-02-01", "2026-03-01", token="test-key")
|
|
self.assertEqual(len(result["items"]), 1)
|
|
self.assertEqual(result["items"][0]["author_handle"], "aidev")
|
|
self.assertEqual(result["items"][0]["engagement"]["likes"], 50)
|
|
self.assertNotIn("error", result)
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_deduplicates_across_queries(self, mock_get):
|
|
tweet = {
|
|
"id": "222",
|
|
"text": "duplicate tweet",
|
|
"author": {"username": "user"},
|
|
}
|
|
mock_get.return_value = {"tweets": [tweet]}
|
|
result = search_xquik("test topic", "2026-01-01", "2026-03-01", depth="default", token="key")
|
|
# Even with multiple queries, same tweet ID should appear only once
|
|
ids = [item.get("id") for item in result["items"]]
|
|
# All items should have unique XQ ids (deduped by tweet ID)
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_auth_error_returns_error(self, mock_get):
|
|
from lib import http as http_mod
|
|
mock_get.side_effect = http_mod.HTTPError("Unauthorized", status_code=401)
|
|
result = search_xquik("test", "2026-01-01", "2026-03-01", token="bad-key")
|
|
self.assertEqual(result["items"], [])
|
|
self.assertIn("auth failed", result.get("error", ""))
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_unpaid_402_surfaces_error_on_real_path(self, mock_get):
|
|
# An unpaid key (402) must surface an error on the normal search path,
|
|
# not settle silently empty — diagnose is opt-in.
|
|
from lib import http as http_mod
|
|
mock_get.side_effect = http_mod.HTTPError("Payment Required", status_code=402)
|
|
result = search_xquik("test", "2026-01-01", "2026-03-01", token="unpaid-key")
|
|
self.assertEqual(result["items"], [])
|
|
self.assertIn("unpaid", result.get("error", "").lower())
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_empty_tweets_list(self, mock_get):
|
|
mock_get.return_value = {"tweets": []}
|
|
result = search_xquik("obscure topic", "2026-01-01", "2026-03-01", token="key")
|
|
self.assertEqual(result["items"], [])
|
|
self.assertNotIn("error", result)
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_non_list_tweets_skipped(self, mock_get):
|
|
mock_get.return_value = {"tweets": "not a list"}
|
|
result = search_xquik("test", "2026-01-01", "2026-03-01", token="key")
|
|
self.assertEqual(result["items"], [])
|
|
|
|
|
|
class TestDepthConfig(unittest.TestCase):
|
|
def test_all_depths_have_limit_and_queries(self):
|
|
for depth_name, cfg in DEPTH_CONFIG.items():
|
|
self.assertIn("limit", cfg, f"{depth_name} missing 'limit'")
|
|
self.assertIn("queries", cfg, f"{depth_name} missing 'queries'")
|
|
|
|
def test_deep_has_highest_limit(self):
|
|
self.assertGreater(DEPTH_CONFIG["deep"]["limit"], DEPTH_CONFIG["default"]["limit"])
|
|
self.assertGreater(DEPTH_CONFIG["default"]["limit"], DEPTH_CONFIG["quick"]["limit"])
|
|
|
|
def test_deep_has_most_queries(self):
|
|
self.assertGreater(DEPTH_CONFIG["deep"]["queries"], DEPTH_CONFIG["quick"]["queries"])
|
|
|
|
class TestIsOwn(unittest.TestCase):
|
|
def test_own_tweet_detected(self):
|
|
from lib.xquik import _is_own
|
|
self.assertTrue(_is_own("https://x.com/elonmusk/status/123", "elonmusk"))
|
|
self.assertTrue(_is_own("https://twitter.com/elonmusk/status/123", "@elonmusk"))
|
|
|
|
def test_other_author_not_own(self):
|
|
from lib.xquik import _is_own
|
|
self.assertFalse(_is_own("https://x.com/someoneelse/status/123", "elonmusk"))
|
|
|
|
def test_empty_handle_or_url(self):
|
|
from lib.xquik import _is_own
|
|
self.assertFalse(_is_own("", "elonmusk"))
|
|
self.assertFalse(_is_own("https://x.com/a/status/1", ""))
|
|
|
|
|
|
class TestFromLane(unittest.TestCase):
|
|
def _resp(self, username, tid="1"):
|
|
return {"tweets": [{
|
|
"id": tid, "text": "anything", "createdAt": "2026-06-15T12:00:00Z",
|
|
"likeCount": 10, "author": {"username": username},
|
|
}]}
|
|
|
|
def test_from_query_shape_and_no_topic_anded(self):
|
|
from lib import xquik
|
|
with patch("lib.xquik.http.get", return_value=self._resp("elonmusk")) as m:
|
|
items = xquik.search_handles(["@elonmusk"], "Grok 4", "2026-05-19", "2026-06-18",
|
|
count_per=8, token="k")
|
|
url = m.call_args[0][0]
|
|
self.assertIn("from%3Aelonmusk", url) # from:elonmusk url-encoded
|
|
self.assertIn("since%3A2026-05-19", url)
|
|
self.assertNotIn("Grok", url) # topic must NOT be AND'd into query
|
|
self.assertEqual(1, len(items))
|
|
self.assertEqual("XF1", items[0]["id"]) # FROM-lane id prefix
|
|
|
|
def test_no_token_or_no_handles_returns_empty(self):
|
|
from lib import xquik
|
|
self.assertEqual([], xquik.search_handles(["@x"], "t", "a", "b", token=""))
|
|
self.assertEqual([], xquik.search_handles([], "t", "a", "b", token="k"))
|
|
|
|
def test_item_ids_unique_across_handles(self):
|
|
# Different tweets across two handles must not collide on item id.
|
|
from lib import xquik
|
|
responses = [
|
|
{"tweets": [{"id": "1", "text": "a", "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "h1"}}]},
|
|
{"tweets": [{"id": "2", "text": "b", "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "h2"}}]},
|
|
]
|
|
with patch("lib.xquik.http.get", side_effect=responses):
|
|
items = xquik.search_handles(["h1", "h2"], "topic", "2026-05-19", "2026-06-18", token="k")
|
|
ids = [it["id"] for it in items]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
|
|
class TestAboutLane(unittest.TestCase):
|
|
def test_mentions_drop_own_tweets(self):
|
|
from lib import xquik
|
|
resp = {"tweets": [
|
|
{"id": "1", "text": "@elonmusk nice", "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "fan"}},
|
|
{"id": "2", "text": "my own post", "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "elonmusk"}},
|
|
]}
|
|
with patch("lib.xquik.http.get", return_value=resp) as m:
|
|
items = xquik.search_mentions(["elonmusk"], "2026-05-19", "2026-06-18",
|
|
topic="Grok 4", count_per=5, token="k")
|
|
url = m.call_args[0][0]
|
|
self.assertIn("%40elonmusk", url) # @elonmusk url-encoded
|
|
authors = {it["author_handle"] for it in items}
|
|
self.assertIn("fan", authors)
|
|
self.assertNotIn("elonmusk", authors) # own tweet dropped
|
|
|
|
|
|
class TestExpandGuard(unittest.TestCase):
|
|
@patch("lib.xquik._extract_core_subject", return_value="news")
|
|
def test_bare_generic_core_falls_back_to_topic(self, _m):
|
|
# #607: a single bare generic core must not be the query for a
|
|
# multi-word topic — fall back to the full topic.
|
|
qs = expand_xquik_queries("Grok 4 news", "quick")
|
|
self.assertEqual(["Grok 4 news"], qs)
|
|
|
|
@patch("lib.xquik._extract_core_subject", return_value="Grok 4")
|
|
def test_multiword_core_kept(self, _m):
|
|
qs = expand_xquik_queries("Grok 4 latest", "quick")
|
|
self.assertEqual(["Grok 4"], qs)
|
|
|
|
|
|
class TestProbeWorks(unittest.TestCase):
|
|
"""U5: honest diagnose probe — tri-state, surfaces the unpaid (402) case."""
|
|
|
|
def setUp(self):
|
|
import lib.xquik as xq
|
|
xq._probe_cache = ("unset", "")
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_funded_key_works(self, mock_get):
|
|
from lib import xquik
|
|
mock_get.return_value = {"tweets": [{"id": "1"}]}
|
|
self.assertIs(True, xquik.probe_works("k"))
|
|
self.assertEqual("ok", xquik.probe_reason())
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_unpaid_402_is_false_with_reason(self, mock_get):
|
|
from lib import xquik, http as http_mod
|
|
mock_get.side_effect = http_mod.HTTPError("Payment Required", status_code=402)
|
|
self.assertIs(False, xquik.probe_works("k"))
|
|
self.assertIn("unpaid", xquik.probe_reason())
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_auth_401_is_false(self, mock_get):
|
|
from lib import xquik, http as http_mod
|
|
mock_get.side_effect = http_mod.HTTPError("Unauthorized", status_code=401)
|
|
self.assertIs(False, xquik.probe_works("k"))
|
|
self.assertIn("auth failed", xquik.probe_reason())
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_timeout_is_inconclusive_fail_open(self, mock_get):
|
|
from lib import xquik
|
|
mock_get.side_effect = TimeoutError("timed out")
|
|
self.assertIsNone(xquik.probe_works("k"))
|
|
|
|
def test_no_token_is_false(self):
|
|
from lib import xquik
|
|
self.assertIs(False, xquik.probe_works(""))
|
|
self.assertIn("no XQUIK_API_KEY", xquik.probe_reason())
|
|
|
|
@patch("lib.xquik.http.get")
|
|
def test_result_is_cached(self, mock_get):
|
|
from lib import xquik
|
|
mock_get.return_value = {"tweets": [{"id": "1"}]}
|
|
xquik.probe_works("k")
|
|
xquik.probe_works("k")
|
|
self.assertEqual(1, mock_get.call_count)
|
|
|
|
|
|
class TestDiagnoseSurfacesXquik(unittest.TestCase):
|
|
"""U5: get_x_source_status reports xquik as the active X source when bird/
|
|
xAI/xurl are absent, and surfaces the probe reason."""
|
|
|
|
def setUp(self):
|
|
import lib.xquik as xq
|
|
xq._probe_cache = ("unset", "")
|
|
|
|
@patch("lib.xquik.http.get")
|
|
@patch("lib.xurl_x.is_available", return_value=False)
|
|
@patch("lib.bird_x.get_bird_status")
|
|
def test_xquik_is_active_x_source_when_only_key(self, mock_bird, _xurl, mock_get):
|
|
from lib import env
|
|
mock_bird.return_value = {"installed": False, "authenticated": False,
|
|
"username": "", "can_install": False}
|
|
mock_get.return_value = {"tweets": [{"id": "1"}]}
|
|
status = env.get_x_source_status({"XQUIK_API_KEY": "k"}, probe=True)
|
|
self.assertEqual("xquik", status["source"])
|
|
self.assertTrue(status["xquik_available"])
|
|
self.assertIs(True, status["xquik_working"])
|
|
|
|
@patch("lib.xurl_x.is_available", return_value=False)
|
|
@patch("lib.bird_x.get_bird_status")
|
|
def test_unpaid_xquik_not_active_source(self, mock_bird, _xurl):
|
|
from lib import env, http as http_mod
|
|
import lib.xquik as xq
|
|
mock_bird.return_value = {"installed": False, "authenticated": False,
|
|
"username": "", "can_install": False}
|
|
with patch("lib.xquik.http.get", side_effect=http_mod.HTTPError("pay", status_code=402)):
|
|
status = env.get_x_source_status({"XQUIK_API_KEY": "k"}, probe=True)
|
|
self.assertIsNone(status["source"]) # unpaid key is not a usable X source
|
|
self.assertIs(False, status["xquik_working"])
|
|
self.assertIn("unpaid", status["xquik_status"])
|
|
|
|
|
|
class TestMentionedHandles(unittest.TestCase):
|
|
"""U3: xquik items carry leading-run @mentions so the first-party
|
|
interaction signal fires (shared parser with bird)."""
|
|
|
|
def _tweet(self, text):
|
|
return {
|
|
"id": "1", "text": text, "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "subject"},
|
|
}
|
|
|
|
def test_leading_mentions_captured(self):
|
|
item = _parse_tweet(self._tweet("@jack @pmarca thoughts on this"), 0, "topic")
|
|
self.assertEqual(["jack", "pmarca"], item["mentioned_handles"])
|
|
|
|
def test_midbody_mention_ignored(self):
|
|
item = _parse_tweet(self._tweet("I think @jack is right"), 0, "topic")
|
|
self.assertEqual([], item["mentioned_handles"])
|
|
|
|
def test_no_mentions(self):
|
|
item = _parse_tweet(self._tweet("Grok 4 just shipped"), 0, "topic")
|
|
self.assertEqual([], item["mentioned_handles"])
|
|
|
|
|
|
class TestNormalizePropagatesMentions(unittest.TestCase):
|
|
"""U3: _normalize_x carries xquik mentioned_handles into metadata so rerank
|
|
can read them."""
|
|
|
|
def test_mentioned_handles_reach_metadata(self):
|
|
from lib import normalize
|
|
item = _parse_tweet(
|
|
{"id": "1", "text": "@jack hi", "createdAt": "2026-06-15T12:00:00Z",
|
|
"author": {"username": "subject"}}, 0, "topic")
|
|
normalized = normalize.normalize_source_items("xquik", [item], "2026-05-19", "2026-06-18")
|
|
self.assertEqual(["jack"], normalized[0].metadata.get("mentioned_handles"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|