Yong-yuan-X
e74b0e1e93
Add a pytest-discovered tests/conftest.py for the last30days scripts path and remove duplicate per-file sys.path.insert boilerplate from tests. Normalize affected imports to rely on the shared scripts path and remove the now-unneeded E402 suppressions.
305 lines
11 KiB
Python
305 lines
11 KiB
Python
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from lib.bird_x import parse_bird_response
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
VENDORED_BIRD = REPO_ROOT / "skills" / "last30days" / "scripts" / "lib" / "vendor" / "bird-search" / "bird-search.mjs"
|
|
|
|
|
|
class TestBirdXEngagementZero(unittest.TestCase):
|
|
def test_zero_likes_preserved(self):
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
"likeCount": 0,
|
|
"retweetCount": 5,
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertEqual(0, items[0]["engagement"]["likes"])
|
|
self.assertEqual(5, items[0]["engagement"]["reposts"])
|
|
|
|
@unittest.skipUnless(shutil.which("node"), "node is required for vendored Bird tests")
|
|
class TestVendoredBirdRuntime(unittest.TestCase):
|
|
def test_check_uses_env_credentials_without_browser_cookie_dependency(self):
|
|
env = os.environ.copy()
|
|
env["AUTH_TOKEN"] = "dummy-auth"
|
|
env["CT0"] = "dummy-ct0"
|
|
|
|
result = subprocess.run(
|
|
["node", str(VENDORED_BIRD), "--check"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertTrue(payload["authenticated"])
|
|
self.assertEqual("env AUTH_TOKEN", payload["source"])
|
|
|
|
def test_check_with_browser_lookup_disabled_returns_json_warnings(self):
|
|
env = os.environ.copy()
|
|
env.pop("AUTH_TOKEN", None)
|
|
env.pop("CT0", None)
|
|
env["BIRD_DISABLE_BROWSER_COOKIES"] = "1"
|
|
|
|
result = subprocess.run(
|
|
["node", str(VENDORED_BIRD), "--check"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(1, result.returncode, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertFalse(payload["authenticated"])
|
|
self.assertTrue(payload["warnings"])
|
|
self.assertIn("Missing auth_token", " ".join(payload["warnings"]))
|
|
|
|
def test_browser_cookie_helpers_lazy_load_sweet_cookie(self):
|
|
sweet_cookie_dir = (
|
|
REPO_ROOT
|
|
/ "skills"
|
|
/ "last30days"
|
|
/ "scripts"
|
|
/ "lib"
|
|
/ "vendor"
|
|
/ "bird-search"
|
|
/ "lib"
|
|
/ "node_modules"
|
|
/ "@steipete"
|
|
/ "sweet-cookie"
|
|
)
|
|
if sweet_cookie_dir.exists():
|
|
self.skipTest("vendored sweet-cookie test stub already exists")
|
|
|
|
sweet_cookie_dir.mkdir(parents=True)
|
|
(sweet_cookie_dir / "package.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "@steipete/sweet-cookie",
|
|
"type": "module",
|
|
"exports": "./index.js",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(sweet_cookie_dir / "index.js").write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
export async function getCookies(options) {
|
|
const browser = options.browsers?.[0] ?? "unknown";
|
|
return {
|
|
cookies: [
|
|
{ name: "auth_token", value: `${browser}-auth`, domain: "x.com" },
|
|
{ name: "ct0", value: `${browser}-ct0`, domain: "x.com" },
|
|
],
|
|
warnings: [],
|
|
};
|
|
}
|
|
"""
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
"node",
|
|
"--input-type=module",
|
|
"-e",
|
|
textwrap.dedent(
|
|
"""
|
|
import {
|
|
extractCookiesFromSafari,
|
|
extractCookiesFromChrome,
|
|
extractCookiesFromFirefox,
|
|
} from "./skills/last30days/scripts/lib/vendor/bird-search/lib/cookies.js";
|
|
|
|
const payload = await Promise.all([
|
|
extractCookiesFromSafari(),
|
|
extractCookiesFromChrome("Profile 1"),
|
|
extractCookiesFromFirefox("default-release"),
|
|
]);
|
|
process.stdout.write(JSON.stringify(payload));
|
|
"""
|
|
),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual("Safari", payload[0]["cookies"]["source"])
|
|
self.assertEqual('Chrome profile "Profile 1"', payload[1]["cookies"]["source"])
|
|
self.assertEqual(
|
|
'Firefox profile "default-release"', payload[2]["cookies"]["source"]
|
|
)
|
|
self.assertEqual("safari-auth", payload[0]["cookies"]["authToken"])
|
|
self.assertEqual("chrome-auth", payload[1]["cookies"]["authToken"])
|
|
self.assertEqual("firefox-auth", payload[2]["cookies"]["authToken"])
|
|
finally:
|
|
shutil.rmtree(sweet_cookie_dir, ignore_errors=True)
|
|
for path in [sweet_cookie_dir.parent, sweet_cookie_dir.parent.parent]:
|
|
try:
|
|
path.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
def test_none_likes_when_missing(self):
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test tweet with no engagement fields",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
# no likeCount, like_count, or favorite_count
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertIsNone(items[0]["engagement"])
|
|
|
|
def test_fallback_to_second_key(self):
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
"like_count": 7,
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertEqual(7, items[0]["engagement"]["likes"])
|
|
|
|
def test_zero_does_not_fall_through(self):
|
|
"""likeCount=0 should not fall through to like_count=10."""
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
"likeCount": 0,
|
|
"like_count": 10,
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertEqual(0, items[0]["engagement"]["likes"])
|
|
|
|
def test_engagement_none_when_all_fields_missing(self):
|
|
"""All-None engagement dict should become None, not propagate."""
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertIsNone(items[0]["engagement"])
|
|
|
|
def test_engagement_preserved_when_any_field_present(self):
|
|
"""Engagement dict kept when at least one metric exists."""
|
|
tweets = [
|
|
{
|
|
"id": "1",
|
|
"text": "test",
|
|
"permanent_url": "https://x.com/u/status/1",
|
|
"likeCount": 5,
|
|
}
|
|
]
|
|
items = parse_bird_response(tweets, "test query")
|
|
self.assertIsNotNone(items[0]["engagement"])
|
|
self.assertEqual(5, items[0]["engagement"]["likes"])
|
|
|
|
|
|
class TestRunBirdSearchJsonDecodeRetry(unittest.TestCase):
|
|
"""When bird-search returns non-JSON stdout, retry the subprocess.
|
|
|
|
Twitter's edge sometimes serves an HTML anti-bot interstitial in place of
|
|
JSON. Before this fix, that response made json.loads raise JSONDecodeError
|
|
and the function returned {"items": []} with no diagnostic — silent-empty
|
|
against an orchestrator that can't distinguish "Twitter blocked us" from
|
|
"no tweets matched the query."
|
|
"""
|
|
|
|
def _make_result(self, stdout: str, stderr: str = "", returncode: int = 0):
|
|
from lib.subproc import SubprocResult
|
|
return SubprocResult(returncode=returncode, stdout=stdout, stderr=stderr)
|
|
|
|
def test_retries_subprocess_on_html_interstitial_then_succeeds(self):
|
|
"""First subprocess attempt returns HTML; second returns JSON → success."""
|
|
from unittest import mock
|
|
from lib import bird_x
|
|
|
|
html_interstitial = "<!DOCTYPE html><html><body>Rate limited</body></html>"
|
|
json_success = '[{"id": "1", "text": "tweet"}]'
|
|
|
|
results = [
|
|
(self._make_result(stdout=html_interstitial), None),
|
|
(self._make_result(stdout=json_success), None),
|
|
]
|
|
|
|
with mock.patch.object(bird_x, "_invoke_bird_subprocess", side_effect=results), \
|
|
mock.patch.object(bird_x.time, "sleep") as mock_sleep:
|
|
response = bird_x._run_bird_search("test", count=10, timeout=30)
|
|
|
|
self.assertNotIn("error", response)
|
|
self.assertEqual(response["items"], [{"id": "1", "text": "tweet"}])
|
|
# Should have slept between the failed first attempt and the retry.
|
|
mock_sleep.assert_called_once_with(bird_x.JSON_DECODE_RETRY_DELAY)
|
|
|
|
def test_returns_error_after_all_retries_exhausted(self):
|
|
"""All attempts return HTML → error dict with diagnostic + items=[]."""
|
|
from unittest import mock
|
|
from lib import bird_x
|
|
|
|
html_interstitial = "<!DOCTYPE html><html>blocked</html>"
|
|
results = [
|
|
(self._make_result(stdout=html_interstitial), None),
|
|
(self._make_result(stdout=html_interstitial), None),
|
|
]
|
|
|
|
with mock.patch.object(bird_x, "_invoke_bird_subprocess", side_effect=results), \
|
|
mock.patch.object(bird_x.time, "sleep"):
|
|
response = bird_x._run_bird_search("test", count=10, timeout=30)
|
|
|
|
self.assertIn("error", response)
|
|
self.assertIn("Invalid JSON response", response["error"])
|
|
# Diagnostic message names the anti-bot interstitial so it's
|
|
# distinguishable from a genuine no-results case in logs.
|
|
self.assertIn("anti-bot interstitial", response["error"].lower())
|
|
self.assertEqual(response["items"], [])
|
|
|
|
def test_terminal_subprocess_error_is_not_retried(self):
|
|
"""Subprocess timeout / spawn failure → terminal error, no retry."""
|
|
from unittest import mock
|
|
from lib import bird_x
|
|
|
|
timeout_error = {"error": "Search timed out after 30s", "items": []}
|
|
results = [(None, timeout_error)]
|
|
|
|
with mock.patch.object(bird_x, "_invoke_bird_subprocess", side_effect=results), \
|
|
mock.patch.object(bird_x.time, "sleep") as mock_sleep:
|
|
response = bird_x._run_bird_search("test", count=10, timeout=30)
|
|
|
|
self.assertEqual(response, timeout_error)
|
|
mock_sleep.assert_not_called()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|