diff --git a/.agents/skills/python-tests/SKILL.md b/.agents/skills/python-tests/SKILL.md new file mode 100644 index 000000000..9015795a4 --- /dev/null +++ b/.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=""[, scope=""]) + def _fixture() -> : + ... + ``` + 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 "" + +# Run all tests +uv run pytest +``` + +## References + +- `pyproject.toml` +- `tests/conftest.py` diff --git a/.github/workflows/ci_python.yml b/.github/workflows/ci_python.yml index eed47c830..d1e6b5930 100644 --- a/.github/workflows/ci_python.yml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index d638e9cf1..43bce2f39 100644 --- a/pyproject.toml +++ b/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] diff --git a/tests/_utils/mock_api_server.py b/tests/_utils/mock_api_server.py new file mode 100644 index 000000000..187b2ec2e --- /dev/null +++ b/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" diff --git a/tests/_utils/utils.py b/tests/_utils/utils.py new file mode 100644 index 000000000..105541edc --- /dev/null +++ b/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)) diff --git a/tests/conftest.py b/tests/conftest.py index 333c61d28..8a90668a0 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_hermes_cli.py b/tests/test_hermes_cli.py index f821d6e2b..4e1e06ca9 100644 --- a/tests/test_hermes_cli.py +++ b/tests/test_hermes_cli.py @@ -1,9 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import json import types from pathlib import Path +import pytest +import yaml + +from _utils.utils import update_hermes_cli_relay_base_url from nemo_fabric import FabricClient @@ -59,3 +64,165 @@ async def test_hermes_cli_multi_turn(hermes_agent_dir: Path, hermes_cli_session_ assert session['model'] == 'test-model' assert session['source'] == 'fabric' assert session['title'] == runtime_id + + +class TestHermesE2E: + """ + E2E Hermes tests, which communicate with a mock API server not requiring an API key. + """ + + @pytest.fixture(autouse=True) + async def run_hermes_cli_relay( + self, + nemo_relay: types.ModuleType, + mock_nvidia_api_key: str, + code_review_agent_dir: Path, + api_server: str, + ): + assert nemo_relay is not None + assert mock_nvidia_api_key == "test123" + self.code_review_agent_dir = code_review_agent_dir + self.api_server = api_server + update_hermes_cli_relay_base_url(code_review_agent_dir, api_server) + + async with FabricClient() as client: + self.result = await client.run( + code_review_agent_dir, + profile="hermes_cli_relay", + input_text="Reply with exactly: relay ok", + ) + + self.output = self.result["output"] + self.artifacts = self.result["artifacts"] + self.artifact_root = Path(self.artifacts["root"]).resolve() + self.relay_artifacts = self.output["relay_artifacts"] + + async def test_artifacts(self): + assert self.result["status"] == "succeeded" + assert self.result["adapter_kind"] == "process" + assert self.result["metadata"]["adapter_runner"] == "process" + assert self.result["telemetry"]["relay_enabled"] is True + assert self.result["telemetry"]["metadata"]["relay_mode"] == "sdk" + + output = self.output + assert output["adapter"] == "cli" + assert output["harness"] == "hermes" + assert output["mode"] == "hermes_cli_oneshot" + assert output["base_url"] == f"{self.api_server}/v1" + assert output["returncode"] == 0 + assert output["error"] is None + assert output["relay_runtime"]["enabled"] is True + assert output["relay_runtime"]["mode"] == "sdk" + assert output["relay_runtime"]["emitter"] == "hermes.observability/nemo_relay" + + hermes_home = Path(output["hermes_home"]).resolve() + hermes_config_path = Path(output["hermes_config_path"]).resolve() + assert hermes_home.is_dir() + assert hermes_home.is_relative_to(self.code_review_agent_dir) + assert hermes_config_path.is_file() + assert hermes_config_path.is_relative_to(self.code_review_agent_dir) + + hermes_config = yaml.safe_load(hermes_config_path.read_text()) + assert hermes_config["model"]["provider"] == "nvidia" + assert hermes_config["model"]["default"] == "nvidia/nemotron-3-nano-30b-a3b" + assert hermes_config["model"]["base_url"] == f"{self.api_server}/v1" + assert hermes_config["plugins"]["enabled"] == ["observability/nemo_relay"] + assert output["hermes_native_config"]["plugins"] == ["observability/nemo_relay"] + + expected_artifact_root = ( + self.code_review_agent_dir / "artifacts" / "hermes-cli-relay" + ).resolve() + assert self.artifact_root == expected_artifact_root + assert self.artifact_root.is_dir() + + artifact_by_name = { + artifact["name"]: artifact + for artifact in self.artifacts["artifacts"] + } + assert "relay_config" in artifact_by_name + assert "stdout" in artifact_by_name + + relay_config_path = Path(artifact_by_name["relay_config"]["path"]).resolve() + assert relay_config_path.is_file() + assert relay_config_path.is_relative_to(self.artifact_root) + relay_config = json.loads(relay_config_path.read_text()) + assert relay_config["schema_version"] == "fabric.relay/v1alpha1" + assert relay_config["relay"]["enabled"] is True + assert relay_config["fabric"]["profile"] == "hermes_cli_relay" + + fabric_invocation_path = Path(output["fabric_invocation"]).resolve() + assert fabric_invocation_path.is_file() + assert fabric_invocation_path.is_relative_to(self.artifact_root) + assert fabric_invocation_path.name == "adapter-invocation.json" + + async def test_atof_artifacts(self): + kinds = {artifact["kind"] for artifact in self.relay_artifacts} + assert "atof" in kinds + + atof_paths = [ + Path(artifact["path"]).resolve() + for artifact in self.relay_artifacts + if artifact["kind"] == "atof" + ] + assert atof_paths + assert all(path.exists() for path in atof_paths) + assert all(path.is_relative_to(self.artifact_root) for path in atof_paths) + + atof_records = [ + json.loads(line) + for line in atof_paths[0].read_text().strip().splitlines() + ] + expected_atof_fields = { + "atof_version", + "attributes", + "category", + "data", + "kind", + "metadata", + "name", + "parent_uuid", + "scope_category", + "timestamp", + "uuid", + } + actual_atof_fields = set().union(*(record.keys() for record in atof_records)) + assert len(atof_records) == 6 + assert actual_atof_fields.issuperset(expected_atof_fields) + assert all( + record["metadata"]["model"] == "nvidia/nemotron-3-nano-30b-a3b" + and record["metadata"]["platform"] == "cli" + for record in atof_records + ) + assert ( + atof_records[0]["name"] + == f"hermes-session-{atof_records[0]['metadata']['session_id']}" + ) + assert atof_records[-1]["name"] == "hermes.session.end" + + async def test_atif_artifacts(self): + kinds = {artifact["kind"] for artifact in self.relay_artifacts} + assert "atif" in kinds + + atif_paths = [ + Path(artifact["path"]).resolve() + for artifact in self.relay_artifacts + if artifact["kind"] == "atif" + ] + assert atif_paths + assert all(path.exists() for path in atif_paths) + assert all(path.is_relative_to(self.artifact_root) for path in atif_paths) + + trajectory = json.loads(atif_paths[0].read_text()) + assert trajectory["agent"]["name"] in {"code-review-agent", "Hermes Agent"} + steps = trajectory["steps"] + assert len(steps) == 5 + + first_step = steps[0] + assert first_step["message"] == "hermes.turn.start" + assert first_step["extra"]["event_payload"]["is_first_turn"] is True + + last_step = steps[-1] + assert last_step["message"] == "hermes.session.end" + assert last_step["extra"]["event_payload"]["completed"] is True + assert last_step["extra"]["invocation"]["framework"] == "nemo_relay" + assert last_step["extra"]["invocation"]["status"] == "completed"