from __future__ import annotations

import json
from pathlib import Path

import pytest

from claude_oauth import (
    CLAUDE_CODE_CLIENT_ID,
    TOKEN_URL,
    USAGE_URL,
    USER_AGENT,
    ClaudeOAuthAuthError,
    ClaudeOAuthCredentialsError,
    ClaudeOAuthLoggedOutError,
    ClaudeOAuthProtocolError,
    ClaudeOAuthRequest,
    ClaudeOAuthResponse,
    ClaudeOAuthSession,
    ClaudeOAuthTimeoutError,
    ClaudeOAuthTransportError,
)


class _FakeTransport:
    def __init__(self, steps: list[object]) -> None:
        self._steps = list(steps)
        self.requests: list[tuple[ClaudeOAuthRequest, float]] = []

    def __call__(self, request: ClaudeOAuthRequest, timeout_secs: float) -> ClaudeOAuthResponse:
        self.requests.append((request, timeout_secs))
        if not self._steps:
            raise AssertionError("unexpected request")
        step = self._steps.pop(0)
        if isinstance(step, Exception):
            raise step
        assert isinstance(step, ClaudeOAuthResponse)
        return step


def _write_credentials(
    path: Path,
    *,
    access_token: str = "access-old",
    refresh_token: str = "refresh-old",
    expires_at: int = 4_100_000_000_000,
) -> None:
    payload: dict[str, object] = {
        "claudeAiOauth": {
            "accessToken": access_token,
            "refreshToken": refresh_token,
            "expiresAt": expires_at,
            "scopes": ["org:read"],
            "subscriptionType": "pro",
            "rateLimitTier": "team",
        }
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def _json_response(status_code: int, payload: object) -> ClaudeOAuthResponse:
    return ClaudeOAuthResponse(status_code=status_code, body=json.dumps(payload))


def test_get_usage_returns_usage(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport([_json_response(200, {"used": 12, "resetAt": "soon"})])

    usage = ClaudeOAuthSession(credentials_path, transport=transport, timeout_secs=2.5).get_usage()

    assert usage == {"used": 12, "resetAt": "soon"}
    assert len(transport.requests) == 1
    request, timeout_secs = transport.requests[0]
    assert timeout_secs == 2.5
    assert request.method == "GET"
    assert request.url == USAGE_URL
    assert request.body is None
    assert request.headers["Authorization"] == "Bearer access-old"
    assert request.headers["anthropic-beta"] == "oauth-2025-04-20"
    assert request.headers["User-Agent"] == USER_AGENT


def test_get_usage_exposes_retry_after_for_rate_limit_response(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport(
        [
            ClaudeOAuthResponse(
                status_code=429,
                body='{"error":"rate_limited"}',
                headers={"Retry-After": "442"},
            )
        ]
    )

    with pytest.raises(ClaudeOAuthTransportError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert exc_info.value.retry_after_seconds == 442.0


def test_session_never_refreshes_or_rewrites_credentials(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    before = credentials_path.read_text(encoding="utf-8")
    transport = _FakeTransport([_json_response(401, {"error": "expired"})])

    with pytest.raises(ClaudeOAuthTransportError):
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert len(transport.requests) == 1
    assert credentials_path.read_text(encoding="utf-8") == before
    assert list(tmp_path.iterdir()) == [credentials_path]


def test_expired_token_reloads_from_disk_without_a_network_call(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, access_token="access-stale", expires_at=1_700_000_000_000)
    transport = _FakeTransport([_json_response(200, {"used": 7})])
    session = ClaudeOAuthSession(credentials_path, transport=transport)
    _write_credentials(credentials_path, access_token="access-rotated")

    assert session.get_usage() == {"used": 7}
    assert [request.url for request, _timeout in transport.requests] == [USAGE_URL]
    assert transport.requests[0][0].headers["Authorization"] == "Bearer access-rotated"


def test_expired_token_with_no_newer_file_fails_without_a_network_call(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    before = credentials_path.read_text(encoding="utf-8")
    transport = _FakeTransport([])

    with pytest.raises(ClaudeOAuthTransportError):
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert transport.requests == []
    assert credentials_path.read_text(encoding="utf-8") == before


def test_session_reuses_credentials_rotated_by_the_cli_after_401(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    usage = {"five_hour": {"utilization": 10}}

    class _RotatingTransport(_FakeTransport):
        def __call__(self, request: ClaudeOAuthRequest, timeout_secs: float):
            if not self.requests:
                _write_credentials(credentials_path, access_token="access-rotated")
            return super().__call__(request, timeout_secs)

    transport = _RotatingTransport(
        [_json_response(401, {"error": "expired"}), _json_response(200, usage)]
    )
    session = ClaudeOAuthSession(credentials_path, transport=transport)

    assert session.get_usage() == usage
    assert transport.requests[1][0].headers["Authorization"] == "Bearer access-rotated"


def test_401_on_the_latest_on_disk_token_reports_logged_out(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)

    class _RotatingTransport(_FakeTransport):
        def __call__(self, request: ClaudeOAuthRequest, timeout_secs: float):
            if not self.requests:
                _write_credentials(credentials_path, access_token="access-rotated")
            return super().__call__(request, timeout_secs)

    transport = _RotatingTransport(
        [_json_response(401, {"error": "expired"}), _json_response(401, {"error": "revoked"})]
    )

    with pytest.raises(ClaudeOAuthLoggedOutError):
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert [request.url for request, _timeout in transport.requests] == [USAGE_URL, USAGE_URL]


def test_get_usage_does_not_report_403_as_logged_out(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport([_json_response(403, {"error": "forbidden"})])

    with pytest.raises(ClaudeOAuthAuthError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert not isinstance(exc_info.value, ClaudeOAuthLoggedOutError)
    assert [request.url for request, _timeout in transport.requests] == [USAGE_URL]


def test_error_messages_never_leak_tokens(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(
        credentials_path,
        access_token="secret-access-token",
        refresh_token="secret-refresh-token",
        expires_at=1_700_000_000_000,
    )

    with pytest.raises(ClaudeOAuthTransportError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=_FakeTransport([])).get_usage()

    message = str(exc_info.value)
    assert "secret-access-token" not in message
    assert "secret-refresh-token" not in message


def test_get_usage_raises_protocol_error_for_non_object_usage_payload(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport([_json_response(200, ["not", "a", "dict"])])

    with pytest.raises(ClaudeOAuthProtocolError):
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert [request.url for request, _timeout in transport.requests] == [USAGE_URL]


@pytest.mark.parametrize(
    ("writer", "expected_fragment"),
    [
        (None, "missing"),
        ("not-json", "valid JSON object"),
        ({"claudeAiOauth": {"accessToken": "only-access"}}, "refreshToken"),
    ],
)
def test_get_usage_raises_credentials_error_for_invalid_credentials_file(
    tmp_path: Path,
    writer: object,
    expected_fragment: str,
) -> None:
    credentials_path = tmp_path / ".credentials.json"
    if writer == "not-json":
        credentials_path.write_text("{", encoding="utf-8")
    elif isinstance(writer, dict):
        credentials_path.write_text(json.dumps(writer), encoding="utf-8")

    with pytest.raises(ClaudeOAuthCredentialsError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=_FakeTransport([])).get_usage()

    assert expected_fragment in str(exc_info.value)


def test_get_usage_raises_timeout_error_for_transport_timeout(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport([TimeoutError("secret transport timeout")])

    with pytest.raises(ClaudeOAuthTimeoutError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert "secret transport timeout" not in str(exc_info.value)


def test_get_usage_raises_transport_error_for_transport_failure(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport([OSError("secret socket issue")])

    with pytest.raises(ClaudeOAuthTransportError) as exc_info:
        ClaudeOAuthSession(credentials_path, transport=transport).get_usage()

    assert "secret socket issue" not in str(exc_info.value)


def test_get_usage_reading_reports_the_token_that_fetched_the_payload(
    tmp_path: Path,
) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)

    class _RotatingTransport(_FakeTransport):
        def __call__(self, request: ClaudeOAuthRequest, timeout_secs: float):
            if not self.requests:
                _write_credentials(credentials_path, access_token="access-rotated")
            return super().__call__(request, timeout_secs)

    transport = _RotatingTransport(
        [_json_response(401, {}), _json_response(200, {"five_hour": {"percent": 12}})]
    )

    reading = ClaudeOAuthSession(credentials_path, transport=transport).get_usage_reading()

    assert reading.payload == {"five_hour": {"percent": 12}}
    assert reading.access_token == "access-rotated"


def _grant_response(
    access_token: str = "access-new",
    refresh_token: str = "refresh-new",
    expires_in: int = 28_800,
) -> ClaudeOAuthResponse:
    return _json_response(
        200,
        {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expires_in": expires_in,
        },
    )


def test_rotation_refreshes_and_persists_the_grant(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    credentials_path.chmod(0o600)
    transport = _FakeTransport([_grant_response(), _json_response(200, {"used": 3})])
    session = ClaudeOAuthSession(credentials_path, transport=transport, allow_rotation=True)

    reading = session.get_usage_reading()

    assert reading.payload == {"used": 3}
    assert reading.access_token == "access-new"
    refresh_request = transport.requests[0][0]
    assert refresh_request.method == "POST"
    assert refresh_request.url == TOKEN_URL
    assert refresh_request.body is not None
    body = refresh_request.body.decode("utf-8")
    assert "grant_type=refresh_token" in body
    assert "refresh_token=refresh-old" in body
    assert CLAUDE_CODE_CLIENT_ID in body
    assert transport.requests[1][0].headers["Authorization"] == "Bearer access-new"
    on_disk = json.loads(credentials_path.read_text(encoding="utf-8"))["claudeAiOauth"]
    assert on_disk["accessToken"] == "access-new"
    assert on_disk["refreshToken"] == "refresh-new"
    assert on_disk["expiresAt"] > 1_700_000_000_000
    assert on_disk["scopes"] == ["org:read"]
    assert on_disk["subscriptionType"] == "pro"
    assert credentials_path.stat().st_mode & 0o777 == 0o600
    leftovers = {p.name for p in tmp_path.iterdir()} - {
        ".credentials.json",
        ".credentials.json.rotate.lock",
    }
    assert leftovers == set()


def test_rotation_disallowed_by_default(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    before = credentials_path.read_text(encoding="utf-8")

    with pytest.raises(ClaudeOAuthTransportError):
        ClaudeOAuthSession(credentials_path, transport=_FakeTransport([])).get_usage()

    assert credentials_path.read_text(encoding="utf-8") == before


def test_rotation_invalid_grant_reports_logged_out_and_leaves_file_alone(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    before = credentials_path.read_text(encoding="utf-8")
    transport = _FakeTransport([_json_response(400, {"error": "invalid_grant"})])

    with pytest.raises(ClaudeOAuthLoggedOutError):
        ClaudeOAuthSession(
            credentials_path, transport=transport, allow_rotation=True
        ).get_usage()

    assert credentials_path.read_text(encoding="utf-8") == before


def test_rotation_transient_refresh_failure_is_a_transport_error(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    transport = _FakeTransport(
        [
            ClaudeOAuthResponse(
                status_code=503,
                body='{"error":"temporarily_unavailable"}',
                headers={"Retry-After": "120"},
            )
        ]
    )

    with pytest.raises(ClaudeOAuthTransportError) as exc_info:
        ClaudeOAuthSession(
            credentials_path, transport=transport, allow_rotation=True
        ).get_usage()

    assert not isinstance(exc_info.value, ClaudeOAuthLoggedOutError)
    assert exc_info.value.retry_after_seconds == 120.0


def test_rotation_prefers_a_grant_rotated_by_someone_else_meanwhile(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    session = ClaudeOAuthSession(
        credentials_path,
        transport=_FakeTransport([_json_response(200, {"used": 8})]),
        allow_rotation=True,
    )
    original_load = session._load_credentials
    loads = {"count": 0}

    def rotate_before_second_load():
        loads["count"] += 1
        if loads["count"] == 2:
            _write_credentials(credentials_path, access_token="access-cli-rotated")
        return original_load()

    session._load_credentials = rotate_before_second_load

    assert session.get_usage() == {"used": 8}


def test_rotation_after_401_uses_the_new_grant(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path)
    transport = _FakeTransport(
        [
            _json_response(401, {"error": "expired"}),
            _grant_response(access_token="access-after-401"),
            _json_response(200, {"used": 9}),
        ]
    )

    reading = ClaudeOAuthSession(
        credentials_path, transport=transport, allow_rotation=True
    ).get_usage_reading()

    assert reading.access_token == "access-after-401"
    assert json.loads(credentials_path.read_text(encoding="utf-8"))["claudeAiOauth"][
        "accessToken"
    ] == "access-after-401"


def test_rotation_keeps_the_old_refresh_token_when_none_is_returned(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    transport = _FakeTransport(
        [
            _json_response(200, {"access_token": "access-new", "expires_in": 28_800}),
            _json_response(200, {"used": 1}),
        ]
    )

    ClaudeOAuthSession(credentials_path, transport=transport, allow_rotation=True).get_usage()

    on_disk = json.loads(credentials_path.read_text(encoding="utf-8"))["claudeAiOauth"]
    assert on_disk["refreshToken"] == "refresh-old"


def test_rotation_journal_replay_recovers_an_orphaned_grant_without_a_request(
    tmp_path: Path,
) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    journal = tmp_path / ".credentials.json.rotate-journal.json"
    journal.write_text(
        json.dumps(
            {
                "claudeAiOauth": {
                    "accessToken": "access-journaled",
                    "refreshToken": "refresh-journaled",
                    "expiresAt": 4_100_000_000_000,
                    "scopes": ["org:read"],
                }
            }
        ),
        encoding="utf-8",
    )
    transport = _FakeTransport([_json_response(200, {"used": 9})])

    reading = ClaudeOAuthSession(
        credentials_path, transport=transport, allow_rotation=True
    ).get_usage_reading()

    assert reading.access_token == "access-journaled"
    on_disk = json.loads(credentials_path.read_text(encoding="utf-8"))["claudeAiOauth"]
    assert on_disk["refreshToken"] == "refresh-journaled"
    assert not journal.exists()
    # only the usage GET went out — no refresh token was spent
    assert len(transport.requests) == 1
    assert transport.requests[0][0].method == "GET"


def test_rotation_journal_replay_runs_even_when_rotation_is_disallowed(
    tmp_path: Path,
) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    journal = tmp_path / ".credentials.json.rotate-journal.json"
    journal.write_text(
        json.dumps(
            {
                "claudeAiOauth": {
                    "accessToken": "access-journaled",
                    "refreshToken": "refresh-journaled",
                    "expiresAt": 4_100_000_000_000,
                }
            }
        ),
        encoding="utf-8",
    )
    transport = _FakeTransport([_json_response(200, {"used": 9})])

    reading = ClaudeOAuthSession(credentials_path, transport=transport).get_usage_reading()

    assert reading.access_token == "access-journaled"
    assert not journal.exists()


def test_rotation_journal_matching_the_file_is_cleared_and_rotation_proceeds(
    tmp_path: Path,
) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    journal = tmp_path / ".credentials.json.rotate-journal.json"
    journal.write_text(credentials_path.read_text(encoding="utf-8"), encoding="utf-8")
    transport = _FakeTransport([_grant_response(), _json_response(200, {"used": 4})])

    reading = ClaudeOAuthSession(
        credentials_path, transport=transport, allow_rotation=True
    ).get_usage_reading()

    assert reading.access_token == "access-new"
    assert not journal.exists()


def test_unwritable_directory_refuses_to_spend_the_refresh_token(tmp_path: Path) -> None:
    directory = tmp_path / "home"
    credentials_path = directory / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    transport = _FakeTransport([])
    directory.chmod(0o500)
    try:
        with pytest.raises(ClaudeOAuthCredentialsError):
            ClaudeOAuthSession(
                credentials_path, transport=transport, allow_rotation=True
            ).get_usage()
    finally:
        directory.chmod(0o700)

    # the refresh POST never happened: the one-time token is intact on disk
    assert transport.requests == []
    on_disk = json.loads(credentials_path.read_text(encoding="utf-8"))["claudeAiOauth"]
    assert on_disk["refreshToken"] == "refresh-old"


def test_rotation_writes_a_restorable_backup_of_the_new_grant(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    backup_path = tmp_path / "rescue" / "acct.credentials.json.auto"
    transport = _FakeTransport([_grant_response(), _json_response(200, {"used": 3})])

    ClaudeOAuthSession(
        credentials_path,
        transport=transport,
        allow_rotation=True,
        rotation_backup_path=backup_path,
    ).get_usage()

    saved = json.loads(backup_path.read_text(encoding="utf-8"))["claudeAiOauth"]
    assert saved["accessToken"] == "access-new"
    assert saved["refreshToken"] == "refresh-new"
    assert backup_path.stat().st_mode & 0o777 == 0o600


def test_refresh_post_uses_the_generous_token_timeout(tmp_path: Path) -> None:
    credentials_path = tmp_path / ".credentials.json"
    _write_credentials(credentials_path, expires_at=1_700_000_000_000)
    transport = _FakeTransport([_grant_response(), _json_response(200, {"used": 3})])

    ClaudeOAuthSession(
        credentials_path, transport=transport, timeout_secs=5.0, allow_rotation=True
    ).get_usage()

    assert transport.requests[0][1] == 30.0
    assert transport.requests[1][1] == 5.0