"""Storage path handling, rate limiting, and response hardening."""
from __future__ import annotations

import pytest

from dsh_sync.config import Config
from dsh_sync.ratelimit import SlidingWindow
from dsh_sync.storage import artifact_path, artifact_relpath


# ------------------------------------------------------------------ storage


def _cfg(tmp_path) -> Config:
    return Config(
        data_dir=tmp_path,
        plugins_dir=tmp_path / "plugins",
        database_url="postgresql://unused",
        bootstrap_admin_token=None,
        invite_code=None,
        admin_username="admin",
        max_plugin_bytes=1024,
    )


def test_artifact_relpath_is_relative(tmp_path):
    """Stored paths must not embed a machine-specific prefix."""
    rel = artifact_relpath("demo", "1.0.0")
    assert rel == "demo/1.0.0.tar.gz"
    assert not rel.startswith("/") and ":\\" not in rel


def test_artifact_path_resolves_under_plugins_dir(tmp_path):
    cfg = _cfg(tmp_path)
    p = artifact_path(cfg, "demo/1.0.0.tar.gz")
    assert p == (tmp_path / "plugins" / "demo" / "1.0.0.tar.gz").resolve()
    assert p.is_relative_to(cfg.plugins_dir.resolve())


def test_artifact_path_accepts_legacy_absolute_rows(tmp_path):
    """Rows written by older builds hold absolute paths; they must still load."""
    cfg = _cfg(tmp_path)
    absolute = tmp_path / "plugins" / "demo" / "2.0.0.tar.gz"
    assert artifact_path(cfg, str(absolute)) == absolute


def test_artifact_path_rejects_escape(tmp_path):
    cfg = _cfg(tmp_path)
    with pytest.raises(ValueError):
        artifact_path(cfg, "../../etc/passwd")


# ---------------------------------------------------------------- ratelimit


def test_sliding_window_allows_up_to_limit():
    w = SlidingWindow(limit=3, window_seconds=60)
    for _ in range(3):
        w.check("1.2.3.4")


def test_sliding_window_blocks_over_limit():
    from fastapi import HTTPException

    w = SlidingWindow(limit=2, window_seconds=60)
    w.check("k")
    w.check("k")
    with pytest.raises(HTTPException) as e:
        w.check("k")
    assert e.value.status_code == 429
    assert "Retry-After" in e.value.headers


def test_sliding_window_is_per_key():
    w = SlidingWindow(limit=1, window_seconds=60)
    w.check("a")
    w.check("b")  # different peer, own budget


def test_sliding_window_expires():
    import time

    w = SlidingWindow(limit=1, window_seconds=0.05)
    w.check("k")
    time.sleep(0.08)
    w.check("k")  # window rolled over


def test_invite_endpoint_rate_limited(client):
    """Wrong invite codes stop being free after the budget is spent."""
    codes = [client.post("/v1/tokens", json={"username": f"u{i}", "invite_code": "nope"})
             for i in range(12)]
    assert all(c.status_code == 403 for c in codes[:10]), [c.status_code for c in codes]
    assert codes[10].status_code == 429
    assert "Retry-After" in codes[10].headers


# ------------------------------------------------------------------ headers


def test_security_headers_on_api_responses(client):
    h = client.get("/healthz").headers
    assert h["x-content-type-options"] == "nosniff"
    assert h["x-frame-options"] == "DENY"
    assert h["referrer-policy"] == "no-referrer"
    # camera must stay same-origin: the PWA's in-app scanner needs getUserMedia
    assert "camera=(self)" in h["permissions-policy"]


def test_security_headers_on_pages(client):
    for path in ("/app", "/apk"):
        h = client.get(path).headers
        assert h["x-content-type-options"] == "nosniff", path
        assert h["x-frame-options"] == "DENY", path


def test_successful_redeem_clears_the_rate_limit(client):
    """A user who mistypes the code must not stay locked out after succeeding."""
    for i in range(5):
        assert client.post("/v1/tokens", json={"username": f"x{i}", "invite_code": "bad"}
                           ).status_code == 403
    ok = client.post("/v1/tokens", json={"username": "good", "invite_code": "invite-123"})
    assert ok.status_code == 200, ok.text

    from conftest import INVITE_CODE

    # Bucket was cleared, so the next peer-budget starts fresh rather than
    # inheriting the 5 failures.
    again = client.post("/v1/tokens", json={"username": "good2", "invite_code": INVITE_CODE})
    assert again.status_code == 200, again.text