"""Submitting, listing, the endpoint, rechecks and the directory page."""

from datetime import timedelta
import functools

import pytest

from mews import app, check, db, pages
from mews.fetch import Fetcher, registered_domain


@pytest.fixture
def loopback(monkeypatch):
    """Let the service read the test server, and give it a registrable domain.

    The test server answers on 127.0.0.1, which has no public suffix, so the
    name it would be listed under is supplied here rather than relaxing the
    rule in the service.
    """
    monkeypatch.setattr(app, "Fetcher", functools.partial(Fetcher, allow_loopback=True))
    monkeypatch.setattr(
        check,
        "registered_domain",
        lambda host: "example.test" if host == "127.0.0.1" else registered_domain(host),
    )


@pytest.fixture
def service(tmp_path, monkeypatch, loopback):
    """The app, wired to an empty database and a temporary directory page."""
    database = tmp_path / "mews.db"
    monkeypatch.setenv("MEWS_DB", str(database))
    monkeypatch.setenv("MEWS_DIRECTORY", str(tmp_path / "directory.html"))
    handle = db.connect(str(database))
    db.init(handle)
    handle.close()
    return tmp_path


def post(url: str = "", *, listing: bool = False, client: str = "203.0.113.5"):
    """Send one form submission to the WSGI app and return status and body."""
    from io import BytesIO
    from urllib.parse import urlencode

    fields = {"url": url, "list" if listing else "check": "x"}
    data = urlencode(fields).encode()
    captured: dict = {}

    def start_response(status, headers):
        captured["status"] = status
        captured["headers"] = headers

    body = b"".join(
        app.application(
            {
                "PATH_INFO": "/result",
                "REQUEST_METHOD": "POST",
                "CONTENT_LENGTH": str(len(data)),
                "wsgi.input": BytesIO(data),
                "HTTP_X_REAL_IP": client,
            },
            start_response,
        )
    )
    return captured, body.decode()


def test_a_conforming_page_is_listed(service, site, conforming):
    url = site.add("/index.html", conforming())
    captured, body = post(url, listing=True)
    assert captured["status"] == "200 OK"
    assert "Your site is listed" in body

    handle = db.connect()
    assert [row["domain"] for row in db.listed(handle)] == ["example.test"]
    assert (service / "directory.html").exists()
    assert "example.test" in (service / "directory.html").read_text()


def test_checking_does_not_list(service, site, conforming):
    url = site.add("/index.html", conforming())
    _, body = post(url, listing=False)
    assert "This page conforms" in body
    assert db.listed(db.connect()) == []


def test_a_failing_page_is_not_listed_and_says_why(service, site, fail_page):
    url = site.add("/index.html", fail_page("div"))
    _, body = post(url, listing=True)
    assert "doesn't conform" in body
    assert "Section 4.1" in body
    assert db.listed(db.connect()) == []


def test_every_response_is_a_mews_page(service, site, conforming):
    from mews.lint import validate_bytes

    url = site.add("/index.html", conforming())
    for listing in (False, True):
        _, body = post(url, listing=listing)
        assert validate_bytes(body.encode()).conforms


def test_no_response_sets_a_cookie(service, site, conforming):
    url = site.add("/index.html", conforming())
    captured, _ = post(url, listing=True)
    assert not [name for name, _ in captured["headers"] if name.lower() == "set-cookie"]


def test_responses_are_not_cached(service, site, conforming):
    url = site.add("/index.html", conforming())
    captured, _ = post(url, listing=True)
    assert ("Cache-Control", "no-store") in captured["headers"]


def test_an_address_that_is_not_a_page_is_refused(service):
    _, body = post("http://10.0.0.1/", listing=True)
    assert "domain name rather than an IP address" in body


def test_an_unreachable_page_is_reported(service, site):
    _, body = post(site.base + "/missing.html", listing=True)
    assert "That page returned 404" in body


def test_resubmitting_a_site_updates_the_same_row(
    service, site, conforming, monkeypatch
):
    """One row per registrable domain, however many pages of it are submitted."""
    first = site.add("/a.html", conforming())
    post(first, listing=True)
    # Past the per-domain cooldown, which would otherwise turn this away.
    later = db.now() + timedelta(hours=2)
    monkeypatch.setattr(db, "now", lambda: later)
    second = site.add("/b.html", conforming())
    post(second, listing=True, client="203.0.113.6")
    rows = db.listed(db.connect())
    assert len(rows) == 1
    assert rows[0]["url"] == second


def test_a_blocked_domain_is_refused(service, site, conforming):
    handle = db.connect()
    db.block(handle, "example.test", "spam")
    url = site.add("/index.html", conforming())
    _, body = post(url, listing=True)
    assert "taken out of the directory" in body


def test_the_endpoint_answers_other_paths_and_methods_politely(service):
    from io import BytesIO

    captured: dict = {}

    def start_response(status, headers):
        captured["status"] = status

    body = b"".join(
        app.application(
            {
                "PATH_INFO": "/result",
                "REQUEST_METHOD": "GET",
                "wsgi.input": BytesIO(b""),
            },
            start_response,
        )
    ).decode()
    assert captured["status"].startswith("405")
    assert "Nothing to show" in body


def test_rate_limits_stop_a_flood(connection):
    client = db.ip_hash(connection, "203.0.113.9")
    for _ in range(db.IP_LISTINGS_PER_HOUR):
        assert (
            db.rate_limited(connection, client=client, domain="a.example", listing=True)
            is None
        )
        db.record_submission(
            connection,
            url="https://a.example/",
            domain="other.example",
            client=client,
            outcome="listed",
        )
    assert "Try again in an hour" in db.rate_limited(
        connection, client=client, domain="a.example", listing=True
    )


def test_checking_a_page_does_not_block_listing_it(connection):
    """Checking first is the sensible move; it must not cost the listing."""
    client = db.ip_hash(connection, "203.0.113.9")
    db.record_submission(
        connection,
        url="https://a.example/",
        domain="a.example",
        client=client,
        outcome="checked",
    )
    assert (
        db.rate_limited(connection, client=client, domain="a.example", listing=True)
        is None
    )


def test_a_domain_has_to_wait_between_submissions(connection):
    client = db.ip_hash(connection, "203.0.113.9")
    db.record_submission(
        connection,
        url="https://a.example/",
        domain="a.example",
        client=client,
        outcome="rejected",
    )
    assert "ten minutes" in db.rate_limited(
        connection, client=client, domain="a.example", listing=True
    )


def test_a_domain_that_keeps_failing_waits_a_day(connection, monkeypatch):
    client = db.ip_hash(connection, "203.0.113.9")
    base = db.now()
    for index in range(db.DOMAIN_REJECTS_BEFORE_SLOWDOWN):
        monkeypatch.setattr(db, "now", lambda i=index: base - timedelta(hours=i + 1))
        db.record_submission(
            connection,
            url="https://a.example/",
            domain="a.example",
            client=client,
            outcome="rejected",
        )
    monkeypatch.setattr(db, "now", lambda: base)
    assert "try again tomorrow" in db.rate_limited(
        connection, client=client, domain="a.example", listing=True
    )


# --- rechecks ----------------------------------------------------------


def _list_site(connection, url):
    return db.upsert_site(
        connection,
        domain="example.test",
        url=url,
        title="Test site",
        description="",
        language="en",
        etag=None,
        last_modified=None,
    )[0]


def test_a_page_that_still_conforms_stays(connection, site, conforming):
    url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
    site_id = _list_site(connection, url)
    with Fetcher(allow_loopback=True) as fetcher:
        assert (
            check.recheck(
                connection, db.site(connection, "example.test"), fetcher=fetcher
            )
            == "pass"
        )
    assert db.site(connection, "example.test")["state"] == "listed"
    assert db.site(connection, "example.test")["etag"] == '"v1"'
    assert site_id


def test_a_page_that_stopped_conforming_is_dropped(connection, site, fail_page):
    url = site.add("/index.html", fail_page("div"))
    _list_site(connection, url)
    with Fetcher(allow_loopback=True) as fetcher:
        assert (
            check.recheck(
                connection, db.site(connection, "example.test"), fetcher=fetcher
            )
            == "fail"
        )
    row = db.site(connection, "example.test")
    assert row["state"] == "removed"
    assert "Section 4.1" in row["reason"]
    assert db.listed(connection) == []


def test_a_not_modified_answer_passes_without_rereading(connection, site, conforming):
    url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
    _list_site(connection, url)
    site.add("/index.html", b"", status=304, headers={"ETag": '"v1"'})
    connection.execute("UPDATE sites SET etag = '\"v1\"'")
    with Fetcher(allow_loopback=True) as fetcher:
        assert (
            check.recheck(
                connection, db.site(connection, "example.test"), fetcher=fetcher
            )
            == "pass"
        )
    assert db.site(connection, "example.test")["state"] == "listed"
    assert "If-None-Match" in site.requests[-1][1]


def test_downtime_takes_several_tries_to_drop_a_site(connection, site, conforming):
    url = site.add("/index.html", conforming())
    _list_site(connection, url)
    site.routes.clear()
    site.add("/index.html", b"broken", status=503)
    for _ in range(db.TRANSIENT_LIMIT - 1):
        with Fetcher(allow_loopback=True) as fetcher:
            assert (
                check.recheck(
                    connection, db.site(connection, "example.test"), fetcher=fetcher
                )
                == "unreachable"
            )
        assert db.site(connection, "example.test")["state"] == "listed"
    with Fetcher(allow_loopback=True) as fetcher:
        assert (
            check.recheck(
                connection, db.site(connection, "example.test"), fetcher=fetcher
            )
            == "fail"
        )
    assert db.site(connection, "example.test")["state"] == "removed"


def test_only_sites_that_are_due_are_rechecked(connection, monkeypatch):
    _list_site(connection, "https://example.test/")
    assert db.due(connection) == []
    later = db.now() + timedelta(days=db.RECHECK_DAYS + 1)
    monkeypatch.setattr(db, "now", lambda: later)
    assert len(db.due(connection)) == 1


# --- the directory page ------------------------------------------------


def test_the_directory_page_conforms(connection, tmp_path):
    from mews.lint import validate_bytes

    _list_site(connection, "https://example.test/")
    target = tmp_path / "directory.html"
    pages.write_directory(str(target), db.listed(connection))
    assert validate_bytes(target.read_bytes()).conforms


def test_a_hostile_title_is_escaped_and_stripped(connection, tmp_path):
    from mews.lint import validate_bytes

    connection.execute(
        "INSERT INTO sites (domain, url, title, state, listed_at) "
        "VALUES ('x.test', 'https://x.test/', '<script>‮oops</script>', "
        "'listed', '2026-01-01T00:00:00+00:00')"
    )
    connection.commit()
    target = tmp_path / "directory.html"
    pages.write_directory(str(target), db.listed(connection))
    text = target.read_text()
    assert "<script>" not in text
    assert "&lt;script&gt;" in text
    assert validate_bytes(target.read_bytes()).conforms


def test_the_directory_page_is_readable_by_the_web_server(connection, tmp_path):
    """It is served off disk by another user, whatever umask the service has."""
    import os
    import stat

    target = tmp_path / "directory.html"
    old_umask = os.umask(0o077)
    try:
        pages.write_directory(str(target), db.listed(connection))
    finally:
        os.umask(old_umask)
    assert stat.S_IMODE(target.stat().st_mode) == 0o644


def test_an_empty_directory_still_conforms(connection, tmp_path):
    from mews.lint import validate_bytes

    target = tmp_path / "directory.html"
    pages.write_directory(str(target), db.listed(connection))
    assert validate_bytes(target.read_bytes()).conforms
    assert "Nothing here yet" in target.read_text()


def test_a_page_that_would_not_conform_leaves_the_old_one_alone(
    connection, tmp_path, monkeypatch
):
    target = tmp_path / "directory.html"
    pages.write_directory(str(target), db.listed(connection))
    before = target.read_text()
    monkeypatch.setattr(pages, "directory", lambda rows: "<p>not a page</p>")
    with pytest.raises(ValueError, match="doesn't conform"):
        pages.write_directory(str(target), db.listed(connection))
    assert target.read_text() == before