from __future__ import annotations import gc from dataclasses import replace from pathlib import Path import pytest pytest.importorskip("fastapi.testclient") from fastapi import FastAPI from agency_swarm.integrations import openclaw as openclaw_mod from agency_swarm.integrations.openclaw import attach_openclaw_to_fastapi, build_openclaw_responses_model from agency_swarm.utils.model_utils import get_usage_tracking_model_name from tests.integration.fastapi._openclaw_test_support import ( _build_openclaw_config, reset_openclaw_current_app_defaults, ) @pytest.fixture(autouse=True) def _reset_openclaw_defaults(monkeypatch: pytest.MonkeyPatch) -> None: reset_openclaw_current_app_defaults(monkeypatch) def _clear_openclaw_env(monkeypatch: pytest.MonkeyPatch) -> None: for name in [ "APP_TOKEN", "OPENCLAW_DEFAULT_MODEL", "OPENCLAW_GATEWAY_TOKEN", "OPENCLAW_PROVIDER_MODEL", "OPENCLAW_PROXY_API_KEY", "OPENCLAW_PROXY_BASE_URL", "OPENCLAW_PROXY_HOST", "OPENCLAW_PROXY_PORT", "PORT", ]: monkeypatch.delenv(name, raising=False) def test_reset_openclaw_defaults_drains_stale_app_finalizers(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "http://127.0.0.1:8000/openclaw/v1") stale_app = FastAPI() attach_openclaw_to_fastapi( stale_app, replace( _build_openclaw_config(tmp_path / "stale"), default_model="openclaw:stale", provider_model="openai/gpt-4o", ), ) stale_app = None reset_openclaw_current_app_defaults(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "http://127.0.0.1:8000/openclaw/v1") first_app = FastAPI() attach_openclaw_to_fastapi( first_app, replace( _build_openclaw_config(tmp_path / "first"), default_model="openclaw:first", provider_model="openai/gpt-4o", ), ) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "http://127.0.0.1:9000/openclaw/v1") second_app = FastAPI() attach_openclaw_to_fastapi( second_app, replace( _build_openclaw_config(tmp_path / "second"), port=9000, default_model="openclaw:second", provider_model="openai/gpt-5.4-mini", ), ) gc.collect() first_model = build_openclaw_responses_model(base_url="http://127.0.0.1:8000/openclaw/v1", api_key="first-token") second_model = build_openclaw_responses_model( base_url="http://127.0.0.1:9000/openclaw/v1", api_key="second-token", ) assert first_model.model == "openclaw:first" assert second_model.model == "openclaw:second" def test_build_openclaw_model_uses_current_app_proxy_defaults(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_PORT", "9000") monkeypatch.setenv("APP_TOKEN", "app-token") monkeypatch.setenv("OPENCLAW_PROXY_API_KEY", "proxy-token") monkeypatch.setenv("OPENCLAW_GATEWAY_TOKEN", "gateway-token") config = replace( _build_openclaw_config(tmp_path), port=9000, default_model="openclaw:custom", provider_model="anthropic/claude-sonnet-4-5", ) attach_openclaw_to_fastapi(FastAPI(), config) model = build_openclaw_responses_model(base_url="http://127.0.0.1:9000/openclaw/v1") assert model.model == "openclaw:custom" assert get_usage_tracking_model_name(model) == "anthropic/claude-sonnet-4-5" assert model._client.api_key == "app-token" @pytest.mark.parametrize( ("server_url", "base_url"), [ ("http://localhost:9000", "http://127.0.0.1:9000/openclaw/v1"), ("https://example.com/{stage}", "https://example.com/prod/openclaw/v1"), ], ) def test_attach_openclaw_uses_app_server_urls_for_current_app_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, server_url: str, base_url: str, ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("APP_TOKEN", "app-token") monkeypatch.setenv("OPENCLAW_PROXY_API_KEY", "proxy-token") monkeypatch.setenv("OPENCLAW_GATEWAY_TOKEN", "gateway-token") app = FastAPI(servers=[{"url": server_url}]) attach_openclaw_to_fastapi( app, replace( _build_openclaw_config(tmp_path), default_model="openclaw:custom", provider_model="anthropic/claude-sonnet-4-5", ), ) model = build_openclaw_responses_model(base_url=base_url) assert model.model == "openclaw:custom" assert get_usage_tracking_model_name(model) == "anthropic/claude-sonnet-4-5" assert model._client.api_key == "app-token" def test_attach_openclaw_ignores_relative_server_urls_for_current_app_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("APP_TOKEN", "app-token") monkeypatch.setenv("OPENCLAW_PROXY_API_KEY", "proxy-token") app = FastAPI(servers=[{"url": "/api"}]) attach_openclaw_to_fastapi(app, _build_openclaw_config(tmp_path)) model = build_openclaw_responses_model(base_url="https://example.com/api/openclaw/v1") assert model.model == "openclaw:main" assert get_usage_tracking_model_name(model) == "openclaw:main" assert model._client.api_key == "proxy-token" def test_attach_openclaw_rejects_conflicting_current_app_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_PORT", "8000") attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path / "first"), default_model="openclaw:first", provider_model="openai/gpt-4o", ), ) with pytest.raises(ValueError, match="Conflicting current-app OpenClaw defaults"): attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path / "second"), default_model="openclaw:second", provider_model="openai/gpt-5.4-mini", ), ) def test_attach_openclaw_rolls_back_partial_registration_failures( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_PORT", "9000") conflict_url = "https://conflict.example/openclaw/v1" openclaw_mod.openclaw_model.register_current_app_openclaw_defaults( default_model="openclaw:existing", provider_model="openai/gpt-4o", base_url=conflict_url, ) with pytest.raises(ValueError, match="Conflicting"): attach_openclaw_to_fastapi( FastAPI(servers=[{"url": "https://conflict.example"}]), replace( _build_openclaw_config(tmp_path), default_model="openclaw:new", provider_model="openai/gpt-5.4-mini", ), ) default_url_key = openclaw_mod.openclaw_model._normalize_openclaw_proxy_url("http://127.0.0.1:9000/openclaw/v1") assert default_url_key not in openclaw_mod.openclaw_model._CURRENT_APP_OPENCLAW_DEFAULTS assert default_url_key not in openclaw_mod.openclaw_model._CURRENT_APP_OPENCLAW_DEFAULT_COUNTS def test_attach_openclaw_releases_defaults_when_app_is_garbage_collected( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_PORT", "8000") attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path / "first"), default_model="openclaw:first", provider_model="openai/gpt-4o", ), ) gc.collect() attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path / "second"), default_model="openclaw:second", provider_model="openai/gpt-5.4-mini", ), ) def test_attach_openclaw_keeps_distinct_proxy_defaults_separate( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "http://127.0.0.1:8000/openclaw/v1") first_app = FastAPI() second_app = FastAPI() attach_openclaw_to_fastapi( first_app, replace( _build_openclaw_config(tmp_path / "first"), default_model="openclaw:first", provider_model="openai/gpt-4o", ), ) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "http://127.0.0.1:9000/openclaw/v1") attach_openclaw_to_fastapi( second_app, replace( _build_openclaw_config(tmp_path / "second"), port=9000, default_model="openclaw:second", provider_model="openai/gpt-5.4-mini", ), ) first_model = build_openclaw_responses_model(base_url="http://127.0.0.1:8000/openclaw/v1", api_key="first-token") second_model = build_openclaw_responses_model( base_url="http://127.0.0.1:9000/openclaw/v1", api_key="second-token", ) assert first_model.model == "openclaw:first" assert second_model.model == "openclaw:second" def test_attach_openclaw_uses_public_proxy_defaults_for_exact_current_app_url( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "https://worker.example/openclaw/v1") monkeypatch.setenv("APP_TOKEN", "app-token") monkeypatch.setenv("OPENCLAW_PROXY_API_KEY", "proxy-token") attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path), default_model="openclaw:custom", provider_model="openai/gpt-5.4-mini", ), ) model = build_openclaw_responses_model(base_url="https://worker.example/openclaw/v1") assert model.model == "openclaw:custom" assert get_usage_tracking_model_name(model) == "openai/gpt-5.4-mini" assert model._client.api_key == "app-token" def test_attach_openclaw_does_not_reuse_public_proxy_defaults_for_other_urls( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _clear_openclaw_env(monkeypatch) monkeypatch.setenv("OPENCLAW_PROXY_BASE_URL", "https://external.example/openclaw/v1") monkeypatch.setenv("OPENCLAW_PROVIDER_MODEL", "openai/gpt-5.4-mini") monkeypatch.setenv("APP_TOKEN", "app-token") monkeypatch.setenv("OPENCLAW_PROXY_API_KEY", "proxy-token") attach_openclaw_to_fastapi( FastAPI(), replace( _build_openclaw_config(tmp_path), default_model="openclaw:custom", provider_model="openai/gpt-4o", ), ) model = build_openclaw_responses_model(base_url="https://other.example/openclaw/v1") assert model.model == "openclaw:main" assert get_usage_tracking_model_name(model) == "openclaw:main" assert model._client.api_key == "proxy-token"