"""Tests for the viewer. Run: ../.venv/bin/python test_viewer.py

The category tests are offline and always run -- they pin the ai_search split,
which is the thing most likely to be broken by an edit to crawlers.py.

The app tests use Flask's test client against a fake S3, so they need no
network and no credentials.
"""

import os
import sys

os.environ.setdefault("VIEWER_USER", "testuser")
os.environ.setdefault("VIEWER_PASSWORD", "testpass")

import categories as cat  # noqa: E402
import s3data  # noqa: E402
import sites as sites_mod  # noqa: E402

FAILURES = []


def check(name, actual, expected):
    if actual != expected:
        FAILURES.append("%s\n    expected: %r\n    actual:   %r" % (name, expected, actual))


# --- the confirmed grouping ------------------------------------------------
# Signed off 2026-07-16. `search` means a conventional search engine only;
# AI companies that index pages live in ai_search and must stay visible.
for bot in ("OAI-SearchBot", "PerplexityBot", "YouBot", "Claude-SearchBot"):
    check("%s is ai_search" % bot, cat.category_for(bot), "ai_search")
    check("%s counts as AI" % bot, cat.is_ai(cat.category_for(bot)), True)

for bot in ("Bingbot", "Amazonbot", "Applebot"):
    check("%s is search" % bot, cat.category_for(bot), "search")
    check("%s is not AI" % bot, cat.is_ai(cat.category_for(bot)), False)

# Flagged during review as easy to get wrong, so pinned:
check("Applebot-Extended stays AI", cat.category_for("Applebot-Extended"), "ai_training")
check("DuckAssistBot stays assistant", cat.category_for("DuckAssistBot"), "ai_assistant")
check("GoogleOther stays training", cat.category_for("GoogleOther"), "ai_training")
check("Timpibot stays training", cat.category_for("Timpibot"), "ai_training")
check("GPTBot unchanged", cat.category_for("GPTBot"), "ai_training")
check("ChatGPT-User unchanged", cat.category_for("ChatGPT-User"), "ai_assistant")

# The stored category is overridden, not trusted: every event in S3 predates
# the split and says "search" for OAI-SearchBot.
check("stored category ignored", cat.category_for("OAI-SearchBot", stored="search"), "ai_search")
# An unknown bot must never be silently counted as AI.
check("unknown bot -> search", cat.category_for("Wat-2027", stored=None), "search")
check("unknown bot honours stored", cat.category_for("Wat-2027", stored="ai_training"), "ai_training")
check("exactly 3 AI groups", set(cat.AI_GROUPS), {"ai_training", "ai_assistant", "ai_search"})


# --- fake S3 ---------------------------------------------------------------
# Shaped like the real bucket, because the site identity depends on that shape:
# the shard stem carries the WP Engine install and the partition carries the
# vhost, and they are not the same string. Install "alpha" serves a.com AND its
# WP Engine holding domain a.wpengine.com -- one site, two partitions.
EVENTS = {
    "events/date=2026-07-08/site=a.com/20260709-0017-alpha.jsonl": [
        {"timestamp": "2026-07-08T01:00:00Z", "site": "a.com", "ai_platform": "GPTBot",
         "category": "ai_training", "url": "/dog-bite/", "status_code": 200},
        # stored as "search" on purpose: the viewer must regroup it to ai_search
        {"timestamp": "2026-07-08T02:00:00Z", "site": "a.com", "ai_platform": "OAI-SearchBot",
         "category": "search", "url": "/cats/", "status_code": 200},
    ],
    "events/date=2026-07-08/site=a.wpengine.com/20260709-0017-alpha.jsonl": [
        {"timestamp": "2026-07-08T03:00:00Z", "site": "a.wpengine.com", "ai_platform": "Bingbot",
         "category": "search", "url": "/dog-bite/", "status_code": 404},
    ],
    "events/date=2026-07-08/site=b.com/20260709-0017-bravo.jsonl": [
        {"timestamp": "2026-07-08T04:00:00Z", "site": "b.com", "ai_platform": "ClaudeBot",
         "category": "ai_training", "url": "/x/", "status_code": 200},
    ],
}
# "charlie" has a log file and no events -- the zero-hit case. It exists only
# here, because an empty log produces no vhost and so no event partition.
MANIFEST = {
    "processed": {
        "logs/nginx/20260709-0017-alpha.apachestyle.log.gz": {"hits": 3, "dates": ["2026-07-08"]},
        "logs/nginx/20260709-0017-bravo.apachestyle.log.gz": {"hits": 1, "dates": ["2026-07-08"]},
        "logs/nginx/20260709-0017-charlie.apachestyle.log.gz": {"hits": 0, "dates": []},
    }
}
# Base events are all real pages, so content_hits == hits and by_platform_content
# mirrors by_platform. The content test below rewrites these once it adds the
# non-content hits, so the summary keeps agreeing with the events.
SUMMARY = {
    "date": "2026-07-08",
    "totals": {"hits": 4, "content_hits": 4, "bytes": 40, "sites": 3, "platforms": 4},
    "by_platform": {
        "GPTBot": {"hits": 1, "content_hits": 1, "bytes": 10, "category": "ai_training"},
        "OAI-SearchBot": {"hits": 1, "content_hits": 1, "bytes": 10, "category": "search"},
        "Bingbot": {"hits": 1, "content_hits": 1, "bytes": 10, "category": "search"},
        "ClaudeBot": {"hits": 1, "content_hits": 1, "bytes": 10, "category": "ai_training"},
    },
    "by_site": {
        "a.com": {"hits": 2, "content_hits": 2, "bytes": 20,
                  "by_platform": {"GPTBot": 1, "OAI-SearchBot": 1},
                  "by_platform_content": {"GPTBot": 1, "OAI-SearchBot": 1}},
        "a.wpengine.com": {"hits": 1, "content_hits": 1, "bytes": 10,
                           "by_platform": {"Bingbot": 1},
                           "by_platform_content": {"Bingbot": 1}},
        "b.com": {"hits": 1, "content_hits": 1, "bytes": 10,
                  "by_platform": {"ClaudeBot": 1},
                  "by_platform_content": {"ClaudeBot": 1}},
    },
    "by_url": {"/dog-bite/": 2},
}


def fake_get_object(Bucket, Key):
    import io
    import json as _json
    if Key.startswith("summary/"):
        return {"Body": io.BytesIO(_json.dumps(SUMMARY).encode())}
    if Key == "manifest/processed.json":
        return {"Body": io.BytesIO(_json.dumps(MANIFEST).encode())}
    lines = "\n".join(_json.dumps(e) for e in EVENTS[Key])
    return {"Body": io.BytesIO(lines.encode())}


class FakeClient(object):
    def get_object(self, Bucket, Key):
        return fake_get_object(Bucket, Key)

    def list_objects_v2(self, **kw):
        return {}

    def get_paginator(self, _name):
        class P(object):
            def paginate(self, Bucket, Prefix, Delimiter=None):
                if Prefix.startswith("summary/"):
                    yield {"Contents": [{"Key": "summary/date=2026-07-08.json"}]}
                    return
                keys = [k for k in EVENTS if k.startswith(Prefix)]
                if Delimiter:
                    pres = sorted({k[:k.index("/", len(Prefix)) + 1] for k in keys
                                   if "/" in k[len(Prefix):]})
                    yield {"CommonPrefixes": [{"Prefix": p} for p in pres]}
                else:
                    yield {"Contents": [{"Key": k} for k in keys]}
        return P()


s3data.client = lambda: FakeClient()
s3data._ttl_cache.clear()
s3data._summary_cache.clear()
s3data._shard_cache.clear()

import app as A  # noqa: E402

# --- data layer ------------------------------------------------------------
check("available_dates", s3data.available_dates(), ["2026-07-08"])
check("all_sites", s3data.all_sites(), ["a.com", "a.wpengine.com", "b.com"])
check("sites_for(day)", s3data.sites_for(["2026-07-08"]), ["a.com", "a.wpengine.com", "b.com"])
check("sites_for(none)", s3data.sites_for([]), [])
check("dates_in_range clamps to real days",
      s3data.dates_in_range("2026-07-01", "2026-07-31"), ["2026-07-08"])
check("dates_in_range outside -> none", s3data.dates_in_range("2020-01-01", "2020-01-02"), [])
check("shard_keys, all sites", sorted(s3data.shard_keys(["2026-07-08"])), sorted(EVENTS))
check("shard_keys, one install -> its one shard",
      s3data.shard_keys(["2026-07-08"], ["bravo"]),
      ["events/date=2026-07-08/site=b.com/20260709-0017-bravo.jsonl"])
# The point of keying on the install: one filter picks up every vhost it serves,
# including the WP Engine holding domain.
check("shard_keys, install spanning two vhosts -> both shards",
      sorted(s3data.shard_keys(["2026-07-08"], ["alpha"])),
      ["events/date=2026-07-08/site=a.com/20260709-0017-alpha.jsonl",
       "events/date=2026-07-08/site=a.wpengine.com/20260709-0017-alpha.jsonl"])
check("shard_keys, unknown install -> nothing", s3data.shard_keys(["2026-07-08"], ["nope"]), [])
check("summary drops by_url", "by_url" in s3data.summary("2026-07-08"), False)

# AI-only must exclude Bingbot but keep OAI-SearchBot despite its stored category
rows, total, by_group, _ = s3data.query(["2026-07-08"], groups=list(cat.AI_GROUPS))
check("AI-only total", total, 3)
check("AI-only groups", by_group, {"ai_training": 2, "ai_search": 1})
check("AI-only excludes Bingbot", [r for r in rows if r[s3data.PLATFORM] == "Bingbot"], [])

rows, total, by_group, _ = s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS))
check("with search, total", total, 4)
check("with search, groups", by_group, {"ai_training": 2, "ai_search": 1, "search": 1})

_, total, _, _ = s3data.query(["2026-07-08"], sites=["bravo"], groups=list(cat.AI_GROUPS))
check("site filter", total, 1)
# Links made before site= meant the install still carry a vhost. They must keep
# working rather than render a confident "0 hits".
with A.app.test_request_context("/?site=b.com"):
    check("an old vhost link resolves to its install", A._filters()["site"], "bravo")
with A.app.test_request_context("/?site=a.wpengine.com"):
    check("an old holding-domain link resolves too", A._filters()["site"], "alpha")
with A.app.test_request_context("/?site=bravo"):
    check("an install passes through unchanged", A._filters()["site"], "bravo")
with A.app.test_request_context("/?site=nosuchthing"):
    check("an unknown site is not widened to everything",
          A._filters()["site"], "nosuchthing")
# Filtering by install must gather both of alpha's vhosts (Bingbot is dropped by
# the AI-only group filter, not by the site filter).
_, total, _, _ = s3data.query(["2026-07-08"], sites=["alpha"], groups=list(cat.ALL_GROUPS))
check("site filter spans an install's vhosts", total, 3)
_, total, _, _ = s3data.query(["2026-07-08"], platforms=["OAI-SearchBot"], groups=list(cat.AI_GROUPS))
check("platform filter on an ai_search bot", total, 1)
_, total, _, _ = s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS), url_query="dog-bite")
check("url search", total, 2)
_, total, _, _ = s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS), url_query="DOG-BITE")
check("url search is case-insensitive", total, 2)
_, total, _, _ = s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS), url_query="nope")
check("url search, no match", total, 0)

rows, _, _, _ = s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS))
check("newest first", [r[s3data.TS] for r in rows],
      sorted([r[s3data.TS] for r in rows], reverse=True))

# --- the shard-fetch pool is shared and bounded ----------------------------
# One pool per process, not one per request: N concurrent requests must not run
# N*_WORKERS threads against a connection pool sized for _WORKERS. That is what
# collapsed the server under concurrent "View Hits".
check("the fetch pool is a singleton", s3data._executor() is s3data._executor(), True)
check("the fetch pool is bounded to _WORKERS",
      s3data._executor()._max_workers, s3data._WORKERS)
# Live thread count never exceeds the pool, however many queries run at once, and
# every one still returns the right answer.
import threading as _threading  # noqa: E402
_peak = [0]
_stop = _threading.Event()
def _watch():
    while not _stop.is_set():
        _peak[0] = max(_peak[0], sum(1 for t in _threading.enumerate()
                                     if t.name.startswith("s3fetch")))
_results, _errs = [], []
def _scan():
    try:
        _results.append(s3data.query(["2026-07-08"], groups=list(cat.ALL_GROUPS))[1])
    except Exception as exc:                       # noqa: BLE001
        _errs.append(repr(exc))
_w = _threading.Thread(target=_watch, daemon=True); _w.start()
_ts = [_threading.Thread(target=_scan) for _ in range(8)]
for _t in _ts: _t.start()
for _t in _ts: _t.join()
_stop.set()
check("concurrent queries all succeed", _errs, [])
check("concurrent queries agree", set(_results), {4})
check("live fetch threads stay within the pool", _peak[0] <= s3data._WORKERS, True)

# --- auth ------------------------------------------------------------------
client = A.app.test_client()
check("overview needs login", client.get("/").status_code, 302)
check("detail needs login", client.get("/detail").status_code, 302)
check("wrong password rejected",
      b"Wrong username" in client.post("/login", data={"username": "testuser",
                                                       "password": "nope"}).data, True)
r = client.post("/login", data={"username": "testuser", "password": "testpass"})
check("correct password -> redirect", r.status_code, 302)
check("logged in overview -> 200", client.get("/").status_code, 200)

# open redirect: ?next= must not send us off-site
r = client.post("/login?next=https://evil.example/x",
                data={"username": "testuser", "password": "testpass"})
check("no open redirect", "evil.example" in (r.headers.get("Location") or ""), False)

# --- pages -----------------------------------------------------------------
def platform_table(html):
    """Just the 'By AI platform' rows. The whole page mentions Bingbot in the
    checkbox's own tooltip ('Include search engines (Bing, ...)'), which is not
    data -- searching the full page would give a false positive."""
    if "By AI platform" not in html:
        return ""
    return html.split("By AI platform")[1].split("</table>")[0]


body = platform_table(client.get("/?preset=custom&start=2026-07-08&end=2026-07-08").data.decode())
check("overview hides Bingbot by default", "Bingbot" in body, False)
check("overview shows OAI-SearchBot by default", "OAI-SearchBot" in body, True)
check("overview labels it AI search", "AI search" in body, True)

body = platform_table(
    client.get("/?preset=custom&start=2026-07-08&end=2026-07-08&include_search=1").data.decode())
check("overview shows Bingbot when asked", "Bingbot" in body, True)

def table_body(html):
    """Just the data rows. The filter dropdown lists every bot by design, so
    searching the whole page would find Bingbot even when it is filtered out."""
    if "<tbody>" not in html:
        return ""
    return html.split("<tbody>")[-1].split("</tbody>")[0]


rows_html = table_body(client.get("/detail?start=2026-07-08&end=2026-07-08").data.decode())
check("detail hides Bingbot by default", "Bingbot" in rows_html, False)
check("detail shows OAI-SearchBot by default", "OAI-SearchBot" in rows_html, True)
check("detail shows 3 AI rows", rows_html.count("<tr>"), 3)

rows_html = table_body(
    client.get("/detail?start=2026-07-08&end=2026-07-08&include_search=1").data.decode())
check("detail shows Bingbot when asked", "Bingbot" in rows_html, True)
check("detail shows all 4 rows with search", rows_html.count("<tr>"), 4)

# The dropdown must still offer every bot, including the hidden ones.
page = client.get("/detail?start=2026-07-08&end=2026-07-08").data.decode()
check("filter dropdown still offers Bingbot", 'value="Bingbot"' in page, True)


# --- the totals must reconcile, on every view -------------------------------
# "Every hit is one hit and belongs to exactly one category", so the category
# boxes must always sum to the Hits box. Hits is derived from the per-group
# counts to make this structural, and this pins it.
def stat_boxes(html):
    """(hits, {group: n}) as actually rendered. Reads the Hits big box and the
    category mini boxes -- i.e. exactly the numbers a person adds up on screen."""
    import re
    bigs = re.findall(
        r'<div class="big \w+">\s*<div class="n">([\d,]+)</div>\s*<div class="l">(\w+)', html)
    hits = next((int(v.replace(",", "")) for v, lab in bigs if lab == "Hits"), None)
    minis = re.findall(r'<div class="mini ([\w]+)">\s*<div class="n">([\d,]+)</div>', html)
    return hits, {g: int(v.replace(",", "")) for g, v in minis}


for qs in ("", "&include_search=1"):
    page = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08" + qs).data.decode()
    hits, cats = stat_boxes(page)
    mode = "with search" if qs else "AI only"
    check("overview reconciles (%s)" % mode, sum(cats.values()), hits)
    check("nothing stranded (%s)" % mode, "did not fall into any category" in page, False)

# Every bot in the parser's registry must resolve to exactly one display group.
# This is the check that would have named a bot whose hits went missing.
seen_groups = [cat.category_for(label) for label, _ in cat.AI_CRAWLERS.values()]
check("every bot resolves to a display group",
      [g for g in seen_groups if g not in cat.ALL_GROUPS], [])
check("every bot resolves to exactly one group",
      all(isinstance(g, str) for g in seen_groups), True)

# The guard itself: a bot whose category has no box must be caught loudly at
# startup, not silently dropped from the totals.
_saved = dict(cat.AI_CRAWLERS)
try:
    cat.AI_CRAWLERS["RogueBot"] = ("RogueBot", "ai_agent")  # no box for this
    try:
        cat._check_registry()
        check("rogue category is caught", "did not raise", "UncategorisedBot")
    except cat.UncategorisedBot as exc:
        check("rogue category is caught", "RogueBot" in str(exc), True)
        check("...and the message names the bad category", "ai_agent" in str(exc), True)
finally:
    cat.AI_CRAWLERS.clear()
    cat.AI_CRAWLERS.update(_saved)
check("registry is clean as shipped", cat._check_registry(), None)

# and the same for the detail page's own group counts
for qs in ("", "&include_search=1"):
    rows, total, by_group, _ = s3data.query(
        ["2026-07-08"],
        groups=list(cat.ALL_GROUPS if qs else cat.AI_GROUPS))
    check("detail reconciles (%s)" % ("with search" if qs else "AI only"),
          sum(by_group.values()), total)

# The bug this replaced: the site rollup classified a bot WITHOUT the stored
# category, so an unknown bot fell into `search` there while the stat boxes put
# it in `ai_training` -- one hit counted under two categories.
check("classification is stable with a stored hint",
      cat.category_for("Unknown-2027", "ai_training"), "ai_training")
check("...and that is what the fallback must agree with",
      cat.category_for("Unknown-2027", None), "search")

# Layout: two big boxes (Sites, Hits), then 4 small boxes -- or 5 with search.
import re as _re  # noqa: E402


def layout(html):
    # The Sites label carries a "/N" total span ahead of the word; Hits does not.
    bigs = _re.findall(r'<div class="big \w+">\s*<div class="n">[\d,]+</div>\s*'
                       r'<div class="l">(?:<span class="of">[^<]*</span>)?\s*(\w+)', html)
    minis = _re.findall(r'<div class="mini ?[\w]*">\s*<div class="n">[\d,]+</div>\s*'
                        r'<div class="l">([^<\n]*)', html)
    grid = _re.search(r'<div class="grid( five)?"', html)
    return bigs, [m.strip() for m in minis], bool(grid.group(1)) if grid else False


page = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08").data.decode()
bigs, minis, five = layout(page)
check("two big boxes: Sites then Hits", bigs, ["Sites", "Hits"])
check("four small boxes when AI-only", minis,
      ["Bots", "AI training", "AI assistant", "AI search"])
check("2x2 grid when four", five, False)
_, cats = stat_boxes(page)
check("no Search engines box when AI-only", "search" in cats, False)

page = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08&include_search=1").data.decode()
bigs, minis, five = layout(page)
check("still two big boxes with search", bigs, ["Sites", "Hits"])
check("five small boxes with search", minis,
      ["Bots", "AI training", "AI assistant", "AI search", "Search engines"])
check("3+2 grid when five", five, True)
_, cats = stat_boxes(page)
check("Search engines box present when included", "search" in cats, True)

# Big-box backgrounds: brand blue behind Sites, line grey behind Hits.
check("sites box is brand blue",
      "background:var(--brand)" in page.split(".big.sites {")[1].split("}")[0], True)
check("hits box is line grey",
      "background:var(--line)" in page.split(".big.hits  {")[1].split("}")[0], True)
# The numbers stay neutral ink on both, as asked earlier.
check("big number is ink, not brand blue",
      "--brand-ink" in page.split(".big .n")[1].split("}")[0], False)
check("big number is ink", "color:var(--ink)" in page.split(".big .n")[1].split("}")[0], True)

# Info icon: solid black circle, white lowercase i.
css = page.split(".info {")[1].split("}")[0]
check("info icon is a black circle", "background:#000" in css, True)
check("info icon text is white", "color:#fff" in css, True)
check("info icon is round", "border-radius:50%" in css, True)
# The tooltip must wrap: the label sets nowrap and the pseudo-element inherits it.
tip = page.split(".info::after {")[1].split("}")[0]
check("tooltip forces wrapping", "white-space:normal" in tip, True)
check("tooltip has a bounded width", "max-width:260px" in tip, True)
check("tooltip has padding", "padding:9px 11px" in tip, True)

# Bytes are gone from the UI entirely.
check("no Bytes served stat", "Bytes served" in page, False)
check("no Bytes column", ">Bytes<" in page, False)

# --- date presets -----------------------------------------------------------
from datetime import date as _date  # noqa: E402

anchor = _date(2026, 7, 16)  # a Thursday
# The default landing preset: a single day (the newest), so a cold first paint
# scans ~180 shards, not a week's ~3,250.
check("latest_day is a single day", A._preset_range("latest_day", anchor), (anchor, anchor))
check("latest_day is the default", A.DEFAULT_PRESET, "latest_day")
check("this_week starts Monday", A._preset_range("this_week", anchor)[0], _date(2026, 7, 13))
check("this_week ends at anchor", A._preset_range("this_week", anchor)[1], anchor)
check("this_month starts on the 1st", A._preset_range("this_month", anchor)[0], _date(2026, 7, 1))
check("last_month is all of June", A._preset_range("last_month", anchor),
      (_date(2026, 6, 1), _date(2026, 6, 30)))
check("last_90 spans 90 days inclusive", A._preset_range("last_90", anchor)[0], _date(2026, 4, 18))
# January must roll back to December of the previous year.
check("last_month crosses the year", A._preset_range("last_month", _date(2026, 1, 15)),
      (_date(2025, 12, 1), _date(2025, 12, 31)))

with A.app.test_request_context("/?preset=this_week"):
    check("preset wins over dates", A._selected_range()[0], "this_week")
with A.app.test_request_context("/?start=2026-07-08&end=2026-07-08"):
    check("explicit dates -> custom", A._selected_range()[0], "custom")
with A.app.test_request_context("/"):
    check("default preset", A._selected_range()[0], A.DEFAULT_PRESET)
with A.app.test_request_context("/?preset=nonsense"):
    check("bogus preset falls back", A._selected_range()[0], A.DEFAULT_PRESET)
# Presets anchor on the newest day we hold, not the wall clock, or the page
# would be empty every morning until the nightly parse lands.
check("anchor is the newest data day", A._today(), _date(2026, 7, 8))

# Custom with no dates defaults to a 7-day window (6 days back, inclusive),
# ending on the newest day with data -- not the single day it used to be. The
# fake bucket only holds one day, so the start clamps up to it.
with A.app.test_request_context("/?preset=custom"):
    _, s, e, _ = A._selected_range()
    check("custom default ends on newest data day", e, "2026-07-08")
    check("custom default clamps start to earliest data day", s, "2026-07-08")
# The clamp only bites at the edge of the data; away from it the window is a
# full 7 days. Verified directly on _preset math via a wider synthetic range.
_wide = sorted(["2026-07-%02d" % d for d in range(1, 21)], reverse=True)
_saved_dates = A.s3data.available_dates
A.s3data.available_dates = lambda: _wide
A.s3data._ttl_cache.clear()
try:
    with A.app.test_request_context("/?preset=custom"):
        _, s, e, _ = A._selected_range()
        check("custom default is 7 days wide", (s, e), ("2026-07-14", "2026-07-20"))
    # An explicit start is always honoured over the default.
    with A.app.test_request_context("/?preset=custom&start=2026-07-05&end=2026-07-20"):
        check("explicit start beats the 7-day default", A._selected_range()[1], "2026-07-05")
finally:
    A.s3data.available_dates = _saved_dates
    A.s3data._ttl_cache.clear()

# --- export -----------------------------------------------------------------
r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08")
check("export is csv", r.mimetype, "text/csv")
check("export is an attachment",
      "attachment" in r.headers.get("Content-Disposition", ""), True)
body = r.data.decode("utf-8-sig")
lines = [l for l in body.splitlines() if l.strip()]
check("export header", lines[0],
      "timestamp_utc,site,ai_platform,category,url,status_code")
# AI-only by default: 3 data rows, and no Bingbot.
check("export respects the AI-only default", len(lines) - 1, 3)
check("export excludes search by default", "Bingbot" in body, False)

r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08&include_search=1")
check("export includes search when asked", "Bingbot" in r.data.decode("utf-8-sig"), True)
check("export row count header", r.headers.get("X-Row-Count"), "4")

r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08&site=bravo")
check("export respects the site filter", len(
    [l for l in r.data.decode("utf-8-sig").splitlines() if l.strip()]) - 1, 1)
r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08"
               "&include_search=1&q=dog-bite")
check("export respects the URL search", len(
    [l for l in r.data.decode("utf-8-sig").splitlines() if l.strip()]) - 1, 2)
check("export needs login", A.app.test_client().get("/export.csv").status_code, 302)

# --- chrome -----------------------------------------------------------------
page = client.get("/").data.decode()
check("logo is served locally, not hotlinked", "/static/blushark-digital.png" in page, True)
check("logo does not hotlink the original host", "trustedlegalpartners.com" in page, False)
check("logo links to the dashboard", '<a class="logo" href="/"' in page, True)
check("no Overview nav link", ">Overview<" in page, False)
check("no Detail nav link", ">Detail<" in page, False)
check("logout still there", ">Log out<" in page, True)
check("brand blue is the logo's", "--brand:#00bcff" in page, True)
check("site names link to detail", 'href="/detail?' in page, True)
check("north-east arrow on View hits", "&#8599;" in page or "↗" in page, True)
check("tooltips present", 'class="info"' in page, True)
# Jinja autoescapes, so the apostrophe arrives as &#39; -- match on escaped text.
check("bot tooltip text", "OpenAI&#39;s collector" in page, True)
check("group tooltip text", "help train an AI model" in page, True)


# --- content filter --------------------------------------------------------
import content as C  # noqa: E402

for url, expect in [
    ("/case-types/dog-bite/", None), ("/", None),
    ("/js/main.js?v=1", "asset"), ("/style.css", "asset"), ("/logo.svg", "asset"),
    ("/favicon.ico", "asset"), ("/fonts/inter.woff2", "asset"), ("/img/HERO.JPG", "asset"),
    ("/robots.txt", "robots"),
    ("/wp-json/wp/v2/posts", "api"),
    ("/.env", "probe"), ("/.env.bak", "probe"), ("/.git/config", "probe"),
    ("/.ssh/id_ed25519", "probe"), ("/secrets.yml", "probe"),
    ("/service-account.json", "probe"), ("/wp-config.php.bak", "probe"),
    ("/config/database.yaml", "probe"), ("/.aws/credentials", "probe"),
]:
    check("classify %s" % url, C.classify(url), expect)

# Segment matching, not substring: these are real pages and must NOT be hidden.
for url in ["/blog/the-secret-to-winning-your-case/",
            "/uploads/2023/flyer.envelope.pdf",
            "/practice/environmental-law/",
            "/articles/git-good-at-hiring/",
            "/blog/secrets-of-a-good-claim/",
            "/team/robert-ssh-jones/"]:
    check("real page kept: %s" % url, C.is_content(url), True)

# --- the toggle ------------------------------------------------------------
with A.app.test_request_context("/"):
    check("content filter defaults ON", A._content_only(), True)
with A.app.test_request_context("/?content=0"):
    check("content=0 turns it off", A._content_only(), False)
# The form pairs a hidden 0 with the checkbox's 1; the LAST value wins. Reading
# the first (request.args.get) would leave the box permanently off.
with A.app.test_request_context("/?content=0&content=1"):
    check("ticked checkbox (0 then 1) -> ON", A._content_only(), True)
with A.app.test_request_context("/?content=0"):
    check("unticked checkbox (0 only) -> OFF", A._content_only(), False)

# Filtering happens before counting, so counts and rows always agree.
# Add three non-content hits AND keep the fake summary consistent with them: the
# overview reads the summary either way -- its plain counts when the filter is
# off, its content_hits/by_platform_content when on (just as the parser writes
# both). The two must agree with the events or toggling would move the numbers
# for the wrong reason. The invariant is asserted below. The added hits are an
# asset and two probes, so every content_* count stays at its base value.
EVENTS["events/date=2026-07-08/site=a.com/20260709-0017-alpha.jsonl"] += [
    {"timestamp": "2026-07-08T05:00:00Z", "site": "a.com", "ai_platform": "GPTBot",
     "category": "ai_training", "url": "/js/app.js", "status_code": 200},
    {"timestamp": "2026-07-08T06:00:00Z", "site": "a.com", "ai_platform": "GPTBot",
     "category": "ai_training", "url": "/.ssh/id_ed25519", "status_code": 404},
    {"timestamp": "2026-07-08T07:00:00Z", "site": "a.com", "ai_platform": "OAI-SearchBot",
     "category": "search", "url": "/secrets.yml", "status_code": 404},
]
SUMMARY["totals"]["hits"] = 7                       # content_hits stays 4
SUMMARY["by_platform"]["GPTBot"]["hits"] = 3        # content_hits stays 1
SUMMARY["by_platform"]["OAI-SearchBot"]["hits"] = 2  # content_hits stays 1
SUMMARY["by_site"]["a.com"] = {"hits": 5, "content_hits": 2, "bytes": 50,
                               "by_platform": {"GPTBot": 3, "OAI-SearchBot": 2},
                               "by_platform_content": {"GPTBot": 1, "OAI-SearchBot": 1}}
s3data._shard_cache.clear()
s3data._summary_cache.clear()

_, total_all, _, _ = s3data.query(["2026-07-08"], groups=list(cat.AI_GROUPS))
_, total_content, by_group, _ = s3data.query(["2026-07-08"], groups=list(cat.AI_GROUPS),
                                             content_only=True)
check("unfiltered counts everything", total_all, 6)
check("content filter drops the 3 non-content rows", total_content, 3)
check("filtered counts still reconcile", sum(by_group.values()), total_content)

rows, _, _, _ = s3data.query(["2026-07-08"], groups=list(cat.AI_GROUPS), content_only=True)
check("no assets in the rows", [r for r in rows if ".js" in r[s3data.URL]], [])
check("no probes in the rows", [r for r in rows if "ssh" in r[s3data.URL]], [])

# rollup (the detail/export path, and the overview's fallback for un-rebuilt
# days) must agree with query()
plats, sites = s3data.rollup(["2026-07-08"], groups=list(cat.AI_GROUPS), content_only=True)
check("rollup total matches query total", sum(plats.values()), total_content)
# rollup counts installs, not vhosts: alpha's two partitions are one site.
check("rollup counts installs, not vhosts", sorted(sites), ["alpha", "bravo"])
plats_all, _ = s3data.rollup(["2026-07-08"], groups=list(cat.AI_GROUPS))
check("unfiltered rollup matches unfiltered query", sum(plats_all.values()), total_all)

# End to end: the overview still reconciles with the filter on, and the numbers
# drop when it is on.
page_on = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08").data.decode()
hits_on, cats_on = stat_boxes(page_on)
check("overview reconciles with filter ON", sum(cats_on.values()), hits_on)
page_off = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08&content=0").data.decode()
hits_off, cats_off = stat_boxes(page_off)
check("overview reconciles with filter OFF", sum(cats_off.values()), hits_off)
check("filter ON shows fewer hits than OFF", hits_on < hits_off, True)
# The overview now reads the summary's precomputed content_hits; this pins them
# against the same rows counted straight from the events, so the parser's
# precompute and a live scan cannot drift apart.
check("summary content_hits match the shard count", hits_on, total_content)
# And the same when nothing is filtered: the overview's plain summary counts
# must equal the events. If either pair drifts, toggling the filter would move
# the numbers for the wrong reason.
check("summaries agree with the events when unfiltered", hits_off, total_all)

# Export respects it too.
r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08")
check("export applies the content filter", r.headers.get("X-Row-Count"), str(total_content))
check("export names the filter in the filename",
      "content-pages" in r.headers.get("Content-Disposition", ""), True)
r = client.get("/export.csv?preset=custom&start=2026-07-08&end=2026-07-08&content=0")
check("export honours filter off", r.headers.get("X-Row-Count"), str(total_all))

check("checkbox rendered on detail", 'name="content"' in client.get("/detail").data.decode(), True)

# --- By site: live search + client-side pagination --------------------------
page = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08").data.decode()
site_card = page.split("By site")[1]
check("site search box present", 'id="siteSearch"' in site_card, True)
check("site pager present", 'id="sitePager"' in site_card, True)
check("empty-state row present", 'id="siteEmpty"' in site_card, True)
check("25 sites per page", "PER_PAGE = 25" in site_card, True)

# Every site must be in the DOM (filtering/paging is client-side), and carry a
# lowercased data-site so the search can match case-insensitively. Both the
# domain and the install are in the key: people search by domain, but a zero-hit
# site has no domain to search by.
rendered = _re.findall(r'<tr data-site="([^"]+)"', site_card)
check("every tracked site rendered, hits or not", sorted(rendered),
      ["a.com alpha", "b.com bravo", "charlie charlie"])
check("data-site is lowercased for the search",
      all(s == s.lower() for s in rendered), True)

# Rows must arrive already sorted by hits, highest first: the JS pages them in
# DOM order and never re-sorts.
hits_in_order = [int(h.replace(",", "")) for h in
                 _re.findall(r'<td class="num">([\d,]+)</td>', site_card)]
check("site rows are sorted by hits desc", hits_in_order,
      sorted(hits_in_order, reverse=True))
check("the zero-hit site sorts last", hits_in_order[-1], 0)

# A site whose name differs in case still matches: verified via the data-site
# attribute the search reads, not the display text.
check("display name kept, match key lowercased",
      '<tr data-site="a.com alpha"' in site_card and ">a.com</a>" in site_card, True)

# --- site identity: one install is one site ---------------------------------
# The whole point. alpha serves a.com and a.wpengine.com; that is ONE site with
# 3 hits, not two sites. The holding domain must never appear as its own row.
check("WP Engine holding domain is not a row of its own",
      [s for s in rendered if "wpengine" in s], [])
# Bingbot only hit the holding domain, and it is a search engine -- so include
# search to make it count, and alpha's row must then carry it. This is the check
# that would fail if the merge broke: without it a.wpengine.com would be its own
# row with 1 hit and a.com would stay at 2.
with_search = client.get("/?preset=custom&start=2026-07-08&end=2026-07-08"
                         "&include_search=1").data.decode().split("By site")[1]
check("no holding-domain row even when search is included",
      [s for s in _re.findall(r'<tr data-site="([^"]+)"', with_search) if "wpengine" in s], [])
check("a.com row carries the holding domain's hits too",
      _re.search(r'<tr data-site="a\.com alpha"[^>]*>.*?<td class="num">(\d+)</td>',
                 with_search, _re.S).group(1), "3")

# The stat box: sites-with-hits over sites-tracked, and the first must fit
# inside the second or the headline is lying.
sites_box = page.split('<div class="big sites">')[1].split("</div>\n    </div>")[0]
with_hits = int(_re.search(r'<div class="n">([\d,]+)</div>', sites_box).group(1))
tracked = int(_re.search(r'<span class="of">/([\d,]+)</span>', sites_box).group(1))
check("stat box counts sites with hits", with_hits, 2)          # alpha, bravo
check("stat box counts every tracked site", tracked, 3)         # + charlie, empty log
check("sites with hits fits inside sites tracked", with_hits <= tracked, True)
check("the tracked total comes from the manifest, not from events",
      tracked, len(MANIFEST["processed"]))

# tracked_installs is derived from log FILENAMES, so an empty log still counts.
check("tracked installs include the one with no events",
      s3data.tracked_installs(), ["alpha", "bravo", "charlie"])
check("install -> vhosts is read from the event keys",
      s3data.install_domains(["2026-07-08"]),
      {"alpha": {"a.com", "a.wpengine.com"}, "bravo": {"b.com"}})
check("vhost -> install inverts cleanly", s3data.domain_to_install(["2026-07-08"]),
      {"a.com": "alpha", "a.wpengine.com": "alpha", "b.com": "bravo"})

# The display name never depends on hit counts -- the label must not move when
# the date range does.
check("real domain beats the holding domain",
      sites_mod.display_name("alpha", {"a.com", "a.wpengine.com"}), "a.com")
check("an empty log falls back to the install name",
      sites_mod.display_name("charlie", set()), "charlie")
check("only a holding domain -> use it",
      sites_mod.display_name("solo", {"solo.wpenginepowered.com"}), "solo.wpenginepowered.com")
# The real cases that defeat string matching, pinned.
check("install name need not resemble the domain",
      sites_mod.display_name("amdmedicalpllc", {"amdmedicalgroup.com"}), "amdmedicalgroup.com")
check("exact match wins over other real domains",
      sites_mod.display_name("buckfirelaw", {"buckfirelaw.com", "michiganautolaws.com"}),
      "buckfirelaw.com")
check("prefix match wins when there is no exact one",
      sites_mod.display_name("blushark", {"blusharkdigital.com", "lsa411.com"}),
      "blusharkdigital.com")
check("+N counts the other real domains, not the alias",
      sites_mod.extra_domain_count("alpha", {"a.com", "a.wpengine.com"}), 0)
check("+N counts genuinely distinct brands on one install",
      sites_mod.extra_domain_count("buckfirelaw",
                                   {"buckfirelaw.com", "michiganautolaws.com",
                                    "buckfirelaw.wpengine.com"}), 1)

if FAILURES:
    print("FAILED %d check(s):\n" % len(FAILURES))
    for f in FAILURES:
        print("  " + f)
    sys.exit(1)
print("all viewer checks passed")