feat: harden multi-account backend and add tests
This commit is contained in:
@@ -0,0 +1,604 @@
|
||||
"""
|
||||
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 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()
|
||||
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
|
||||
|
||||
def test_create_job_allows_different_accounts(self):
|
||||
JobRunner = self._import_job_runner()
|
||||
runner = JobRunner()
|
||||
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
|
||||
|
||||
def test_create_job_allows_after_previous_completes(self):
|
||||
JobRunner = self._import_job_runner()
|
||||
runner = JobRunner()
|
||||
j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"})
|
||||
j1.status = "completed"
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user