basicmachines-co-basic-memory/tests/cli/test_auth_cli_auth.py

import json
import os
import stat
import time
from contextlib import asynccontextmanager

import httpx
import pytest

from basic_memory.cli.auth import CLIAuth


def _make_mock_transport(handler):
    return httpx.MockTransport(handler)


@pytest.mark.asyncio
async def test_cli_auth_request_device_authorization_uses_injected_http_client(
    tmp_path, monkeypatch
):
    """Integration-style test: exercise the request flow with real httpx plumbing (MockTransport)."""
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_ENV", "test")

    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path.endswith("/oauth2/device_authorization")
        body = (await request.aread()).decode()
        # sanity: client_id should be in form data
        assert "client_id=test-client-id" in body
        return httpx.Response(
            200,
            json={
                "device_code": "devcode",
                "user_code": "usercode",
                "verification_uri": "https://example.test/verify",
                "interval": 1,
            },
        )

    transport = _make_mock_transport(handler)

    @asynccontextmanager
    async def client_factory():
        async with httpx.AsyncClient(transport=transport) as client:
            yield client

    auth = CLIAuth(
        client_id="test-client-id",
        authkit_domain="https://example.test",
        http_client_factory=client_factory,
    )

    result = await auth.request_device_authorization()
    assert result is not None
    assert result["device_code"] == "devcode"


def test_cli_auth_generate_pkce_pair_format(tmp_path, monkeypatch):
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_ENV", "test")

    auth = CLIAuth(client_id="cid", authkit_domain="https://example.test")
    verifier, challenge = auth.generate_pkce_pair()

    # PKCE verifier/challenge should be URL-safe base64 without padding.
    assert verifier
    assert challenge
    assert "=" not in verifier
    assert "=" not in challenge
    # code verifier length should be in recommended bounds (rough sanity).
    assert 43 <= len(verifier) <= 128


@pytest.mark.asyncio
async def test_cli_auth_save_load_and_get_valid_token_roundtrip(tmp_path, monkeypatch):
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_ENV", "test")

    auth = CLIAuth(client_id="cid", authkit_domain="https://example.test")

    tokens = {
        "access_token": "at",
        "refresh_token": "rt",
        "expires_in": 3600,
        "token_type": "Bearer",
    }
    auth.save_tokens(tokens)

    loaded = auth.load_tokens()
    assert loaded is not None
    assert loaded["access_token"] == "at"
    assert loaded["refresh_token"] == "rt"
    assert auth.is_token_valid(loaded) is True

    valid = await auth.get_valid_token()
    assert valid == "at"

    # Permission should be 600 on POSIX systems
    if os.name != "nt":
        mode = auth.token_file.stat().st_mode
        assert stat.S_IMODE(mode) == 0o600


@pytest.mark.asyncio
async def test_cli_auth_refresh_flow_uses_injected_http_client(tmp_path, monkeypatch):
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_ENV", "test")

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path.endswith("/oauth2/token"):
            body = (await request.aread()).decode()
            assert "grant_type=refresh_token" in body
            return httpx.Response(
                200,
                json={
                    "access_token": "new-at",
                    "refresh_token": "new-rt",
                    "expires_in": 3600,
                    "token_type": "Bearer",
                },
            )
        raise AssertionError(f"Unexpected request: {request.method} {request.url}")

    transport = _make_mock_transport(handler)

    @asynccontextmanager
    async def client_factory():
        async with httpx.AsyncClient(transport=transport) as client:
            yield client

    auth = CLIAuth(
        client_id="cid",
        authkit_domain="https://example.test",
        http_client_factory=client_factory,
    )

    # Write an expired token file manually (so we control expires_at precisely).
    auth.token_file.parent.mkdir(parents=True, exist_ok=True)
    auth.token_file.write_text(
        json.dumps(
            {
                "access_token": "old-at",
                "refresh_token": "old-rt",
                "expires_at": int(time.time()) - 10,
                "token_type": "Bearer",
            }
        ),
        encoding="utf-8",
    )

    token = await auth.get_valid_token()
    assert token == "new-at"