639 lines
25 KiB
Python
639 lines
25 KiB
Python
"""
|
|
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
|