Spaces:

cipher-attacks
/

Dbu-central-bot

Running on Zero

analyzer + other-stream courses
b473ad4 1 day ago
33.6 kB
"""Pure-logic tests (no Telegram/DB needed). Run: python -m pytest -q   (or python tests/test_logic.py)"""
import sys
from datetime import date, timedelta
from pathlib import Path
from types import SimpleNamespace

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import asyncio  # noqa: E402

from app.core import cache  # noqa: E402
from app.core.aio import gather_results, gather_safe  # noqa: E402
from app.services.ingest import clean_title, parse_caption  # noqa: E402


def _user(**kw):
    base = dict(streak=0, best_streak=0, last_active_on=None, points=0)
    base.update(kw)
    return SimpleNamespace(**base)


def test_parse_caption_ok():
    assert parse_caption("3 exam Midterm 2017") == (3, "exam", "Midterm 2017")
    assert parse_caption("12 #slides") == (12, "slide", None)


def test_parse_caption_bad():
    assert parse_caption(None) is None
    assert parse_caption("exam 3") is None
    assert parse_caption("3 unknownkind x") is None


def test_cache_roundtrip_and_isolation():
    cache.put(7, {"points": 5})
    got = cache.get(7)
    assert got == {"points": 5}
    got["points"] = 99  # mutating the copy must not touch the cache
    assert cache.get(7) == {"points": 5}
    cache.invalidate(7)
    assert cache.get(7) is None


class _Unhashable:
    """Mimics aiogram method objects: awaitable, but unhashable (pydantic model)."""

    __hash__ = None

    def __init__(self, value):
        self.value = value

    def __await__(self):
        async def run():
            return self.value

        return run().__await__()


def test_plain_gather_fails_but_ours_works():
    async def go():
        try:
            await asyncio.gather(_Unhashable(1))
            plain_ok = True
        except TypeError:
            plain_ok = False
        assert plain_ok is False  # this is the bug that broke the library buttons
        assert await gather_results(_Unhashable(1), _Unhashable(2)) == [1, 2]
        await gather_safe(_Unhashable(3))

    asyncio.run(go())


def test_clean_title():
    assert clean_title("Math1003_Module_Final.pdf", None, "x") == "Math1003 Module Final"
    assert clean_title(None, "Physics slides\n#physics", "x") == "Physics slides"
    assert clean_title(None, None, "fallback") == "fallback"


import random  # noqa: E402

from app.core import season  # noqa: E402
from app.services import xp_rules as R  # noqa: E402
from app.services.fuzzy import match_score  # noqa: E402
from types import SimpleNamespace as NS  # noqa: E402
from app.core.seed_courses import FRESHMAN_COURSES  # noqa: E402
from app.services import classifier as C  # noqa: E402
from app.services.season_logic import Candidate, draw, flags, rank_candidates  # noqa: E402
from datetime import datetime, timedelta  # noqa: E402


def test_checkin_xp_ramps_then_caps():
    assert [R.checkin_xp(n) for n in (1, 2, 3, 4, 5, 6, 30)] == [3, 4, 5, 6, 7, 7, 7]


def test_levels():
    assert R.level_for(0) == (1, 25)
    assert R.level_for(25) == (2, 50)
    assert R.level_for(2300) == (10, None)
    assert R.level_for(10_000) == (10, None)


def test_next_streak():
    assert R.next_streak(None, "2026-10-14", 0, "2026-10-13") == 1
    assert R.next_streak("2026-10-13", "2026-10-14", 4, "2026-10-13") == 5
    assert R.next_streak("2026-10-10", "2026-10-14", 4, "2026-10-13") == 1
    assert R.next_streak("2026-10-14", "2026-10-14", 4, "2026-10-13") == 4


def test_week_helpers():
    monday = datetime(2026, 10, 12, 0, 5, tzinfo=season.ET)
    assert season.week_key(monday) == "2026W42"
    assert season.week_key(monday - timedelta(minutes=10)) == "2026W41"
    assert season.previous_week_key(monday) == "2026W41"
    assert season.days_left(monday) == 7
    assert season.days_left(datetime(2026, 10, 18, 23, 0, tzinfo=season.ET)) == 1


def _c(uid, xp, days=5, trust=100, banned=False, total=10, ref=0):
    return Candidate(uid, xp, days, trust, banned, total, ref)


def test_ranking_filters_cheaters_and_orders():
    cands = [
        _c(1, 90), _c(2, 120, days=2),  # too few active days
        _c(3, 100, trust=50),  # flagged
        _c(4, 100, banned=True), _c(5, 30),  # under min xp
        _c(6, 100, total=2),  # brand-new account
        _c(7, 95), _c(8, 95, days=7),
    ]
    ranked = rank_candidates(cands, min_xp=40, min_days=4)
    # 95xp/7 days beats 95xp/5 days beats 90xp; everyone else is filtered out
    assert [c.user_id for c in ranked] == [8, 7, 1]


def test_flags():
    assert flags(_c(1, 100, ref=80)) == ["referral-heavy"]
    assert "low-trust" in flags(_c(1, 100, trust=10))


def test_draw_is_weighted_and_deterministic_with_seed():
    pool = [_c(1, 40), _c(2, 120)]
    picks = [draw(pool, random.Random(i)).user_id for i in range(400)]
    assert picks.count(2) > picks.count(1)
    assert draw([], random.Random(1)) is None


def test_fuzzy_search_tolerates_typos():
    physics = ["General Physics", "Phys1009"]
    assert match_score(["phyy"], physics) is not None  # the typo from the screenshot
    assert match_score(["physic"], physics) is not None
    assert match_score(["calc"], ["Calculus I", "Math1011"]) is not None
    assert match_score(["biology"], physics) is None
    assert match_score(["english", "skills"], ["Communicative English Language Skills I"]) is not None
    assert match_score(["english", "physics"], ["Communicative English Language Skills I"]) is None
    # better matches rank higher
    assert match_score(["physics"], physics) > match_score(["phyy"], physics)


COURSES = [
    NS(id=i, stream=None if st == "all" else st, year=y, code=code, name=name)
    for i, (st, y, _sem, code, name) in enumerate(FRESHMAN_COURSES, start=1)
]


def _pick(file_name, caption="", text=""):
    sources = [(file_name, "name"), (caption, "name"), (text, "text")]
    best, confident, sugg = C.decide(C.rank_courses(COURSES, sources))
    return (best.name if best else None), confident


def test_classifier_courses():
    assert _pick("Math_1003_Module.pdf") == ("Mathematics for Natural Sciences", True)
    assert _pick("General Physics Final Exam 2016.pdf") == ("General Physics", True)
    assert _pick("Entrepreneurship ppt chapter 1.pptx") == ("Entrepreneurship", True)
    assert _pick("Eng 1002 - Communicative English II notes.pdf")[0] == "Communicative English Language Skills II"
    assert _pick("english skills I handout.pdf")[0] == "Communicative English Language Skills I"
    assert _pick("chem module.pdf")[0] == "General Chemistry"
    assert _pick("calculus chapter 3.pdf")[0] == "Mathematics for Natural Sciences"
    assert _pick("ታሪክ ማስታወሻ.pdf")[0] == "History of Ethiopia and the Horn"


def test_classifier_unsure_goes_to_review():
    assert _pick("IMG_20261014_final.pdf")[1] is False
    assert _pick("random_file123.pdf") == (None, False)
    assert _pick("Mathematics module.pdf")[1] is False  # three math courses tie: ask the admin


def test_classifier_uses_pdf_text_when_name_is_useless():
    text = "Debre Berhan University General Chemistry module atoms and molecules chapter one"
    assert _pick("scan0001.pdf", text=text)[0] == "General Chemistry"


def test_kind_detection():
    assert C.detect_kind("Phys final exam.pdf", "", "", None, None, ".pdf")[0] == "exam"
    assert C.detect_kind("lecture1.pptx", "", "", None, None, ".pptx")[0] == "slide"
    assert C.detect_kind("Math module.pdf", "", "", 120, None, ".pdf")[0] == "note"
    assert C.detect_kind("Math module.pdf", "", "", 260, None, ".pdf")[0] == "note"
    assert C.detect_kind("calculus.pdf", "", "", 400, None, ".pdf")[0] == "book"
    assert C.detect_kind("Mathematics for Social Sciences.pdf", "", "", 280, None, ".pdf")[0] == "note"  # was #6
    assert C.detect_kind("Geography Module Final.pdf", "", "", 150, None, ".pdf")[0] == "note"  # was #3
    assert C.detect_kind("Physics final.pdf", "", "", None, None, ".pdf")[0] == "exam"
    assert C.detect_kind("Phys mid.pdf", "", "", None, None, ".pdf")[0] == "exam"
    assert C.detect_kind("scan.pdf", "", "Time allowed: 2 hours. Answer all questions", None, None, ".pdf")[0] == "exam"
    assert C.detect_kind("whatever.pdf", "", "", None, None, ".pdf")[0] == "note"


def test_titles():
    assert C.clean_title("Phys_1009_FINAL_EXAM_(1).pdf", None, "General Physics", "exam", "2016", "Final") == "Phys 1009 FINAL EXAM 2016"
    assert C.clean_title("@some_channel 123456789012 notes.pdf", None, "X", "note", None, None) == "Notes"
    assert C.clean_title(None, None, "General Physics", "note", None, None) == "General Physics note"


def test_same_title_different_files_are_told_apart():
    taken = {C.norm_title_key("Math Module")}
    assert C.disambiguate("Math Module", taken, year=None, pages=245, size=5_000_000) == "Math Module (245 p)"
    taken.add(C.norm_title_key("Math Module (245 p)"))
    assert C.disambiguate("Math Module", taken, year=None, pages=245, size=5_000_000) == "Math Module (4.8 MB)"
    taken.add(C.norm_title_key("Math Module (4.8 MB)"))
    assert C.disambiguate("Math Module", taken, year=None, pages=245, size=5_000_000) == "Math Module (2)"


def test_simhash_near_duplicates():
    base = " ".join(f"word{i % 37} term{i % 11}" for i in range(400))
    edited = base + " extra footer line added by a scanner app"
    other = " ".join(f"zeta{i % 53} omega{i % 7}" for i in range(400))
    a, b, c = C.simhash64(base), C.simhash64(edited), C.simhash64(other)
    assert C.hamming(a, b) <= C.NEAR_DUPLICATE_BITS
    assert C.hamming(a, c) > 10
    assert C.simhash64("too short") is None


def test_doc_analyzer_reads_pptx_and_docx():
    import io
    import zipfile

    from app.services.doc_analyzer import analyze

    def make(files: dict) -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w") as z:
            for name, content in files.items():
                z.writestr(name, content)
        return buf.getvalue()

    pptx = make({
        "ppt/slides/slide1.xml": "<a:t>General Chemistry</a:t><a:t>Atoms &amp; molecules</a:t>",
        "ppt/slides/slide2.xml": "<a:t>Chapter one</a:t>",
        "ppt/slides/slide10.xml": "<a:t>Last</a:t>",
    })
    info = analyze("chem.pptx", pptx)
    assert info.pages == 3 and "General Chemistry" in info.text and "Atoms & molecules" in info.text
    docx = make({"word/document.xml": "<w:p><w:t>Entrepreneurship</w:t><w:t xml:space='preserve'> module</w:t></w:p>"})
    assert "Entrepreneurship module" in analyze("m.docx", docx).text
    assert analyze("broken.pdf", b"not a pdf").text == ""  # never raises
    assert analyze("pic.png", b"x").pages is None


def test_watermark_geometry_roundtrip():
    from app.services import watermark as W

    w, h = 595.0, 842.0

    def forward(x, y, r):  # page space -> what the reader sees (the inverse of to_page_space)
        return {0: (x, y), 90: (y, w - x), 180: (w - x, h - y), 270: (h - y, x)}[r]

    for r in (0, 90, 180, 270):
        for x, y in ((10, 20), (300, 400), (595, 842)):
            dx, dy = forward(x, y, r)
            assert W.to_page_space(dx, dy, w, h, r) == (x, y), (r, x, y)
    assert W.display_size(w, h, 90) == (h, w) and W.display_size(w, h, 0) == (w, h)
    assert W.display_url("https://t.me/DBUCentral/") == "t.me/DBUCentral"
    x0, y0, x1, y1 = W.rect_to_page_space(0, 0, 100, 20, w, h, 90)
    assert x0 < x1 and y0 < y1


def test_watermark_stamps_real_pdfs():
    try:
        import io

        from pypdf import PdfReader, PdfWriter
        from reportlab.pdfgen import canvas
    except ImportError:
        return  # libraries not installed here; the bot's requirements.txt installs them
    from app.services import watermark as W

    buf = io.BytesIO()
    c = canvas.Canvas(buf)
    for i in range(3):
        c.setPageSize((595, 842) if i != 1 else (842, 595))
        c.drawString(40, 500, f"Hello {i}")
        c.showPage()
    c.save()
    style = W.Style("DBU Freshman Central", "https://t.me/DBUCentral", True)
    out = W.stamp_pdf(buf.getvalue(), style)
    pages = PdfReader(io.BytesIO(out)).pages
    assert len(pages) == 3
    assert all("t.me/DBUCentral" in p.extract_text() and f"Hello {i}" in p.extract_text() for i, p in enumerate(pages))
    try:
        W.stamp_pdf(out, style)
        raise AssertionError("double stamp allowed")
    except ValueError:
        pass


def test_real_file_names_from_the_channel():
    # non-course files go to the "general resources" bucket instead of being forced into a course
    for name in ("Department cutt point updated.pdf", "quantitative-&-logical-reasoning-509050.pdf", "verbal-reasoning-247769.pdf"):
        assert _pick(name) == ("Freshman Guide and General Resources", True), name
    assert _pick("Revised History Module 2024 3.pdf") == ("History of Ethiopia and the Horn", True)
    assert _pick("Pysychology Module %282%29.pdf") == ("General Psychology", True)  # typo + url-encoding
    assert _pick("Physical Fitness Module final FINAL.pdf") == ("Physical Fitness", True)
    assert _pick("General Physics Phys 1011 Module.pdf") == ("General Physics", True)
    # an album caption must not drag a Physics file into a "Social ..." course
    best, confident, _ = C.decide(C.rank_courses(COURSES, [("General Physics Module.pdf", "name"), ("First Semester Freshman Courses For Social Science Students.", "text")]))
    assert best.name == "General Physics" and confident
    # genuinely ambiguous names still ask the admin
    assert _pick("Communicative English Language Skills.pdf")[1] is False


def test_title_cleanup_from_screenshots():
    assert C.clean_title("Pysychology Module %282%29.pdf", None, "x", "note", None, None) == "Pysychology Module"
    assert C.clean_title("Physical Fitness Module final FINAL.pdf", None, "x", "note", None, None) == "Physical Fitness Module Final"
    assert C.clean_title("Revised History Module 2024 (3).pdf", None, "x", "note", None, None) == "Revised History Module 2024"


def test_gate_handles():
    from app.core.handles import parse_handles

    assert parse_handles("@DBUCentral, DBUcentralforum @dbucentral") == ["@DBUCentral", "@DBUcentralforum"]
    assert parse_handles("") == [] and parse_handles(" , ") == []


def test_display_title_cleans_old_titles_but_not_manual_ones():
    assert C.display_title("Pysychology Module %282%29") == "Pysychology Module (2)"
    assert C.display_title("Physical Fitness Module final FINAL") == "Physical Fitness Module final"
    assert C.display_title("Geography_Module__Final") == "Geography Module Final"
    assert C.display_title("Math Module (2)") == "Math Module (2)"  # disambiguation marker kept
    assert C.display_title("Chapter 1-3 notes") == "Chapter 1-3 notes"  # hand-written titles untouched


def test_course_label_fits_one_line():
    from app.core.labels import course_label

    assert course_label("Math1003", "Calculus") == "Math1003 · Calculus"
    assert course_label(None, "General Biology") == "General Biology"
    long_ = course_label("FLEn1001", "Communicative English Language Skills I")
    assert len(long_) <= 36 and long_.startswith("Communicative English")


def test_bot_is_english_only_and_every_text_works():
    """No Amharic file, the language argument is ignored, every key used in the code exists and formats."""
    import json
    import re

    root = Path(__file__).resolve().parents[1]
    assert not (root / "locales" / "am.json").exists()
    strings = json.loads((root / "locales" / "en.json").read_text(encoding="utf-8"))
    assert not [k for k, v in strings.items() if re.search(r"[\u1200-\u137f]", v)], "Amharic text left in the UI"
    from app.core.i18n import t

    assert t("am", "welcome", name="X") == t("en", "welcome", name="X")  # old call sites keep working

    used = set()
    for f in (root / "app").rglob("*.py"):
        src = f.read_text(encoding="utf-8")
        used |= set(re.findall(r'\bt\(\s*[\w".]+\s*,\s*"([a-z_0-9]+)"', src))
        used |= set(re.findall(r'all_translations\("([a-z_0-9]+)"\)', src))
    dynamic = {f"level_{i}" for i in range(1, 11)} | {
        "stream_natural", "stream_social", "badge_bronze", "badge_silver", "badge_gold",
        "kind_slide", "kind_exam", "kind_note", "kind_book", "contrib_reason_new", "contrib_reason_blocked",
        "contrib_reason_pending", "contrib_reason_daily", "contrib_rejected_duplicate", "contrib_rejected_quality",
        "contrib_rejected_notallowed",
    }
    assert sorted((used | dynamic) - set(strings)) == [], "a text used in the code is missing from en.json"
    for key, text in strings.items():
        text.format(**{p: "x" for p in re.findall(r"\{(\w+)\}", text)})  # no stray braces


def test_every_command_is_in_the_manual():
    """If you add a command and forget the manual, this fails. And the manual never lists a phantom command."""
    import re

    from app.bot import manual

    root = Path(__file__).resolve().parents[1] / "app" / "bot" / "handlers"
    in_code = {"start"}  # CommandStart()
    for f in root.glob("*.py"):
        for args in re.findall(r"\bCommand\(([^)]*)\)", f.read_text(encoding="utf-8")):
            in_code |= set(re.findall(r'"([a-z_]+)"', args))
    documented = manual.all_command_names()
    assert in_code - documented == set(), f"missing from /help: {sorted(in_code - documented)}"
    assert documented - in_code == set(), f"documented but not in the bot: {sorted(documented - in_code)}"
    text = "\n".join(manual.render())
    for name in documented:
        assert f"/{name}" in text


def test_manual_fits_telegram_and_html_is_balanced():
    import re

    from app.bot import manual

    for message in manual.render():
        assert len(message) <= 4096
        for tag in ("b", "code"):
            assert len(re.findall(rf"<{tag}>", message)) == len(re.findall(rf"</{tag}>", message)), (tag, message[:60])
    descriptions = manual.menu_commands()
    assert all(3 <= len(d) <= 256 for _, d in descriptions)
    assert all(re.fullmatch(r"[a-z0-9_]{1,32}", n) for n, _ in descriptions)  # Telegram's rule for command names


def test_announcement_matches_the_settings():
    from app.bot import manual

    on = manual.announcement_text({"prizes": [100, 60, 40], "raffle": 20, "min_days": 3, "min_xp": 30}, "MyBot")
    assert "@MyBot" in on and "100 birr" in on and "20 birr raffle" in on and "3+ days" in on and "30+ XP" in on
    off = manual.announcement_text({"prizes_enabled": False}, "MyBot")
    assert "WEEKLY PRIZES" not in off and "birr" not in off  # money is only mentioned when it is switched on


def test_growth_payloads_and_links():
    from app.services import growth as G

    assert G.parse_start_payload(None) == ("none", "")
    assert G.parse_start_payload("7187749925") == ("ref", "7187749925")
    assert G.parse_start_payload("c12") == ("course", "12")
    assert G.parse_start_payload("f40") == ("file", "40")
    assert G.parse_start_payload("Dorm5") == ("tag", "dorm5")
    assert G.parse_start_payload("x" * 40) == ("none", "")  # too long / junk is ignored
    assert [G.source_label(x) for x in (None, "123", "c1", "inline", "poster")] == ["organic", "referral", "link", "inline", "poster"]
    url = G.share_url("https://t.me/Bot?start=1", "ሰላም & hello")
    assert url.startswith("https://t.me/share/url?url=https%3A%2F%2Ft.me%2FBot%3Fstart%3D1&text=")
    assert "%26" in url and " " not in url  # '&' in the text is encoded, no raw spaces
    assert G.safe_tag("dorm_5") == "dorm_5" and G.safe_tag("123") == "poster" and G.safe_tag("bad tag!") == "poster"


def test_poster_is_a_valid_one_page_pdf():
    try:
        import io

        from pypdf import PdfReader
        from reportlab.pdfgen import canvas  # noqa: F401
    except ImportError:
        return
    from app.services.poster import make_poster_pdf

    data = make_poster_pdf("https://t.me/DBUFreshBot?start=poster", "DBUFreshBot", "https://t.me/DBUCentral")
    reader = PdfReader(io.BytesIO(data))
    assert len(reader.pages) == 1
    text = reader.pages[0].extract_text()
    assert "@DBUFreshBot" in text and "DBU Freshman Central" in text and "t.me/DBUCentral" in text


def test_contribution_rules():
    from datetime import datetime, timedelta

    from app.services import contrib_rules as K

    assert K.check_file("a.pdf", 1000) is None and K.check_file("A.PPTX", 1000) is None
    assert K.check_file("a.exe", 1000) == "ext" and K.check_file("a.pdf", K.MAX_BYTES + 1) == "size"
    assert K.check_file("a.pdf", None) == "size"

    now = datetime(2026, 10, 20, 12, 0)
    day = lambda n: now - timedelta(days=n)  # noqa: E731
    assert K.blocked_until([day(1), day(2)], now) is None  # two rejections: still fine
    assert K.blocked_until([day(1), day(2), day(3)], now) == day(1) + timedelta(days=7)
    assert K.blocked_until([day(20), day(21), day(22)], now) is None  # too old to count
    assert K.blocked_until([day(9), day(10), day(11)], now) is None  # block (7 days after day 9) already over

    ok = dict(active_days=3, min_days=2, pending=0, pending_limit=5, today=0, daily_limit=8, blocked=None)
    assert K.eligibility(**ok) is None
    assert K.eligibility(**{**ok, "active_days": 1}) == "new"
    assert K.eligibility(**{**ok, "pending": 5}) == "pending"
    assert K.eligibility(**{**ok, "today": 8}) == "daily"
    assert K.eligibility(**{**ok, "blocked": now}) == "blocked"

    assert [K.badge_for(n) for n in (0, 1, 4, 5, 14, 15)] == [None, "bronze", "bronze", "silver", "silver", "gold"]
    assert K.next_badge(0) == (1, "bronze") and K.next_badge(3) == (2, "silver") and K.next_badge(15) is None
    assert K.reward(rewarded_this_week=0, course_files_before=0, xp=20, bonus=5, weekly_cap=5) == (20, 5)
    assert K.reward(rewarded_this_week=0, course_files_before=9, xp=20, bonus=5, weekly_cap=5) == (20, 0)
    assert K.reward(rewarded_this_week=5, course_files_before=0, xp=20, bonus=5, weekly_cap=5) == (0, 0)


def test_qr_png_really_scans():
    """Decode the PNG with OpenCV's QR reader: proves the code is valid, not just pretty."""
    try:
        import cv2
        import numpy as np
    except ImportError:
        return
    from app.services.qr import qr_png

    link = "https://t.me/DBUFreshBot?start=dorm5"
    for transparent in (False, True):
        png = qr_png(link, transparent=transparent)
        img = cv2.imdecode(np.frombuffer(png, np.uint8), cv2.IMREAD_UNCHANGED)
        if img.shape[2] == 4:  # transparent: put it on white like a printed page
            alpha = img[:, :, 3:4] / 255.0
            img = (img[:, :, :3] * alpha + 255 * (1 - alpha)).astype("uint8")
        text, _, _ = cv2.QRCodeDetector().detectAndDecode(img)
        assert text == link, (transparent, text)


def test_members_who_never_saw_the_gate_still_count():
    """Regression: gate_passed used to be set only by the 'I joined' button, so students who were already
    channel members got no XP, no streak and no referral credit."""
    root = Path(__file__).resolve().parents[1]
    middleware = (root / "app/bot/middlewares/gate.py").read_text(encoding="utf-8")
    assert "onboarding.mark_passed" in middleware
    assert "onboarding.mark_passed" in (root / "app/bot/handlers/gate.py").read_text(encoding="utf-8")
    assert "SET gate_passed = TRUE" in (root / "app/db/migrate.py").read_text(encoding="utf-8")


def _snapshot():
    from datetime import datetime, timedelta, timezone

    from app.services import analytics as A

    ET = A.ET
    now = datetime(2026, 10, 22, 12, 0, tzinfo=timezone.utc)  # 15:00 Ethiopian time, 22 Oct
    today = now.astimezone(ET).date()

    def user(i, days_ago, source, verified, onboarded, active=0, missing=None):
        created = now - timedelta(days=days_ago, hours=1)
        return A.UserRow(id=i, created=created, joined=created.astimezone(ET).date(), source=source, verified=verified,
                         onboarded=onboarded, active_days=active, missing=missing)

    users = (
        [user(i, 5, "organic", True, True, 3) for i in range(1, 7)]  # 6 joined 5 days ago, all fine
        + [user(i, 5, "organic", False, False, 0, "dbucentralforum") for i in range(7, 13)]  # 6 stuck: forum missing
        + [user(i, 1, "ad1", True, True, 1) for i in range(13, 19)]  # 6 joined yesterday
        + [user(i, 0, "ad1", True, False) for i in range(19, 21)]  # 2 verified today, did not finish setup
    )
    d = lambda n: today - timedelta(days=n)  # noqa: E731
    checkins = {i: {d(5), d(4)} for i in range(1, 4)}  # 3 of the first 12 came back the next day
    checkins.update({i: {d(5)} for i in range(4, 7)})
    checkins.update({i: {d(1), d(0)} for i in range(13, 17)})
    events = (
        [A.EventRow("search", "physics", 3, d(1), 1)] * 3
        + [A.EventRow("search", "chemistry module", 0, d(1), 2)] * 4
        + [A.EventRow("inline", "chemistry module", 0, d(0), 3)]
        + [A.EventRow("search", "x", 0, d(0), 4)]  # too short to count
        + [A.EventRow("download", "1", 1, d(0), 1), A.EventRow("download", "1", 1, d(2), 2)]
    )
    materials = [A.MaterialRow(1, 10, "Physics module", 7, "ok", "note"), A.MaterialRow(2, 10, "Physics exam", 2, "ok", "exam"),
                 A.MaterialRow(3, 11, "Psych", 1, "ok", "note"), A.MaterialRow(4, 12, "Hidden", 9, "pending", "note")]
    courses = {10: "General Physics", 11: "General Psychology", 12: "General Chemistry", 13: "Entrepreneurship"}
    subs = [A.SubRow("approved", 5, now - timedelta(hours=30), now - timedelta(hours=28)),
            A.SubRow("rejected", 6, now - timedelta(hours=10), now - timedelta(hours=6)),
            A.SubRow("pending", 7, now - timedelta(hours=40), None)]
    posts = [A.PostRow(1, now - timedelta(days=1, hours=3), "Big news t.me/Bot?start=ad1", ["ad1"], 12),
             A.PostRow(2, now - timedelta(days=3), "meme", [], None)]
    return A.Snapshot(now=now, users=users, checkins=checkins, events=events, materials=materials, courses=courses, subs=subs, posts=posts)


def test_analytics_numbers():
    from datetime import timedelta

    from app.services import analytics as A

    s = _snapshot()
    f = A.funnel(s.users)
    assert f == dict(joined=20, verified=14, onboarded=12, active3=6)
    assert A.biggest_leak(f) == ("joined", "in channel + forum", 6, 30)
    table = {r["source"]: r for r in A.source_table(s.users)}
    assert table["organic"]["joined"] == 12 and table["organic"]["setup_pct"] == 50
    assert table["ad1"]["joined"] == 8 and table["ad1"]["onboarded"] == 6 and table["ad1"]["setup_pct"] == 75
    assert A.stuck_by_missing(s.users) == [("dbucentralforum", 6)]
    days = A.per_day(s.users, s.today, 7)
    assert len(days) == 7 and days[-1] == (s.today, 2) and days[-2] == (s.today - timedelta(days=1), 6)
    assert sum(n for _, n in days) == 20


def test_analytics_retention_and_activity():
    from datetime import timedelta

    from app.services import analytics as A

    s = _snapshot()
    ret = A.retention(s.users, s.checkins, s.today)
    old = next(r for r in ret["rows"] if r["day"] == s.today - timedelta(days=5))
    assert old["size"] == 12 and old["d1"] == 25  # 3 of the 12 were active the next day
    assert old["d7"] is None  # day 7 has not come yet
    yesterday = next(r for r in ret["rows"] if r["day"] == s.today - timedelta(days=1))
    assert yesterday["d1"] is None  # "tomorrow" is today, which is not over: never counted, never guessed
    assert ret["came_back"] == 7 and ret["eligible"] == 18
    w = A.dau_wau(s.checkins, s.today)
    assert w["wau"] == 10 and dict(w["dau"])[s.today] == 4
    # before activity tracking began a cohort must be n/a, not 0%
    ret2 = A.retention(s.users, {1: {s.today}}, s.today)
    assert all(r["d1"] is None for r in ret2["rows"] if r["day"] < s.today - timedelta(days=2))


def test_analytics_library_searches_contribs_channel():
    from app.services import analytics as A

    s = _snapshot()
    lib = A.library(s)
    assert lib["files"] == 3 and lib["empty"] == ["General Chemistry", "Entrepreneurship"]  # pending files do not count
    assert lib["sparse"] == ["General Physics", "General Psychology"]  # 2 and 1 files: both under 3
    assert lib["missed"][0] == ("chemistry module", 5)  # 4 text + 1 inline; the 1-letter search is ignored
    assert lib["searches"] == 8 and lib["missed_total"] == 5
    assert lib["downloads_week"] == 2 and lib["downloaders_week"] == 2 and lib["top_files"][0].title == "Physics module"
    c = A.contributions(s)
    assert (c["pending"], c["approved"], c["rejected"], c["approval"]) == (1, 1, 1, 50)
    assert c["oldest_pending_hours"] == 40.0 and c["avg_hours"] == 3.0
    ch = A.channel_posts(s)
    assert ch[0]["message_id"] == 1 and ch[0]["tagged"] == 8  # the 8 students whose source is the post's tag "ad1"
    assert A.extract_tags("go t.me/B?start=ad2 or t.me/B?start=123456 and ?start=Dorm5") == ["ad2", "dorm5"]


def test_analytics_insights_and_screens():
    import re

    from app.services import analytics as A

    s = _snapshot()
    ins = A.insights(s)
    titles = [i.title for i in ins]
    assert "Students stop at the join step" in titles  # 6 of 20 = 30% (rule: 15% or more)
    assert "Students search for things you do not have" in titles
    assert "Many courses have no file" in titles  # 2 of 4 courses are empty
    assert "Student files are waiting" in titles  # the oldest pending file is 40 h old
    first = ins[0]
    assert first.priority == 1 and "30%" in first.evidence and "dbucentralforum" in first.evidence
    screens = (A.overview, A.section_growth, A.section_retention, A.section_library, A.section_contrib,
               A.section_channel, A.section_insights)
    for render in screens:
        text = render(s)
        assert 20 < len(text) < 4000, render.__name__
        for tag in ("b", "i"):
            assert len(re.findall(rf"<{tag}>", text)) == len(re.findall(rf"</{tag}>", text)), render.__name__
    empty = A.Snapshot(now=s.now, users=[], checkins={}, events=[], materials=[], courses={}, subs=[], posts=[])
    for render in screens:  # an empty project must not crash anything
        assert render(empty)


REAL_NAMES = [
 [
  "psyscho 2014 mid exam collection.pdf",
  "General Psychology",
  "exam"
 ],
 [
  "Physics mid exam 2014.pdf",
  "General Physics",
  "exam"
 ],
 [
  "General Physics Final Exam.pdf",
  "General Physics",
  "exam"
 ],
 [
  "Math 1003 mid.pdf",
  "Mathematics for Natural Sciences",
  "exam"
 ],
 [
  "Chem Exam 2013 EC.pdf",
  "General Chemistry",
  "exam"
 ],
 [
  "Anthropology Exam Questions.pdf",
  "Social Anthropology",
  "exam"
 ],
 [
  "Logic and critical thinking questions.pdf",
  "Critical Thinking",
  "exam"
 ],
 [
  "ኢኮኖሚክስ ጥያቄ.pdf",
  "Introduction to Economics",
  "exam"
 ],
 [
  "ፊዚክስ ፈተና 2014.pdf",
  "General Physics",
  "exam"
 ],
 [
  "Civics model exam.pdf",
  "Moral and Civic Education",
  "exam"
 ],
 [
  "Past paper English 2013.pdf",
  null,
  "exam"
 ],
 [
  "geography quiz 1.pdf",
  "Geography of Ethiopia and the Horn",
  "exam"
 ],
 [
  "Psychology test 1 answer key.pdf",
  "General Psychology",
  "exam"
 ],
 [
  "Mid-term exam for Entrepreneurship.pdf",
  "Entrepreneurship",
  "exam"
 ],
 [
  "Psychology note.pdf",
  "General Psychology",
  "note"
 ],
 [
  "Chapter 1-3 summary psychology.pdf",
  "General Psychology",
  "note"
 ],
 [
  "ፊዚክስ ማስታወሻ.pdf",
  "General Physics",
  "note"
 ],
 [
  "Economics handout.pdf",
  "Introduction to Economics",
  "note"
 ],
 [
  "Entrepreneurship chapter 2 notes.pdf",
  "Entrepreneurship",
  "note"
 ],
 [
  "Anthropology module.pdf",
  "Social Anthropology",
  "note"
 ],
 [
  "Physics short note for mid.pdf",
  "General Physics",
  "note"
 ],
 [
  "Lecture 3 physics.pdf",
  "General Physics",
  "slide"
 ],
 [
  "Chapter 1 Entrepreneurship.pptx",
  "Entrepreneurship",
  "slide"
 ],
 [
  "ስላይድ ኢኮኖሚክስ.pdf",
  "Introduction to Economics",
  "slide"
 ],
 [
  "General Chemistry textbook.pdf",
  "General Chemistry",
  "book"
 ],
 [
  "ባዮሎጂ መጽሐፍ.pdf",
  "General Biology",
  "book"
 ],
 [
  "document.pdf",
  null,
  "note"
 ],
 [
  "IMG_2034.pdf",
  null,
  "note"
 ],
 [
  "1698765432.pdf",
  null,
  "note"
 ],
 [
  "PSYCH_1007_Module(final).pdf",
  "General Psychology",
  "note"
 ],
 [
  "Psys1007 worksheet 2.pdf",
  "General Psychology",
  "note"
 ],
 [
  "physics lab report.pdf",
  "General Physics",
  "note"
 ],
 [
  "economics assignment 1.pdf",
  "Introduction to Economics",
  "note"
 ],
 [
  "Exm 2014 geography.pdf",
  "Geography of Ethiopia and the Horn",
  "exam"
 ],
 [
  "Psychology Exam collection (2013-2015).pdf",
  "General Psychology",
  "exam"
 ],
 [
  "philosophy notes.pdf",
  null,
  "note"
 ],
 [
  "geometry worked examples.pdf",
  null,
  "note"
 ],
 [
  "notebook.pdf",
  null,
  "note"
 ],
 [
  "finalize registration.pdf",
  null,
  "note"
 ],
 [
  "midnight reading.pdf",
  null,
  "note"
 ],
 [
  "latest news.pdf",
  null,
  "note"
 ]
]


def test_realistic_student_file_names():
    """Amharic names, typos, mid/final/collection, chapter..., plus words that must NOT be misread."""
    import os

    wrong = []
    for name, want_course, want_kind in REAL_NAMES:
        best, confident, _ = C.decide(C.rank_courses(COURSES, [(name, "name")]))
        course = best.name if (best and confident) else None
        kind, _why = C.detect_kind(name, "", "", None, None, os.path.splitext(name)[1].lower())
        if (course, kind) != (want_course, want_kind):
            wrong.append((name, course, kind))
    assert not wrong, wrong


if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
    print("all logic tests passed")