from __future__ import annotations

import asyncio
import sys
import threading
from collections.abc import Callable
from http.cookies import SimpleCookie
from pathlib import Path

import pytest
from starlette.testclient import TestClient
from starlette.requests import Request

ROOT_DIR = Path(__file__).resolve().parents[1]
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from dashboard import authentication
from dashboard.auth_server import create_auth_routes, login


ADMIN_ID = "admin"
ADMIN_PASSWORD = "correct horse battery staple"
SIGNING_SECRET = "s" * 64


@pytest.fixture(autouse=True)
def _configured_auth(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv(authentication.ADMIN_ID_ENV, ADMIN_ID)
    monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, ADMIN_PASSWORD)
    monkeypatch.setenv(authentication.SIGNING_SECRET_ENV, SIGNING_SECRET)
    monkeypatch.setattr(authentication.time, "sleep", lambda _seconds: None)


@pytest.fixture
def client() -> TestClient:
    from starlette.applications import Starlette

    return TestClient(
        Starlette(routes=create_auth_routes()), base_url="https://testserver"
    )


def _cookie_from(response) -> SimpleCookie[str]:
    cookie: SimpleCookie[str] = SimpleCookie()
    cookie.load(response.headers["set-cookie"])
    return cookie


def test_successful_login_sets_hardened_24_hour_cookie(client: TestClient) -> None:
    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"] == "/"
    header = response.headers["set-cookie"]
    assert f"{authentication.AUTH_COOKIE_NAME}=" in header
    assert "HttpOnly" in header
    assert "Secure" in header
    assert "SameSite=lax" in header
    assert "Path=/" in header
    assert f"Max-Age={authentication.DEFAULT_TOKEN_TTL_SECONDS}" in header
    assert ADMIN_PASSWORD not in header
    assert SIGNING_SECRET not in header


def test_remember_me_sets_30_day_cookie(client: TestClient) -> None:
    response = client.post(
        "/api/auth/login",
        data={
            "admin_id": ADMIN_ID,
            "admin_password": ADMIN_PASSWORD,
            "remember_me": "on",
        },
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert (
        f"Max-Age={authentication.REMEMBER_ME_TOKEN_TTL_SECONDS}"
        in response.headers["set-cookie"]
    )


def test_unicode_credentials_can_login_through_adapter(client: TestClient, monkeypatch) -> None:
    unicode_id = "管理员"
    unicode_password = "päss🔐"
    monkeypatch.setenv(authentication.ADMIN_ID_ENV, unicode_id)
    monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, unicode_password)

    response = client.post(
        "/api/auth/login",
        data={"admin_id": unicode_id, "admin_password": unicode_password},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"] == "/"
    assert f"{authentication.AUTH_COOKIE_NAME}=" in response.headers["set-cookie"]


def test_unicode_credential_mismatch_is_generic_and_delayed(
    client: TestClient, monkeypatch
) -> None:
    import time as real_time

    unicode_id = "管理员"
    unicode_password = "päss🔐"
    monkeypatch.setenv(authentication.ADMIN_ID_ENV, unicode_id)
    monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, unicode_password)
    monkeypatch.setattr(authentication.time, "sleep", real_time.sleep)

    started = real_time.monotonic()
    response = client.post(
        "/api/auth/login",
        data={"admin_id": unicode_id, "admin_password": "неверный🔒"},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )
    elapsed = real_time.monotonic() - started

    assert response.status_code == 303
    assert response.headers["location"] == "/?auth_error=invalid"
    assert "set-cookie" not in response.headers
    assert elapsed >= authentication.MINIMUM_FAILURE_DELAY_SECONDS * 0.9


def test_invalid_credentials_do_not_set_cookie_and_use_general_error(
    client: TestClient,
) -> None:
    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": "wrong"},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"] == "/?auth_error=invalid"
    assert "set-cookie" not in response.headers
    assert "password" not in response.headers["location"]
    assert ADMIN_ID not in response.headers["location"]


def test_login_rejects_malformed_percent_encoded_utf8_without_cookie(
    client: TestClient,
) -> None:
    response = client.post(
        "/api/auth/login",
        content=b"admin_id=%FF&admin_password=wrong",
        headers={
            "origin": "https://testserver",
            "content-type": "application/x-www-form-urlencoded",
        },
        follow_redirects=False,
    )

    assert response.status_code == 400
    assert "set-cookie" not in response.headers


def test_missing_configuration_fails_closed(client: TestClient, monkeypatch) -> None:
    monkeypatch.delenv(authentication.SIGNING_SECRET_ENV)

    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"] == "/?auth_error=unavailable"
    assert "set-cookie" not in response.headers


def test_cross_site_login_is_rejected(client: TestClient) -> None:
    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
        headers={"origin": "https://attacker.example"},
        follow_redirects=False,
    )

    assert response.status_code == 403
    assert "set-cookie" not in response.headers


def test_different_port_is_not_same_origin(client: TestClient) -> None:
    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
        headers={"origin": "https://testserver:444"},
        follow_redirects=False,
    )

    assert response.status_code == 403
    assert "set-cookie" not in response.headers


def test_plain_http_is_rejected_even_with_spoofed_https_origin() -> None:
    from starlette.applications import Starlette

    with TestClient(
        Starlette(routes=create_auth_routes()), base_url="http://testserver"
    ) as http_client:
        response = http_client.post(
            "/api/auth/login",
            data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
            headers={
                "origin": "https://testserver",
                "x-forwarded-proto": "https",
            },
            follow_redirects=False,
        )

    assert response.status_code == 403
    assert "set-cookie" not in response.headers


def test_logout_clears_cookie_with_matching_scope(client: TestClient) -> None:
    response = client.post(
        "/api/auth/logout",
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    header = response.headers["set-cookie"]
    assert f"{authentication.AUTH_COOKIE_NAME}=" in header
    assert "Max-Age=0" in header
    assert "HttpOnly" in header
    assert "Secure" in header
    assert "SameSite=lax" in header
    assert "Path=/" in header


def test_cleanup_route_clears_cookie_and_redirects(client: TestClient) -> None:
    response = client.get(
        "/api/auth/cleanup",
        headers={"referer": "https://testserver/"},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"] == "/"
    header = response.headers["set-cookie"]
    assert f"{authentication.AUTH_COOKIE_NAME}=" in header
    assert "Max-Age=0" in header
    assert "HttpOnly" in header
    assert "Secure" in header
    assert "SameSite=lax" in header
    assert "Path=/" in header


def test_cleanup_route_rejects_cross_origin_navigation(client: TestClient) -> None:
    response = client.get(
        "/api/auth/cleanup",
        headers={"referer": "https://attacker.example/"},
        follow_redirects=False,
    )

    assert response.status_code == 403
    assert "set-cookie" not in response.headers


def test_cookie_token_round_trip(client: TestClient) -> None:
    response = client.post(
        "/api/auth/login",
        data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
        headers={"origin": "https://testserver"},
        follow_redirects=False,
    )
    cookie = _cookie_from(response)
    token = cookie[authentication.AUTH_COOKIE_NAME].value

    assert authentication.verify_persistent_token(
        authentication.load_auth_settings(), token
    ) is not None


def _streaming_login_request(
    chunks: list[bytes],
) -> tuple[Request, Callable[[], int]]:
    messages = [
        {
            "type": "http.request",
            "body": chunk,
            "more_body": index < len(chunks) - 1,
        }
        for index, chunk in enumerate(chunks)
    ]
    calls = 0

    async def receive():
        nonlocal calls
        calls += 1
        return messages.pop(0)

    scope = {
        "type": "http",
        "method": "POST",
        "scheme": "https",
        "path": "/api/auth/login",
        "raw_path": b"/api/auth/login",
        "query_string": b"",
        "headers": [
            (b"host", b"testserver"),
            (b"origin", b"https://testserver"),
            (b"content-type", b"application/x-www-form-urlencoded"),
        ],
        "client": ("testclient", 50000),
        "server": ("testserver", 443),
    }
    return Request(scope, receive), lambda: calls


def test_login_offloads_failed_login_delay_from_event_loop(monkeypatch) -> None:
    entered = threading.Event()
    release = threading.Event()
    probe_ran = threading.Event()
    probe_before_release = threading.Event()

    def blocking_delay(*, started_at):
        entered.set()
        if probe_ran.wait(timeout=1):
            probe_before_release.set()
        release.set()

    monkeypatch.setattr(authentication, "enforce_failed_login_delay", blocking_delay)

    async def exercise() -> None:
        request, _ = _streaming_login_request([b"admin_id=admin&admin_password=wrong"])
        task = asyncio.create_task(login(request))

        async def probe() -> None:
            probe_ran.set()

        asyncio.create_task(probe())
        assert await asyncio.to_thread(entered.wait, 1)
        release.set()
        response = await task
        assert response.status_code == 303
        assert response.headers["location"] == "/?auth_error=invalid"
        assert "set-cookie" not in response.headers

    asyncio.run(exercise())
    assert probe_before_release.is_set()


def test_login_rejects_exactly_one_byte_over_streaming_limit(monkeypatch) -> None:
    def unexpected_verification(*_args, **_kwargs):
        raise AssertionError("oversized body reached credential verification")

    monkeypatch.setattr(authentication, "verify_credentials", unexpected_verification)
    request, calls = _streaming_login_request([b"x" * 4097])
    response = asyncio.run(login(request))

    assert response.status_code == 400
    assert "set-cookie" not in response.headers
    assert calls() == 1


def test_login_accepts_body_at_streaming_limit() -> None:
    prefix = b"admin_id=admin&admin_password="
    body = prefix + (b"x" * (4096 - len(prefix)))
    assert len(body) == 4096
    request, calls = _streaming_login_request([body])
    response = asyncio.run(login(request))

    assert response.status_code == 303
    assert calls() == 1


def test_login_stops_reading_stream_after_limit_is_exceeded() -> None:
    request, calls = _streaming_login_request([b"x" * 4096, b"y", b"z"])
    response = asyncio.run(login(request))

    assert response.status_code == 400
    assert calls() == 2