Ilia Alshanetsky
bbf892aecc
bird_x.py and youtube_yt.py had four near-identical copies of the same subprocess cleanup dance (Popen + os.setsid + communicate(timeout) + SIGTERM via killpg + proc.kill() fallback + wait(5)). Extract to lib.subproc.run_with_timeout(), which: - runs the child in its own process group via os.setsid where available - raises SubprocTimeout on timeout - on timeout: SIGTERM the group, fall back to proc.kill(), wait up to 5s - accepts an on_pid callback so bird_x can still register child PIDs with last30days.register_child_pid for whole-process cleanup - captures stdout/stderr as strings in a SubprocResult dataclass Migrated call sites: _run_bird_search, search_handles inner worker, search_youtube, fetch_transcript. With the helper in place, the signal and subprocess imports became dead in both files (plus os in youtube_yt) and went with them. Tests: 9 new subproc tests cover success, non-zero exit, stderr capture, timeout-raises, timeout-kills-group, missing-command, env passthrough, PID callback, and callback-exception suppression. test_env_v3 and test_youtube_yt patch subproc.run_with_timeout instead of the removed bird_x.subprocess and yt-dlp subprocess.
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "skills" / "last30days" / "scripts"))
|
|
|
|
from lib import bird_x, env
|
|
|
|
|
|
class EnvV3Tests(unittest.TestCase):
|
|
def setUp(self):
|
|
self._saved_credentials = dict(bird_x._credentials)
|
|
|
|
def tearDown(self):
|
|
bird_x._credentials.clear()
|
|
bird_x._credentials.update(self._saved_credentials)
|
|
|
|
def test_x_source_prefers_xai_without_bird_probe(self):
|
|
with mock.patch("lib.bird_x.is_bird_authenticated", side_effect=AssertionError("should not probe bird auth")):
|
|
source = env.get_x_source({"XAI_API_KEY": "test"})
|
|
self.assertEqual("xai", source)
|
|
|
|
def test_x_source_uses_bird_with_explicit_cookies(self):
|
|
with mock.patch("lib.bird_x.is_bird_installed", return_value=True):
|
|
source = env.get_x_source({"AUTH_TOKEN": "a", "CT0": "b"})
|
|
self.assertEqual("bird", source)
|
|
self.assertEqual("a", bird_x._credentials["AUTH_TOKEN"])
|
|
self.assertEqual("b", bird_x._credentials["CT0"])
|
|
|
|
def test_bird_auth_never_checks_browser_cookies(self):
|
|
# The guarantee: is_bird_authenticated() must not spawn any child
|
|
# process to probe for cookies. All subprocess paths in bird_x go
|
|
# through subproc.run_with_timeout, so patching that covers it.
|
|
with mock.patch("lib.bird_x.is_bird_installed", return_value=True), mock.patch(
|
|
"lib.bird_x.subproc.run_with_timeout",
|
|
side_effect=AssertionError("browser-cookie whoami should not run"),
|
|
):
|
|
bird_x._credentials.clear()
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
self.assertIsNone(bird_x.is_bird_authenticated())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|