VCR Proxy

import httpx

async with httpx.AsyncClient(base_url="http://localhost:8080") as client:
    # /api/* → api.example.com
    users = await client.get("/api/v1/users")

    # /auth/* → auth.example.com
    token = await client.post("/auth/oauth/token", data={...})

import os
API_URL = os.getenv("API_BASE_URL", "https://api.example.com")

# conftest.py
from pathlib import Path
from collections.abc import AsyncIterator

import httpx
import pytest

from vcr_proxy.app import create_app


@pytest.fixture
async def api_client(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]:
    """HTTP client that replays from cassettes — no network required."""
    app = create_app(
        cassettes_dir=Path("cassettes"),  # committed cassettes
        mode="replay",
        targets={"/api": "https://api.example.com"},
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

# test_users.py
async def test_list_users(api_client: httpx.AsyncClient):
    response = await api_client.get("/api/v1/users")
    assert response.status_code == 200
    assert len(response.json()) > 0


async def test_create_user(api_client: httpx.AsyncClient):
    response = await api_client.post(
        "/api/v1/users",
        json={"name": "Alice"},
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Alice"

@pytest.fixture
async def api_client(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]:
    app = create_app(
        cassettes_dir=Path("cassettes"),
        mode="spy",  # replay hits, record misses
        targets={"/api": "https://api.example.com"},
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

app = create_app(
    cassettes_dir=Path("cassettes"),
    mode="replay",
    targets={
        "/api": "https://api.example.com",
        "/auth": "https://auth.example.com",
        "/payments": "https://payments.stripe.com",
    },
)

@pytest.fixture
async def isolated_client(tmp_path: Path) -> AsyncIterator[httpx.AsyncClient]:
    app = create_app(
        cassettes_dir=tmp_path / "cassettes",
        mode="record",
        targets={"/api": "https://api.example.com"},
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

# conftest.py
from pathlib import Path

import pytest

from vcr_proxy.config import Settings
from vcr_proxy.forward import VCRAddon
from vcr_proxy.models import ProxyMode


@pytest.fixture
def vcr_addon(tmp_path: Path) -> VCRAddon:
    settings = Settings(
        mode=ProxyMode.SPY,
        cassettes_dir=tmp_path / "cassettes",
    )
    return VCRAddon(settings)

# test_forward.py
from mitmproxy.test import tflow, tutils


def make_flow(host, path, method="GET", body=b"", headers=None):
    """Create a mitmproxy flow for testing."""
    hdrs = [(b"host", host.encode())]
    for k, v in (headers or {}).items():
        hdrs.append((k.encode(), v.encode()))
    return tflow.tflow(
        req=tutils.treq(
            method=method.encode(),
            host=host,
            port=443,
            scheme=b"https",
            authority=host.encode(),
            path=path.encode(),
            headers=hdrs,
            content=body,
        )
    )


def test_record_and_replay(vcr_addon):
    # Record a request
    flow = make_flow("api.example.com", "/v1/users")
    vcr_addon.request(flow)

    # Simulate upstream response
    flow.response = tutils.tresp(
        content=b'[{"id": 1}]',
        status_code=200,
    )
    flow.response.headers.clear()
    flow.response.headers["content-type"] = "application/json"
    vcr_addon.response(flow)

    # Switch to replay
    vcr_addon.mode = "replay"

    # Same request now returns from cache
    replay_flow = make_flow("api.example.com", "/v1/users")
    vcr_addon.request(replay_flow)
    assert replay_flow.response.status_code == 200
    assert b'"id": 1' in replay_flow.response.content