"""API tokens: long-lived credentials for scripts and CI.

A token is a credential that outlives a session and is handed to a machine, so
the things worth pinning down are the ones that go wrong quietly:

* only a hash is stored, and the token is returned exactly once,
* a read-scoped token really is read-only, even when its owner is an admin,
* a token never outranks its owner — demote or disable the account and the
  token follows,
* and a token cannot mint another one, so a leak cannot be made permanent.
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
from sqlmodel import Session, delete, select


@pytest.fixture
def svc(db):
    from services import api_token_service

    return api_token_service


@pytest.fixture(autouse=True)
def clean_tokens(db):
    from database import engine
    from models.api_token import ApiToken

    with Session(engine) as session:
        session.exec(delete(ApiToken))
        session.commit()
    yield
    with Session(engine) as session:
        session.exec(delete(ApiToken))
        session.commit()


@pytest.fixture
def owner(db):
    """A throwaway admin, so tests can disable or demote it freely."""
    from database import engine
    from models.user import User
    import auth as auth_mod

    with Session(engine) as session:
        existing = session.exec(select(User).where(User.username == "token-owner")).first()
        if existing:
            session.delete(existing)
            session.commit()
        user = User(
            username="token-owner",
            hashed_password=auth_mod.hash_password("pw"),
            role="admin",
        )
        session.add(user)
        session.commit()
        session.refresh(user)
        user_id = user.id
    yield user_id
    with Session(engine) as session:
        row = session.get(User, user_id)
        if row:
            session.delete(row)
            session.commit()


def _mint(svc, owner_id: int, scope: str = "admin", expires_in_days=None) -> str:
    from database import engine
    from models.user import User

    with Session(engine) as session:
        user = session.get(User, owner_id)
        _row, token = svc.mint(
            session, name=f"test-{scope}", user=user, scope=scope,
            expires_in_days=expires_in_days,
        )
    return token


def _as(client, token: str):
    """A tiny client that authenticates with a raw bearer token."""

    class _Client:
        def get(self, url, **kw):
            return client.get(url, headers={"Authorization": f"Bearer {token}"}, **kw)

        def post(self, url, **kw):
            return client.post(url, headers={"Authorization": f"Bearer {token}"}, **kw)

        def delete(self, url, **kw):
            return client.delete(url, headers={"Authorization": f"Bearer {token}"}, **kw)

    return _Client()


# --------------------------------------------------------------------------- #
# What is stored
# --------------------------------------------------------------------------- #


def test_the_token_is_stored_only_as_a_hash(svc, owner):
    from database import engine
    from models.api_token import ApiToken

    token = _mint(svc, owner)
    with Session(engine) as session:
        row = session.exec(select(ApiToken)).one()
    assert token not in row.token_hash
    assert row.token_hash != token
    # The readable front is kept so the UI can name the row, and it is far too
    # short to authenticate with.
    assert token.startswith(row.prefix)
    assert len(row.prefix) < len(token) / 2


def test_tokens_are_unique_and_prefixed(svc, owner):
    first, second = _mint(svc, owner), _mint(svc, owner)
    assert first != second
    assert first.startswith("sp_") and second.startswith("sp_")


def test_a_tampered_token_does_not_resolve(svc, owner):
    from database import engine

    token = _mint(svc, owner)
    with Session(engine) as session:
        assert svc.resolve(session, token) is not None
        assert svc.resolve(session, token[:-1] + "x") is None
        assert svc.resolve(session, "sp_totallymadeupvalue") is None
        # A JWT must not be mistaken for one.
        assert svc.resolve(session, "eyJhbGciOiJIUzI1NiJ9.e30.x") is None


# --------------------------------------------------------------------------- #
# Using one
# --------------------------------------------------------------------------- #


def test_a_token_authenticates_like_a_session(client, svc, owner):
    response = _as(client, _mint(svc, owner)).get("/api/auth/me")
    assert response.status_code == 200
    assert response.json()["username"] == "token-owner"


def test_a_read_token_cannot_change_anything(client, svc, owner):
    """The owner is an admin; the token is not. That is the point of scopes."""
    api = _as(client, _mint(svc, owner, scope="read"))
    assert api.get("/api/stacks").status_code == 200
    denied = api.post("/api/registries", json={"host": "ghcr.io", "username": "a", "password": "b"})
    assert denied.status_code == 403
    assert "read-only" in denied.json()["detail"]


def test_an_admin_token_may_act(client, svc, owner):
    api = _as(client, _mint(svc, owner, scope="admin"))
    created = api.post(
        "/api/registries",
        json={"host": "quay.io", "username": "a", "password": "b"},
    )
    assert created.status_code == 201, created.text
    api.delete(f"/api/registries/{created.json()['id']}")


def test_using_a_token_records_when(client, svc, owner):
    from database import engine
    from models.api_token import ApiToken

    _as(client, _mint(svc, owner)).get("/api/auth/me")
    with Session(engine) as session:
        assert session.exec(select(ApiToken)).one().last_used_at is not None


def test_the_last_used_write_is_throttled(svc, owner, monkeypatch):
    """Otherwise every API call is also a database write."""
    from database import engine
    from models.api_token import ApiToken

    token = _mint(svc, owner)
    with Session(engine) as session:
        row = svc.resolve(session, token)[0]
        svc.touch(session, row)
        first = row.last_used_at
        svc.touch(session, row)
        assert row.last_used_at == first

        # Far enough in the past and it is written again.
        row.last_used_at = datetime.now(timezone.utc) - timedelta(hours=1)
        session.add(row)
        session.commit()
        svc.touch(session, row)
        assert session.exec(select(ApiToken)).one().last_used_at != first


# --------------------------------------------------------------------------- #
# A token never outranks its owner
# --------------------------------------------------------------------------- #


def test_a_demoted_owner_drops_the_token_to_read_only(client, svc, owner):
    from database import engine
    from models.user import User

    token = _mint(svc, owner, scope="admin")
    with Session(engine) as session:
        user = session.get(User, owner)
        user.role = "user"
        session.add(user)
        session.commit()

    api = _as(client, token)
    assert api.get("/api/stacks").status_code == 200
    assert api.post("/api/registries", json={"host": "x.io", "username": "a", "password": "b"}).status_code == 403


def test_a_disabled_owner_kills_the_token(client, svc, owner):
    from database import engine
    from models.user import User

    token = _mint(svc, owner)
    with Session(engine) as session:
        user = session.get(User, owner)
        user.is_active = False
        session.add(user)
        session.commit()

    assert _as(client, token).get("/api/auth/me").status_code == 401


def test_an_expired_token_is_refused(client, svc, owner):
    from database import engine
    from models.api_token import ApiToken

    token = _mint(svc, owner, expires_in_days=1)
    with Session(engine) as session:
        row = session.exec(select(ApiToken)).one()
        row.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
        session.add(row)
        session.commit()

    response = _as(client, token).get("/api/auth/me")
    assert response.status_code == 401
    assert "not valid" in response.json()["detail"]


def test_a_token_with_no_expiry_does_not_expire(svc, owner):
    from database import engine
    from models.api_token import ApiToken

    _mint(svc, owner)
    with Session(engine) as session:
        assert svc.is_expired(session.exec(select(ApiToken)).one()) is False


# --------------------------------------------------------------------------- #
# A token cannot make itself permanent
# --------------------------------------------------------------------------- #


def test_a_token_cannot_mint_another_token(client, svc, owner):
    api = _as(client, _mint(svc, owner, scope="admin"))
    response = api.post("/api/auth/tokens", json={"name": "second", "scope": "admin"})
    assert response.status_code == 403
    assert "signed-in session" in response.json()["detail"]


def test_a_token_cannot_create_a_user(client, svc, owner):
    api = _as(client, _mint(svc, owner, scope="admin"))
    response = api.post(
        "/api/auth/users", json={"username": "backdoor", "password": "x", "role": "admin"}
    )
    assert response.status_code == 403


def test_a_token_cannot_even_list_tokens(client, svc, owner):
    assert _as(client, _mint(svc, owner, scope="admin")).get("/api/auth/tokens").status_code == 403


# --------------------------------------------------------------------------- #
# Through the API
# --------------------------------------------------------------------------- #


def test_create_shows_the_token_once_and_never_again(as_admin):
    created = as_admin.post("/api/auth/tokens", json={"name": "ci", "scope": "read"})
    assert created.status_code == 201, created.text
    token = created.json()["token"]
    assert token.startswith("sp_")

    listed = as_admin.get("/api/auth/tokens")
    assert token not in listed.text
    assert '"token"' not in listed.text
    assert listed.json()[0]["prefix"] == token[:11]


def test_revoking_a_token_stops_it_working(client, as_admin):
    created = as_admin.post("/api/auth/tokens", json={"name": "ci", "scope": "read"}).json()
    api = _as(client, created["token"])
    assert api.get("/api/auth/me").status_code == 200

    assert as_admin.delete(f"/api/auth/tokens/{created['id']}").status_code == 200
    assert api.get("/api/auth/me").status_code == 401


def test_a_bad_scope_or_empty_name_is_refused(as_admin):
    assert as_admin.post("/api/auth/tokens", json={"name": "x", "scope": "root"}).status_code == 400
    assert as_admin.post("/api/auth/tokens", json={"name": "  ", "scope": "read"}).status_code == 400
    assert (
        as_admin.post(
            "/api/auth/tokens", json={"name": "x", "scope": "read", "expires_in_days": 0}
        ).status_code
        == 400
    )


def test_the_read_only_role_cannot_manage_tokens(as_user):
    assert as_user.get("/api/auth/tokens").status_code == 403
    assert as_user.post("/api/auth/tokens", json={"name": "x"}).status_code == 403