instaarchive-viewer/scripts/test_session_guards.py

"""
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()