Add E2E tests for hermes_cli adapter

.agents/skills/python-tests/SKILL.md

@@ -0,0 +1,42 @@
+---
+name: python-tests
+description: Python tests for Fabric; use this when writing tests
+author: NVIDIA Corporation and Affiliates
+license: Apache-2.0
+---
+
+
+# Python Test Style
+
+- Pytest is used to run tests.
+- Do not add `@pytest.mark.asyncio` to any test. Async tests are automatically detected and run by the async runner; the decorator is unnecessary clutter.
+- Do not add a `-> None` return type annotation to test functions. This is not a common convention in pytest and adds unnecessary verbosity.
+- When mocking a class, do not define a new class. Use `unittest.mock.MagicMock` or `unittest.mock.AsyncMock`, with the `spec` constructor argument when necessary.
+- The name of the mocked class should be prefixed with `mock`, not `fake`.
+- Prefer pytest fixtures over helper methods.
+- Do not repeat fixtures, if a fixture is needed in multiple test files, place it in a `conftest.py` file.
+- When creating a fixture follow this pattern:
+  ```python
+  @pytest.fixture(name="<fixture_name>"[, scope="<scope>"])
+  def <fixture_name>_fixture() -> <return_type>:
+      ...
+  ```
+  Only specify the scope argument when the value is something other than "function".
+- Prefer `pytest.mark.parametrize` over creating individual tests for
+  different input types.
+- If a fixture is needed for a test, but either does not return a value or the value is not used in the test, use the `@pytest.mark.usefixtures` decorator.
+
+## Common Commands
+
+```bash
+# Focused test loop
+uv run pytest -k "<pattern>"
+
+# Run all tests
+uv run pytest
+```
+
+## References
+
+- `pyproject.toml`
+- `tests/conftest.py`

.github/workflows/ci_python.yml

@@ -85,5 +85,5 @@ jobs:
       - name: Run pytest
         run: |
           set -euo pipefail
-          uv sync --group test --no-group dev --extra hermes
+          uv sync --group test --no-group dev --extra hermes --extra relay
           uv run pytest

pyproject.toml

@@ -23,10 +23,12 @@ dev = [
 ]
 
 test = [
+    "fastapi~=0.138",
     "pytest>=8",
     "pytest-asyncio>=0.26",
     "pytest-cov~=7.0",
     "pyyaml>=6.0",
+    "uvicorn~=0.49",
 ]
 
 [tool.uv]

tests/_utils/mock_api_server.py

@@ -0,0 +1,191 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+import threading
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+import uvicorn
+
+
+@contextmanager
+def mock_api_server(port: int) -> Iterator[str]:
+    """
+    Context manager for a mock API server.
+
+    Use the /_requests endpoint to inspect captured chat-completion payloads after a test action.
+    Use the /_scenario endpoint to configure the server to return a specific status code for subsequent requests.
+
+    Args:
+        port (int): The port on which the server will listen.
+
+    Yields:
+        str: The base URL of the mock API server.
+    """
+
+    app = FastAPI()
+    app.state.requests = []
+    app.state.status_code = 200
+
+    @app.get("/health")
+    def health() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get("/v1/models")
+    def models() -> dict[str, object]:
+        return {
+            "object": "list",
+            "data": [
+                {
+                    "id": "fabric-echo",
+                    "object": "model",
+                    "created": 0,
+                    "owned_by": "fabric-test",
+                }
+            ],
+        }
+
+    @app.get("/_requests")
+    def requests() -> list[dict[str, object]]:
+        """GET this after a test action to inspect captured chat-completion payloads."""
+
+        return list(app.state.requests)
+
+    @app.post("/_scenario")
+    async def scenario(request: Request) -> dict[str, int]:
+        """POST JSON such as {"status_code": 429} before a test action to change responses."""
+
+        payload = await request.json()
+        app.state.status_code = int(payload.get("status_code", 200))
+        return {"status_code": app.state.status_code}
+
+    @app.post("/v1/chat/completions")
+    async def chat_completions(request: Request):
+        payload = await request.json()
+        app.state.requests.append(payload)
+        if app.state.status_code != 200:
+            return JSONResponse(
+                status_code=app.state.status_code,
+                content={
+                    "error": {
+                        "message": f"configured status {app.state.status_code}",
+                        "type": "api_error",
+                    }
+                },
+            )
+
+        messages = payload.get("messages") or []
+        user_messages = [
+            message
+            for message in messages
+            if isinstance(message, dict) and message.get("role") == "user"
+        ]
+        latest = user_messages[-1].get("content", "") if user_messages else ""
+        content = f"echo user_count={len(user_messages)} latest={latest}"
+        if payload.get("stream") is True:
+            return StreamingResponse(
+                _stream_chat_completion(payload, content),
+                media_type="text/event-stream",
+            )
+
+        return JSONResponse(
+            {
+                "id": "chatcmpl-fabric-test",
+                "object": "chat.completion",
+                "created": 0,
+                "model": payload.get("model", "fabric-echo"),
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": content},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {
+                    "prompt_tokens": 0,
+                    "completion_tokens": 0,
+                    "total_tokens": 0,
+                },
+            }
+        )
+
+    base_url = f"http://127.0.0.1:{port}"
+    config = uvicorn.Config(
+        app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+        access_log=False,
+        lifespan="off",
+        ws="none",
+    )
+    server = uvicorn.Server(config)
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+
+    deadline = time.monotonic() + 5
+    while not server.started:
+        if not thread.is_alive():
+            raise RuntimeError("mock API server failed to start")
+        if time.monotonic() > deadline:
+            raise RuntimeError("mock API server did not start within 5 seconds")
+        time.sleep(0.01)
+
+    try:
+        yield base_url
+    finally:
+        server.should_exit = True
+        thread.join(timeout=5)
+
+
+def _stream_chat_completion(payload: dict[str, object], content: str) -> Iterator[str]:
+    model = payload.get("model", "fabric-echo")
+    chunks = [
+        {
+            "id": "chatcmpl-fabric-test",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": model,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"role": "assistant"},
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-fabric-test",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": model,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"content": content},
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-fabric-test",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": model,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {},
+                    "finish_reason": "stop",
+                }
+            ],
+        },
+    ]
+
+    for chunk in chunks:
+        yield f"data: {json.dumps(chunk)}\n\n"
+    yield "data: [DONE]\n\n"

tests/_utils/utils.py

@@ -0,0 +1,22 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+
+import yaml
+
+
+def update_hermes_cli_relay_base_url(code_review_agent_dir: Path, api_server: str):
+    """
+    Update the base URL in the Hermes CLI relay profile.
+
+    Since the api_server uses a random available TCP port, the base_url needs to be updated for each test.
+
+    Args:
+        code_review_agent_dir (Path): The path to the code review agent directory.
+        api_server (str): The API server URL.
+    """
+    profile_path = code_review_agent_dir / "profiles" / "hermes-cli-relay.yaml"
+    profile = yaml.safe_load(profile_path.read_text())
+    profile["harness"]["settings"]["base_url"] = f"{api_server}/v1"
+    profile_path.write_text(yaml.safe_dump(profile, sort_keys=False))

tests/conftest.py

@@ -1,7 +1,11 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import os
 import shutil
 import sys
 import types
+from collections.abc import Iterator
 from pathlib import Path
 
 import pytest
@@ -48,6 +52,18 @@ def hermes_agent_dir_fixture(hermes_cli_agent_dir_src: Path, tmp_path: Path) ->
     assert agent_dir.exists(), f"Missing fake Hermes CLI agent directory: {agent_dir}"
     return agent_dir.resolve()
 
+@pytest.fixture(name="code_review_agent_dir")
+def code_review_agent_dir_fixture(repo_root: Path, tmp_path: Path) -> Path:
+    """
+    Creates a temporary copy of the example code review agent directory for testing.
+    """
+    source_dir = repo_root / "examples" / "code-review-agent"
+    assert source_dir.exists(), f"Missing Hermes code review agent directory: {source_dir}"
+    agent_dir = tmp_path / "code-review-agent"
+    shutil.copytree(source_dir, agent_dir, ignore=shutil.ignore_patterns("artifacts"))
+    assert agent_dir.exists(), f"Missing Hermes code review agent directory: {agent_dir}"
+    return agent_dir.resolve()
+
 @pytest.fixture(name="hermes_cli_profile", scope="session")
 def hermes_cli_profile_fixture() -> str:
     return "env_local"
@@ -67,6 +83,12 @@ def hermes_command_fixture(hermes_agent_dir: Path) -> Path:
     ), f"Missing fake Hermes CLI: {hermes_command}"
     return hermes_command.resolve()
 
+@pytest.fixture(name="api_server")
+def api_server_fixture(unused_tcp_port: int) -> Iterator[str]:
+    from _utils.mock_api_server import mock_api_server
+    with mock_api_server(unused_tcp_port) as base_url:
+        yield base_url
+
 @pytest.fixture(name="adapters_common_src_dir", scope="session")
 def adapters_common_src_dir_fixture() -> Path:
     adapters_common_src_dir = CUR_DIR.parent / "adapters" / "common" / "src"
@@ -86,13 +108,19 @@ def hermes_common_fixture(adapters_common: str) -> types.ModuleType:
     import nemo_fabric_adapters.common.hermes as hermes_common  # noqa: E402
     return hermes_common
 
+@pytest.fixture(name="nemo_relay")
+def nemo_relay_fixture() -> types.ModuleType:
+    return pytest.importorskip("nemo_relay", reason="nemo-relay extra is required")
+
 @pytest.fixture(name="hermes_state", scope="session")
 def require_hermes_state_fixture() -> types.ModuleType:
     """
     Fixture to ensure that the hermes_state module is available for tests that require it.
     """
-    try:
-        import hermes_state
-        return hermes_state
-    except ImportError:
-        pytest.skip("Skipping test because hermes-agent is not installed.")
+    return pytest.importorskip("hermes_state", reason="hermes extra is required")
+
+@pytest.fixture(name="mock_nvidia_api_key")
+def mock_nvidia_api_key_fixture() -> str:
+    nak = "test123"
+    os.environ["NVIDIA_API_KEY"] = nak
+    return nak