familiar/tests/test_auth.py

"""EVE SSO auth: PKCE, JWT claims, multi-character store, exchange/refresh."""

import asyncio
import base64
import hashlib

import httpx
import pytest

from conftest import make_jwt
from familiar import auth


@pytest.fixture
def tmp_store(tmp_path, monkeypatch):
    """Point the token/key files at a temp dir."""
    monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
    return tmp_path


def _char(cid, name, **extra):
    return {
        "refresh_token": f"r-{cid}",
        "access_token": f"a-{cid}",
        "expires_at": 10_000,
        "character_id": cid,
        "character_name": name,
        "scopes": [],
        **extra,
    }


# --- pure helpers ---------------------------------------------------------- #
def test_generate_pkce_challenge_matches_verifier():
    verifier, challenge = auth.generate_pkce()
    expected = (
        base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
    )
    assert challenge == expected
    assert "=" not in challenge


def test_build_authorize_url_has_required_params():
    url = auth.build_authorize_url("the-state", "the-challenge")
    assert "response_type=code" in url
    assert "code_challenge=the-challenge" in url
    assert "code_challenge_method=S256" in url
    assert "state=the-state" in url


def test_decode_jwt_and_character_id():
    token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte"})
    claims = auth.decode_jwt_claims(token)
    assert claims["name"] == "Latte"
    assert auth.character_id_from_claims(claims) == 2112000001


def test_character_id_from_bad_sub_raises():
    with pytest.raises(auth.AuthError):
        auth.character_id_from_claims({"sub": "nonsense"})


def test_decode_malformed_token_raises():
    with pytest.raises(auth.AuthError):
        auth.decode_jwt_claims("not-a-jwt")


# --- multi-character store ------------------------------------------------- #
def test_store_starts_empty(tmp_store):
    assert auth.load_store() == {"characters": {}}
    assert auth.list_characters() == []


def test_save_and_list_multiple_characters(tmp_store):
    auth._save_character(_char(1, "Beta"))
    auth._save_character(_char(2, "Alpha"))
    chars = auth.list_characters()
    assert [c["character_name"] for c in chars] == ["Alpha", "Beta"]  # sorted by name


def test_legacy_single_bundle_is_migrated(tmp_store):
    # Write a pre-multi-character file (single bundle at top level).
    auth._write_store({"refresh_token": "r", "character_id": 42, "character_name": "Old"})
    store = auth.load_store()
    assert "42" in store["characters"]
    assert auth.list_characters()[0]["character_name"] == "Old"


def test_remove_character_by_name_and_all(tmp_store):
    auth._save_character(_char(1, "Beta"))
    auth._save_character(_char(2, "Alpha"))
    assert auth.remove_character("Alpha") is True
    assert {c["character_id"] for c in auth.list_characters()} == {1}
    assert auth.remove_character("nobody") is False
    assert auth.remove_character() is True  # removes all
    assert auth.list_characters() == []
    assert auth.remove_character() is False


def test_resolve_character_by_id_and_name(tmp_store):
    auth._save_character(_char(7, "Gamma"))
    assert auth.resolve_character(7) == 7
    assert auth.resolve_character("gamma") == 7
    with pytest.raises(auth.NotAuthenticated):
        auth.resolve_character("ghost")


def test_stored_tokens_are_encrypted(tmp_store):
    auth._save_character(_char(1, "X", refresh_token="super-secret-value"))
    raw = (tmp_store / "token.enc").read_bytes()
    assert b"super-secret-value" not in raw


# --- token requests + get_auth -------------------------------------------- #
def _token_transport(payload):
    return httpx.MockTransport(lambda req: httpx.Response(200, json=payload))


def test_exchange_code_parses():
    payload = {"access_token": "a", "refresh_token": "r", "expires_in": 1200}
    out = asyncio.run(auth.exchange_code("code", "verifier", transport=_token_transport(payload)))
    assert out["refresh_token"] == "r"


def test_token_request_raises_on_error():
    transport = httpx.MockTransport(lambda req: httpx.Response(400, text="bad_request"))
    with pytest.raises(auth.AuthError):
        asyncio.run(auth.refresh_tokens("r", transport=transport))


def test_get_auth_unknown_character_raises(tmp_store):
    with pytest.raises(auth.NotAuthenticated):
        asyncio.run(auth.get_auth(123))


def test_get_auth_uses_unexpired_token(tmp_store):
    auth._save_character(_char(9, "Z", access_token="still-good", expires_at=10_000))
    token, cid = asyncio.run(auth.get_auth(9, now=5_000))
    assert token == "still-good"
    assert cid == 9


def test_get_auth_refreshes_when_expired(tmp_store):
    auth._save_character(_char(9, "Z", refresh_token="r0", expires_at=1_000))
    new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Z"})
    payload = {"access_token": new_access, "refresh_token": "r1", "expires_in": 1200}
    token, cid = asyncio.run(auth.get_auth(9, now=2_000, transport=_token_transport(payload)))
    assert token == new_access
    assert cid == 9
    assert auth.load_store()["characters"]["9"]["refresh_token"] == "r1"


def test_status_lists_characters(tmp_store):
    assert auth.status() == {"authenticated": False, "characters": []}
    auth._save_character(_char(3, "Y"))
    info = auth.status()
    assert info["authenticated"] is True
    assert info["characters"][0]["character_id"] == 3