"""Сквозной тест без сети: поддельные страницы Outlook/Teams → watcher.js → pipeline → «Telegram».
Запуск: CHROMIUM_PATH=/usr/bin/google-chrome python -m tests.test_e2e
"""
import asyncio
import os
import shutil
import tempfile
TMP = tempfile.mkdtemp(prefix="uni-test-")
os.environ.update(TG_TOKEN="1:x", OWNER_ID="1", OPENROUTER_API_KEY="x", UNI_EMAIL="me@uni.edu",
DATA_DIR=TMP, HEADLESS="true", BATCH_WINDOW="1.5", VIP="dziekanat", MUTE="newsletter")
from app import config, pipeline as pl # noqa: E402
from app.browser import session as bs # noqa: E402
from app.db import DB # noqa: E402
pl.BASELINE_SECONDS = 3
MAIL_HTML = """Mail - Outlook
"""
TEAMS_HTML = """Chat | Microsoft Teams
Recent
Group Projekt\\nAnia: ok see you tomorrow everyone\\n12:00
"""
async def main():
s = config.load()
db = DB(s.db_path)
sent = []
async def say(text, silent=False, photo=None):
sent.append((text, silent))
print("\n>>> TG", "(silent)" if silent else "", "\n" + text)
class FakeLLM:
calls = 0
async def triage(self, events, rules, my_name):
FakeLLM.calls += 1
out = []
for i, e in enumerate(events):
t = (e.get("body", "") + e.get("title", "") + e.get("sender", "")).lower()
lvl = 3 if "kolokwium" in t else 2 if "@me" in t else 1
out.append({"i": i, "level": lvl, "sender": e.get("sender", ""), "summary": f"[summary] {t[:60]}"})
return out
p = pl.Pipeline(s, db, FakeLLM(), say)
bm = bs.BrowserManager(s, p.submit, say)
orig_goto = bm._goto
async def goto_with_routes(name):
if not getattr(bm, "_routed", False):
bm._routed = True
await bm.ctx.route("https://outlook.office.com/**", lambda r: r.fulfill(body=MAIL_HTML, content_type="text/html; charset=utf-8"))
await bm.ctx.route("https://teams.microsoft.com/**", lambda r: r.fulfill(body=TEAMS_HTML, content_type="text/html; charset=utf-8"))
await orig_goto(name)
bm._goto = goto_with_routes
await bm.start()
runner = asyncio.create_task(p.run())
await asyncio.sleep(14)
print("\nstates:", bm.states())
print("debug mail:", await bm.debug("mail"))
rows = db.conn.execute("SELECT source, kind, status, level, sender FROM items ORDER BY id").fetchall()
print("\nDB:")
for r in rows:
print(" ", tuple(r))
texts = "\n".join(t for t, _ in sent)
assert bm.states() == {"mail": "ok", "teams": "ok"}, bm.states()
assert texts.count("Kolokwium z analizy") >= 1 and texts.lower().count("kolokwium przeniesione") <= 1, "kolokwium: ровно одно уведомление"
kolok_msgs = [t for t, _ in sent if "kolokwium" in t.lower()]
assert len(kolok_msgs) == 1, f"дубль уведомления: {len(kolok_msgs)}"
assert any("Dziekanat" in t and "🔴" in t for t, _ in sent), "VIP должен быть срочным"
assert not any("number 43" in t for t, _ in sent), "MUTE не должен приходить"
assert any("@me" in t.lower() for t, _ in sent), "Teams: упоминание должно прийти"
assert not any("Old subject" in t for t, _ in sent), "baseline не должен уведомлять"
statuses = {r["status"] for r in rows}
assert "baseline" in statuses
# тиканье времени не создало новое событие для c1
c1 = [r for r in rows if r["sender"] == "Old Sender"]
assert len(c1) == 1 and c1[0]["status"] == "baseline", c1
print(f"\n✅ ВСЁ ОК. Сообщений в TG: {len(sent)}, вызовов LLM: {FakeLLM.calls}")
runner.cancel()
await bm.close()
shutil.rmtree(TMP, ignore_errors=True)
if __name__ == "__main__":
asyncio.run(main())