uni-notify/tests/test_e2e.py

"""Сквозной тест без сети: поддельные страницы 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 = """<!doctype html><title>Mail - Outlook</title><body>
<div role="listbox" id="lb"></div>
<script>
const lb = document.getElementById('lb');
function add(id, txt){ const d=document.createElement('div'); d.setAttribute('role','option'); d.setAttribute('data-convid',id); d.innerText=txt; lb.prepend(d); }
add('c2','Uni Newsletter\\nWeekly newsletter number 42\\nnews news news news\\nMon');
add('c1','Old Sender\\nOld subject that existed before\\nold preview text here\\n09:12');
setTimeout(()=>{ add('c3','Prof. Kowalski\\nKolokwium przeniesione\\nKolokwium z analizy przeniesione na jutro 10:00, sala 204\\n14:05'); }, 5000);
setTimeout(()=>{ new Notification('Prof. Kowalski', {body:'Kolokwium z analizy przeniesione na jutro 10:00, sala 204'}); }, 5300);
setTimeout(()=>{ add('c4','Dziekanat WMI\\nStypendium\\nProsimy o uzupełnienie dokumentów do stypendium do piątku\\n14:07'); }, 7000);
setTimeout(()=>{ add('c5','Uni Newsletter\\nNewsletter number 43 is out\\nmore news and promotions for you\\n14:08'); }, 7500);
// время в карточке «тикает» — не должно давать новых событий
setTimeout(()=>{ lb.querySelector('[data-convid=c1]').innerText='Old Sender\\nOld subject that existed before\\nold preview text here\\n09:13'; }, 8000);
</script></body>"""

TEAMS_HTML = """<!doctype html><title>Chat | Microsoft Teams</title><body>
<div role="tree">
  <div role="treeitem" id="grp">Recent
    <div role="treeitem" data-item-id="t1">Group Projekt\\nAnia: ok see you tomorrow everyone\\n12:00</div>
  </div>
</div>
<script>
setTimeout(()=>{ const t=document.querySelector('[data-item-id=t1]'); t.innerText='Group Projekt\\nAnia: @Me can you send the report by tonight?\\n14:10'; }, 6000);
</script></body>"""


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