tests

tests/test_combined_app.py

@@ -0,0 +1,46 @@
+import importlib
+import os
+import sys
+
+from fastapi.testclient import TestClient
+
+os.environ.setdefault("CONN_STR", "postgresql://dummy:dummy@localhost/dummy")
+
+
+def test_server_dashscope_import_is_cli_safe(monkeypatch):
+    """
+    Importing server_dashscope should not invoke tyro.cli at module import time.
+    """
+    import tyro
+
+    monkeypatch.setattr(
+        tyro,
+        "cli",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("tyro.cli must not run during module import")
+        ),
+    )
+    sys.modules.pop("fastapi_server.server_dashscope", None)
+
+    module = importlib.import_module("fastapi_server.server_dashscope")
+    assert module.app is not None
+    assert module.dashscope_router is not None
+
+
+def test_combined_app_serves_front_and_dashscope_routes():
+    from fastapi_server.combined import app
+
+    client = TestClient(app)
+
+    # front_apis route should be available.
+    front_resp = client.get("/v1/pipelines/graphs")
+    assert front_resp.status_code == 200, front_resp.text
+    assert "available_graphs" in front_resp.json()
+
+    # DashScope route should exist at the same path (missing auth should not be 404).
+    dash_resp = client.post(
+        "/api/v1/apps/blueberry/sessions/test-session/responses",
+        json={"input": {"prompt": "hello"}, "stream": False},
+    )
+    assert dash_resp.status_code != 404, dash_resp.text
+

tests/test_server_pipeline_manager_refresh.py

@@ -0,0 +1,153 @@
+import json
+import time
+
+import pytest
+from fastapi import HTTPException
+
+from lang_agent.components.server_pipeline_manager import ServerPipelineManager
+
+
+class _DummyPipeline:
+    def __init__(self, model: str):
+        self.model = model
+
+
+class _DummyConfig:
+    def __init__(self, llm_name: str = "qwen-plus"):
+        self.llm_name = llm_name
+
+    def setup(self):
+        return _DummyPipeline(model=self.llm_name)
+
+
+def _write_registry(path, pipelines, api_keys=None):
+    content = {"pipelines": pipelines, "api_keys": api_keys or {}}
+    path.write_text(json.dumps(content, indent=2), encoding="utf-8")
+    # Ensure mtime changes reliably on fast CI filesystems.
+    time.sleep(0.01)
+
+
+def test_refresh_registry_picks_up_new_pipeline(tmp_path):
+    registry_path = tmp_path / "pipeline_registry.json"
+    _write_registry(
+        registry_path,
+        pipelines={
+            "default": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-plus"},
+            }
+        },
+    )
+    manager = ServerPipelineManager(
+        default_pipeline_id="default",
+        default_config=_DummyConfig(),
+    )
+    manager.load_registry(str(registry_path))
+
+    with pytest.raises(HTTPException) as exc_info:
+        manager.resolve_pipeline_id(
+            body={"pipeline_id": "blueberry"}, app_id=None, api_key="k1"
+        )
+    assert exc_info.value.status_code == 404
+
+    _write_registry(
+        registry_path,
+        pipelines={
+            "default": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-plus"},
+            },
+            "blueberry": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-max"},
+            },
+        },
+    )
+    changed = manager.refresh_registry_if_needed()
+    assert changed is True
+
+    resolved = manager.resolve_pipeline_id(
+        body={"pipeline_id": "blueberry"}, app_id=None, api_key="k1"
+    )
+    assert resolved == "blueberry"
+
+
+def test_refresh_registry_invalidates_cache_for_changed_pipeline(tmp_path):
+    registry_path = tmp_path / "pipeline_registry.json"
+    _write_registry(
+        registry_path,
+        pipelines={
+            "blueberry": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-plus"},
+            }
+        },
+    )
+    manager = ServerPipelineManager(
+        default_pipeline_id="blueberry",
+        default_config=_DummyConfig(),
+    )
+    manager.load_registry(str(registry_path))
+
+    first_pipeline, first_model = manager.get_pipeline("blueberry")
+    assert first_model == "qwen-plus"
+
+    _write_registry(
+        registry_path,
+        pipelines={
+            "blueberry": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-max"},
+            }
+        },
+    )
+    changed = manager.refresh_registry_if_needed()
+    assert changed is True
+
+    second_pipeline, second_model = manager.get_pipeline("blueberry")
+    assert second_model == "qwen-max"
+    assert second_pipeline is not first_pipeline
+
+
+def test_refresh_registry_applies_disabled_state_immediately(tmp_path):
+    registry_path = tmp_path / "pipeline_registry.json"
+    _write_registry(
+        registry_path,
+        pipelines={
+            "blueberry": {
+                "enabled": True,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-plus"},
+            }
+        },
+    )
+    manager = ServerPipelineManager(
+        default_pipeline_id="blueberry",
+        default_config=_DummyConfig(),
+    )
+    manager.load_registry(str(registry_path))
+    manager.get_pipeline("blueberry")
+
+    _write_registry(
+        registry_path,
+        pipelines={
+            "blueberry": {
+                "enabled": False,
+                "config_file": None,
+                "overrides": {"llm_name": "qwen-plus"},
+            }
+        },
+    )
+    changed = manager.refresh_registry_if_needed()
+    assert changed is True
+
+    with pytest.raises(HTTPException) as exc_info:
+        manager.get_pipeline("blueberry")
+    assert exc_info.value.status_code == 403
+
+