""" Tests for the two guards that stop a dead Instagram session going unnoticed: check-session.py preflight: is there a usable `sessionid` in the Chrome cookie DB? stories-streak.py postflight: did every stories source come back empty for days running? Both are run as subprocesses (the way gdl-cron.sh runs them); neither touches the network. python3 -I -m unittest scripts/test_session_guards.py """ import datetime as dt import os import sqlite3 import subprocess import sys import tempfile import time import unittest from pathlib import Path HERE = Path(__file__).resolve().parent CHROME_EPOCH = 11644473600 # seconds between 1601-01-01 and 1970-01-01 def run(script, *args, env=None): e = {**os.environ, **(env or {})} return subprocess.run([sys.executable, str(HERE / script), *args], capture_output=True, text=True, env=e) class CheckSession(unittest.TestCase): def profile(self, rows): """A fake Chrome profile whose cookie DB holds `rows` = [(host, name, expires_unix|None)].""" tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) d = Path(tmp.name) / "Default" d.mkdir() con = sqlite3.connect(d / "Cookies") con.execute("CREATE TABLE cookies (host_key TEXT, name TEXT, expires_utc INTEGER, has_expires INTEGER)") for host, name, exp in rows: con.execute("INSERT INTO cookies VALUES (?,?,?,?)", (host, name, 0 if exp is None else (exp + CHROME_EPOCH) * 1_000_000, 0 if exp is None else 1)) con.commit() con.close() return tmp.name def check(self, rows): return run("check-session.py", env={"GDL_COOKIE_PROFILE": self.profile(rows)}) def test_valid_sessionid_passes(self): r = self.check([(".instagram.com", "sessionid", time.time() + 200 * 86400)]) self.assertEqual(r.returncode, 0, r.stdout) self.assertIn("session ok", r.stdout) self.assertNotIn("WARNING", r.stdout) def test_missing_sessionid_fails_even_with_other_instagram_cookies(self): # What the real failure looked like: csrftoken/ds_user_id/mid linger after logout. r = self.check([(".instagram.com", "csrftoken", time.time() + 86400), (".instagram.com", "ds_user_id", time.time() + 86400), (".instagram.com", "mid", time.time() + 86400)]) self.assertEqual(r.returncode, 1) self.assertIn("NOT LOGGED IN", r.stdout) def test_expired_sessionid_fails(self): r = self.check([(".instagram.com", "sessionid", time.time() - 86400)]) self.assertEqual(r.returncode, 1) self.assertIn("expired", r.stdout) def test_near_expiry_warns_but_passes(self): r = self.check([(".instagram.com", "sessionid", time.time() + 3 * 86400)]) self.assertEqual(r.returncode, 0) self.assertIn("WARNING", r.stdout) def test_session_cookie_without_expiry_passes(self): r = self.check([(".instagram.com", "sessionid", None)]) self.assertEqual(r.returncode, 0) def test_other_sites_sessionid_does_not_count(self): r = self.check([(".example.com", "sessionid", time.time() + 86400)]) self.assertEqual(r.returncode, 1) def test_missing_profile_fails_loudly(self): r = run("check-session.py", env={"GDL_COOKIE_PROFILE": "/nonexistent/profile"}) self.assertEqual(r.returncode, 1) self.assertIn("no cookie DB", r.stdout) def test_never_prints_cookie_values(self): # The fake DB has no value column at all; the script must not need one. r = self.check([(".instagram.com", "sessionid", time.time() + 90 * 86400)]) self.assertNotIn("sessionid=", r.stdout + r.stderr) class StoriesStreak(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.dir = Path(tmp.name) self.state = self.dir / "streak" self.today = dt.date.today().isoformat() self.yesterday = (dt.date.today() - dt.timedelta(days=1)).isoformat() def out(self, to_sync, empty, skipped=0): lines = [f"sources : {to_sync} to sync, {skipped} skipped"] lines += [f"[instagram][info] No results for https://www.instagram.com/stories/user{i}/" for i in range(empty)] p = self.dir / "run.out" p.write_text("\n".join(lines) + "\n") return p def streak(self, to_sync, empty, skipped=0): return run("stories-streak.py", str(self.out(to_sync, empty, skipped)), str(self.state)) def stored(self): return self.state.read_text().split() def test_all_empty_for_three_days_fails(self): self.state.write_text(f"{self.yesterday} 2\n") r = self.streak(6, 6) self.assertEqual(r.returncode, 1) self.assertIn("SESSION LIKELY DEAD", r.stdout) self.assertEqual(self.stored(), [self.today, "3"]) def test_first_empty_day_is_quiet(self): r = self.streak(6, 6) self.assertEqual(r.returncode, 0) self.assertEqual(self.stored(), [self.today, "1"]) def test_same_day_rerun_does_not_inflate_the_streak(self): self.state.write_text(f"{self.today} 2\n") r = self.streak(6, 6) self.assertEqual(r.returncode, 0) self.assertEqual(self.stored(), [self.today, "2"]) def test_any_story_resets_the_streak(self): self.state.write_text(f"{self.yesterday} 2\n") r = self.streak(6, 5) # one profile had a story self.assertEqual(r.returncode, 0) self.assertEqual(self.stored(), [self.today, "0"]) def test_run_that_fetched_nothing_leaves_state_alone(self): self.state.write_text(f"{self.yesterday} 2\n") r = self.streak(0, 0, skipped=6) self.assertEqual(r.returncode, 0) self.assertEqual(self.stored(), [self.yesterday, "2"]) def test_threshold_is_configurable(self): r = run("stories-streak.py", str(self.out(6, 6)), str(self.state), env={"GDL_EMPTY_STREAK_FAIL": "1"}) self.assertEqual(r.returncode, 1) def test_corrupt_state_file_is_treated_as_no_history(self): self.state.write_text("garbage\n") r = self.streak(6, 6) self.assertEqual(r.returncode, 0) self.assertEqual(self.stored(), [self.today, "1"]) if __name__ == "__main__": unittest.main()