feat: guard against a dead Instagram session, show progress, fix missing pings
- check-session.py: preflight in gdl-cron.sh. No unexpired sessionid in the Chrome profile gdl reads -> abort before any request, ping /76 (red check). - stories-streak.py: postflight. Every fetched stories source empty for 3 consecutive days -> exit 77, for a session Instagram stopped honouring while the cookie is still present. The 2026-10-07 session death went unnoticed for a week because every run still exited 0. - gdl-sync.py: [n/N] source headers, a 60s heartbeat with new-file counts, and per-source timing, so a healthy paced run no longer looks hung. - gdl-sync@.service now runs run-%i.sh instead of gdl-cron.sh: HC_URL is set in the wrappers, so scheduled runs had never pinged Healthchecks. - Tests for both guards (offline) and the progress helpers.
This commit is contained in:
@@ -0,0 +1,160 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user