""" Smoke / integration tests for the multi-account Telegram scraper core logic. These tests do NOT require Telegram credentials or network access. They verify: - Account creation and state isolation - Account import safety (no auto-start continuous) - Account export round-trip - Account deletion (sidecar file cleanup) - Active account persistence helpers - Message search SQL construction - Health payload structure - Continuous orchestrator worker deduplication - load_messages() pagination with search filter """ import os import shutil import sqlite3 import sys import tempfile import threading import time from pathlib import Path from typing import Any, Dict, List, Optional from unittest.mock import MagicMock # ── Mock heavy dependencies before any webui_server import ───────────── # We need to mock qrcode and telethon because they aren't installed in CI. sys.modules["qrcode"] = MagicMock() sys.modules["qrcode.image"] = MagicMock() sys.modules["qrcode.image.svg"] = MagicMock() sys.modules["telethon"] = MagicMock() sys.modules["telethon.errors"] = MagicMock() sys.modules["telethon.errors"].SessionPasswordNeededError = type("SessionPasswordNeededError", (Exception,), {}) sys.modules["telegram_scraper_with_forwarding"] = MagicMock() sys.modules["telegram_scraper_with_forwarding"]._ensure_session_wal = lambda p: None sys.modules["telegram_scraper_with_forwarding"].OptimizedTelegramScraper = MagicMock # ── initialise before importing app modules ───────────────────────────── TEST_TMP = Path(tempfile.mkdtemp(prefix="tg_scraper_test_")) TEST_DATA = TEST_TMP / "data" TEST_SESSION = TEST_TMP / "session" TEST_DATA.mkdir(parents=True, exist_ok=True) TEST_SESSION.mkdir(parents=True, exist_ok=True) os.environ["TELEGRAM_SCRAPER_HOST"] = "127.0.0.1" os.environ["TELEGRAM_SCRAPER_PORT"] = "0" os.environ["TELEGRAM_SCRAPER_START_CONTINUOUS"] = "0" # Now safe to import app modules import app_state # noqa: E402 from app_state import ( # noqa: E402 ACCOUNT_DEFAULTS, account_data_dir, account_exists, account_session_path, get_account_store, get_global_store, list_accounts, load_account, ) from health import health_payload # noqa: E402 def _init_test_env(): """Reset global caches and re-initialise for a clean test.""" app_state._GLOBAL_STORE = None with app_state._account_stores_lock: app_state._ACCOUNT_STORES.clear() _init_test_env() # ── Helpers ──────────────────────────────────────────────────────────────── def make_account_id() -> str: return f"test-{int(time.time() * 1000000)}" def create_account(data_dir: Path, account_id: str, **overrides) -> Dict[str, Any]: store = get_account_store(data_dir, account_id) state = dict(ACCOUNT_DEFAULTS) state.update(overrides) state["label"] = overrides.get("label", account_id) store.save(state) def mutate(g: Dict[str, Any]) -> None: accounts = g.setdefault("accounts", []) if account_id not in accounts: accounts.append(account_id) get_global_store(data_dir).update(mutate) return load_account(data_dir, account_id) def create_channel_db(data_dir: Path, account_id: Optional[str], channel_id: str, messages: List[Dict[str, Any]]) -> None: if account_id: db_dir = account_data_dir(data_dir, account_id) / channel_id else: db_dir = data_dir / channel_id db_dir.mkdir(parents=True, exist_ok=True) db_path = db_dir / f"{channel_id}.db" conn = sqlite3.connect(str(db_path)) conn.execute( """CREATE TABLE IF NOT EXISTS messages ( message_id INTEGER PRIMARY KEY, date TEXT, sender_id INTEGER, first_name TEXT, last_name TEXT, username TEXT, message TEXT, media_type TEXT, media_path TEXT, reply_to INTEGER, post_author TEXT, views INTEGER, forwards INTEGER, reactions TEXT )""" ) for m in messages: conn.execute( "INSERT OR IGNORE INTO messages (message_id, date, message) VALUES (?, ?, ?)", (m["message_id"], m.get("date"), m.get("message")), ) conn.commit() conn.close() # ── Fixture setup / teardown ──────────────────────────────────────────────── def setup_function(): _init_test_env() for child in TEST_DATA.iterdir(): if child.is_dir(): shutil.rmtree(child, ignore_errors=True) else: child.unlink(missing_ok=True) def teardown_function(): _init_test_env() # ═══════════════════════════════════════════════════════════════════════════ # Tests # ═══════════════════════════════════════════════════════════════════════════ class TestAccountLifecycle: """Account creation, isolation, import, export, deletion.""" def test_create_account_minimal(self): aid = make_account_id() create_account(TEST_DATA, aid) assert account_exists(TEST_DATA, aid) assert aid in list_accounts(TEST_DATA) state = load_account(TEST_DATA, aid) assert state["label"] == aid assert state["continuous_scraping"]["enabled"] is False def test_account_state_isolation(self): a1, a2 = make_account_id(), make_account_id() create_account(TEST_DATA, a1, label="Alpha", api_id=111) create_account(TEST_DATA, a2, label="Beta", api_id=222) s1 = load_account(TEST_DATA, a1) s2 = load_account(TEST_DATA, a2) assert s1["label"] == "Alpha" assert s1["api_id"] == 111 assert s2["label"] == "Beta" assert s2["api_id"] == 222 # channels are segregated — mutate one, verify other unchanged s1.setdefault("channels", {})["-100aaa"] = 1 get_account_store(TEST_DATA, a1).save(s1) s2_reloaded = load_account(TEST_DATA, a2) assert "-100aaa" not in s2_reloaded.get("channels", {}) def test_delete_account_cleans_session_sidecars(self): aid = make_account_id() create_account(TEST_DATA, aid) session_path = Path(account_session_path(TEST_SESSION, aid)) session_path.write_text("session-data") Path(str(session_path) + "-wal").write_text("wal") Path(str(session_path) + "-shm").write_text("shm") Path(str(session_path) + "-journal").write_text("journal") acc_dir = account_data_dir(TEST_DATA, aid) if acc_dir.exists(): shutil.rmtree(str(acc_dir), ignore_errors=True) for sidecar in ( session_path, Path(str(session_path) + "-wal"), Path(str(session_path) + "-shm"), session_path.with_suffix(session_path.suffix + "-journal"), ): if sidecar.exists(): sidecar.unlink() assert not session_path.exists() assert not Path(str(session_path) + "-wal").exists() assert not Path(str(session_path) + "-shm").exists() assert not Path(str(session_path) + "-journal").exists() assert not acc_dir.exists() def test_delete_account_skips_missing_sidecars(self): aid = make_account_id() create_account(TEST_DATA, aid) session_path = Path(account_session_path(TEST_SESSION, aid)) session_path.write_text("session-only") acc_dir = account_data_dir(TEST_DATA, aid) if acc_dir.exists(): shutil.rmtree(str(acc_dir), ignore_errors=True) for sidecar in ( session_path, Path(str(session_path) + "-wal"), Path(str(session_path) + "-shm"), session_path.with_suffix(session_path.suffix + "-journal"), ): if sidecar.exists(): sidecar.unlink() assert not session_path.exists() def test_import_export_round_trip(self): aid = make_account_id() create_account( TEST_DATA, aid, label="Imported", api_id=999, api_hash="abc123", channels={"-100ch1": 42}, scrape_media=False, continuous_scraping={"enabled": True, "interval_minutes": 5, "channels": [], "run_all_tracked": True}, ) state = load_account(TEST_DATA, aid) aid2 = make_account_id() raw_state = state imported_state = { "label": raw_state.get("label", aid2), "api_id": raw_state.get("api_id"), "api_hash": raw_state.get("api_hash"), "channels": raw_state.get("channels", {}), "channel_names": raw_state.get("channel_names", {}), "scrape_media": bool(raw_state.get("scrape_media", True)), "forwarding_rules": raw_state.get("forwarding_rules", []), "continuous_scraping": raw_state.get("continuous_scraping") if isinstance(raw_state.get("continuous_scraping"), dict) else { "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, }, } get_account_store(TEST_DATA, aid2).save(imported_state) def mutate_global(g): g.setdefault("accounts", []).append(aid2) get_global_store(TEST_DATA).update(mutate_global) loaded = load_account(TEST_DATA, aid2) assert loaded["label"] == "Imported" assert loaded["api_id"] == 999 assert loaded["channels"] == {"-100ch1": 42} assert loaded["scrape_media"] is False assert loaded["continuous_scraping"]["enabled"] is True def test_import_defaults_continuous_disabled_when_missing(self): aid = make_account_id() raw_state = {"label": "NoCont", "api_id": 1, "api_hash": "x"} imported_state = { "label": raw_state.get("label", aid), "api_id": raw_state.get("api_id"), "api_hash": raw_state.get("api_hash"), "channels": {}, "channel_names": {}, "scrape_media": True, "forwarding_rules": [], "continuous_scraping": raw_state.get("continuous_scraping") if isinstance(raw_state.get("continuous_scraping"), dict) else { "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, }, } get_account_store(TEST_DATA, aid).save(imported_state) def mutate_global(g): g.setdefault("accounts", []).append(aid) get_global_store(TEST_DATA).update(mutate_global) loaded = load_account(TEST_DATA, aid) assert loaded["continuous_scraping"]["enabled"] is False assert loaded["continuous_scraping"]["run_all_tracked"] is True class TestInputSafety: """Small safety checks for values that flow into files or exports.""" def test_channel_id_rejects_path_characters(self): import webui_server as ws_module assert ws_module.normalize_channel_id("@valid_name") == "valid_name" for bad in ("../escape", "nested/channel", r"nested\channel", ".", "..", ""): try: ws_module.normalize_channel_id(bad) except ValueError: continue raise AssertionError(f"accepted unsafe channel_id: {bad!r}") def test_export_account_state_redacts_api_hash_by_default(self): import webui_server as ws_module state = {"label": "Work", "api_id": 123, "api_hash": "secret", "channels": {}} redacted = ws_module.export_account_state(state) assert redacted["api_hash"] is None assert redacted["api_hash_present"] is True assert state["api_hash"] == "secret" full = ws_module.export_account_state(state, include_secrets=True) assert full["api_hash"] == "secret" assert "api_hash_present" not in full class TestContinuousOrchestrator: """Continuous scraping manager safety using mocked webui_server.""" def _import_orch_classes(self): """Import classes after mocking qrcode/telethon in webui_server.""" import webui_server as ws_module return ws_module.PerAccountContinuousScrapeManager, ws_module.ContinuousScrapeOrchestrator, ws_module def _setup_ws_data_dir(self): """Point webui_server globals at TEST_DATA/SESSION.""" import webui_server as ws_module self._ws_orig_data = ws_module.DATA_DIR self._ws_orig_session = ws_module.SESSION_DIR ws_module.DATA_DIR = TEST_DATA ws_module.SESSION_DIR = TEST_SESSION ws_module.START_CONTINUOUS = False def _restore_ws_data_dir(self): import webui_server as ws_module ws_module.DATA_DIR = self._ws_orig_data ws_module.SESSION_DIR = self._ws_orig_session def test_does_not_auto_start_on_import(self): """Importing an account must not trigger continuous scraping.""" PerAccountContinuousScrapeManager, ContinuousScrapeOrchestrator, ws = self._import_orch_classes() self._setup_ws_data_dir() try: orch = ContinuousScrapeOrchestrator() aid = make_account_id() create_account(TEST_DATA, aid, continuous_scraping={ "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, }) orch.add_account(aid, auto_start=False) snap = orch.snapshot_for(aid) assert snap["status"]["running"] is False finally: self._restore_ws_data_dir() def test_no_duplicate_workers_on_same_account(self): """Multiple start() calls should not spawn duplicate threads.""" PerAccountContinuousScrapeManager, _, ws = self._import_orch_classes() self._setup_ws_data_dir() try: aid = make_account_id() # Must set enabled=True or refresh_config() will stop the thread create_account(TEST_DATA, aid, continuous_scraping={ "enabled": True, "interval_minutes": 60, "channels": [], "run_all_tracked": True, }) mgr = PerAccountContinuousScrapeManager(aid) # Override refresh_config to prevent auth check from interfering original_refresh = mgr.refresh_config mgr.refresh_config = lambda: None mgr.start() t1 = mgr.thread mgr.start() # second call — should be no-op t2 = mgr.thread assert t1 is t2, "start() spawned a second thread" # Also verify only one thread is alive for this manager alive_count = sum( 1 for t in threading.enumerate() if t is t1 or t is t2 ) assert alive_count <= 1, "duplicate threads detected" mgr.stop() mgr.refresh_config = original_refresh finally: self._restore_ws_data_dir() def test_disabled_account_not_started(self): """start_all() must not start accounts with enabled=False.""" _, ContinuousScrapeOrchestrator, ws = self._import_orch_classes() self._setup_ws_data_dir() try: orch = ContinuousScrapeOrchestrator() aid = make_account_id() create_account(TEST_DATA, aid, continuous_scraping={ "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, }) orch.start_all() snap = orch.snapshot_for(aid) assert snap["status"]["running"] is False finally: self._restore_ws_data_dir() class TestMessageSearch: """load_messages() SQL correctness with pagination and search.""" def _import_load_messages(self): import webui_server as ws_module # Point at TEST_DATA self._ws_orig = ws_module.DATA_DIR ws_module.DATA_DIR = TEST_DATA return ws_module.load_messages def _restore(self): import webui_server as ws_module ws_module.DATA_DIR = self._ws_orig def test_search_filters_correctly(self): channel_id = "-100searchtest" messages = [ {"message_id": 1, "date": "2024-01-01", "message": "hello world"}, {"message_id": 2, "date": "2024-01-02", "message": "foo bar baz"}, {"message_id": 3, "date": "2024-01-03", "message": "hello again"}, ] create_channel_db(TEST_DATA, "default", channel_id, messages) load_messages = self._import_load_messages() try: result = load_messages("default", channel_id, limit=100, search="hello") finally: self._restore() texts = [m["text"] for m in result] assert "hello world" in texts assert "hello again" in texts assert "foo bar baz" not in texts def test_before_id_pagination_works(self): channel_id = "-100pagination" messages = [ {"message_id": i, "date": f"2024-01-{i:02d}", "message": f"msg-{i}"} for i in range(1, 21) ] create_channel_db(TEST_DATA, "default", channel_id, messages) load_messages = self._import_load_messages() try: result = load_messages("default", channel_id, limit=5, before_message_id=15) finally: self._restore() assert len(result) == 5 ids = [m["message_id"] for m in result] assert all(i < 15 for i in ids) # load_messages returns rows in ascending message_id order (reversed from DESC query) assert ids == [10, 11, 12, 13, 14] def test_search_with_pagination(self): channel_id = "-100searchpages" messages = [ {"message_id": i, "date": f"2024-01-{i:02d}", "message": f"hello-{i}" if i % 2 else f"other-{i}"} for i in range(1, 21) ] create_channel_db(TEST_DATA, "default", channel_id, messages) load_messages = self._import_load_messages() try: result = load_messages("default", channel_id, limit=3, before_message_id=15, search="hello") finally: self._restore() # IDs matching "hello-" below 15: 13, 11, 9, 7, 5, 3, 1 → LIMIT 3 → [9, 11, 13] (ascending) ids = [m["message_id"] for m in result] assert ids == [9, 11, 13], f"got {ids}" assert all("hello" in m["text"] for m in result) def test_empty_result_no_error(self): channel_id = "-100emptysearch" messages = [{"message_id": 1, "date": "2024-01-01", "message": "only one"}] create_channel_db(TEST_DATA, "default", channel_id, messages) load_messages = self._import_load_messages() try: result = load_messages("default", channel_id, limit=100, search="nonexistent") finally: self._restore() assert result == [] class TestHealthPayload: """Health endpoint structure.""" def test_health_payload_structure(self): aid = make_account_id() create_account(TEST_DATA, aid) payload = health_payload( TEST_DATA, TEST_SESSION, get_global_store(TEST_DATA), {}, 0, [aid], ) assert "ok" in payload assert "status" in payload assert "checks" in payload checks = payload["checks"] assert "data_dir" in checks assert "session_dir" in checks assert "accounts" in checks assert "items" in checks["accounts"] assert aid in checks["accounts"]["items"] def test_health_accounts_wrapper(self): a1, a2 = make_account_id(), make_account_id() create_account(TEST_DATA, a1) create_account(TEST_DATA, a2) payload = health_payload( TEST_DATA, TEST_SESSION, get_global_store(TEST_DATA), {}, 0, [a1, a2], ) acc_checks = payload["checks"]["accounts"] assert acc_checks["ok"] is True assert len(acc_checks["items"]) == 2 class TestPersistence: """UI persistence helpers.""" def test_active_account_localstorage(self): """Verifies the save/load logic pattern used in app.js.""" accounts = [ {"id": "work", "label": "Work"}, {"id": "home", "label": "Home"}, ] saved_id = "work" loaded = None if saved_id and any(a["id"] == saved_id for a in accounts): loaded = saved_id if not loaded: loaded = next((a["id"] for a in accounts), None) assert loaded == "work" saved_id = "longgone" loaded = None if saved_id and any(a["id"] == saved_id for a in accounts): loaded = saved_id if not loaded and accounts: loaded = accounts[0]["id"] assert loaded == "work" def test_viewer_fallback_on_missing_account(self): """Viewer's loadViewerAccount fallback logic.""" accounts = [ {"id": "alpha", "label": "Alpha"}, {"id": "beta", "label": "Beta"}, ] requested = "nonexistent" requested_exists = requested and any(a["id"] == requested for a in accounts) account_id = requested if requested_exists else (accounts[0]["id"] if accounts else None) assert account_id == "alpha" account_id2 = requested if (requested and any(a["id"] == requested for a in [])) else None assert account_id2 is None class TestJobDeduplicationSchema: """JobRunner create_job dedup logic.""" def _import_job_runner(self): import webui_server as ws_module return ws_module.JobRunner def test_create_job_dedup_by_account(self): JobRunner = self._import_job_runner() runner = JobRunner() runner.queue.put = lambda job: None j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"}) assert j1.job_id == j2.job_id, "duplicate job was created instead of being reused" assert "Reused existing active job" in j2.logs runner.shutdown(timeout=0) def test_create_job_allows_different_accounts(self): JobRunner = self._import_job_runner() runner = JobRunner() runner.queue.put = lambda job: None j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) time.sleep(0.002) # ensure different timestamp -> different job_id j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc2"}) assert j1.job_id != j2.job_id runner.shutdown(timeout=0) def test_create_job_allows_after_previous_completes(self): JobRunner = self._import_job_runner() runner = JobRunner() runner.queue.put = lambda job: None j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) j1.status = "done" time.sleep(0.002) # ensure different timestamp -> different job_id j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"}) assert j1.job_id != j2.job_id runner.shutdown(timeout=0) class TestAccountHealthSummary: """account_health_summary structure.""" def _import_health_summary(self): import webui_server as ws_module self._ws_orig_data = ws_module.DATA_DIR self._ws_orig_session = ws_module.SESSION_DIR ws_module.DATA_DIR = TEST_DATA ws_module.SESSION_DIR = TEST_SESSION return ws_module.account_health_summary, ws_module.JobRunner, ws_module def test_health_summary_structure(self): account_health_summary, JobRunner, ws = self._import_health_summary() aid = make_account_id() create_account(TEST_DATA, aid, api_id=1, api_hash="x") try: runner = JobRunner() health = account_health_summary(aid, runner) finally: ws.DATA_DIR = self._ws_orig_data ws.SESSION_DIR = self._ws_orig_session assert health["account_id"] == aid assert "label" in health assert "data_dir_exists" in health assert "session_ready" in health assert "api_credentials" in health assert health["api_credentials"] is True assert "channel_count" in health assert "message_count" in health assert "media_count" in health # ── Cleanup all temp data ────────────────────────────────────────────────── def cleanup_test_data(): if TEST_TMP.exists(): shutil.rmtree(str(TEST_TMP), ignore_errors=True) import atexit # noqa: E402 atexit.register(cleanup_test_data) # noqa: E402