From 3d6e2b8b52d4386ff4444a06b26579f47b4e0266 Mon Sep 17 00:00:00 2001 From: DB Hurley Date: Wed, 11 Feb 2026 10:29:08 -0500 Subject: [PATCH] feat: Add comprehensive test suite with 94% coverage Implements Issue #6 - Python SDK Test Suite Tests added: - test_client.py: Client initialization, headers, health checks, dataclasses - test_policy.py: Policy check, enforce, govern context manager - test_errors.py: Exception classes and HTTP error handling - test_admin.py: Admin operations (agents, policies, audit log) - test_proxy.py: Proxy request methods - test_langchain.py: LangChain integration (decorators, wrappers, toolkit) Configuration: - pytest + pytest-asyncio + pytest-cov configured - CI updated to run tests with coverage on Python 3.9-3.12 - Codecov integration for coverage reporting - Added test status badge to README - Added .gitignore for Python artifacts Coverage: 94% (218 statements, 14 missed) - meshguard/__init__.py: 100% - meshguard/client.py: 98% - meshguard/exceptions.py: 100% - meshguard/langchain.py: 82% Closes #6 --- .github/workflows/ci.yml | 36 +++- .gitignore | 78 ++++++++ README.md | 1 + tests/__init__.py | 1 + tests/conftest.py | 54 +++++ tests/test_admin.py | 298 ++++++++++++++++++++++++++++ tests/test_client.py | 277 ++++++++++++++++++++++++++ tests/test_errors.py | 234 ++++++++++++++++++++++ tests/test_langchain.py | 413 +++++++++++++++++++++++++++++++++++++++ tests/test_policy.py | 264 +++++++++++++++++++++++++ tests/test_proxy.py | 175 +++++++++++++++++ 11 files changed, 1822 insertions(+), 9 deletions(-) create mode 100644 .gitignore create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_admin.py create mode 100644 tests/test_client.py create mode 100644 tests/test_errors.py create mode 100644 tests/test_langchain.py create mode 100644 tests/test_policy.py create mode 100644 tests/test_proxy.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 98268d5..f26a0b8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,44 +14,62 @@ jobs: python-version: ["3.9", "3.10", "3.11", "3.12"] steps: - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} + - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -e ".[dev]" || pip install -e . - pip install pytest pytest-asyncio - - name: Run tests - run: pytest -v || echo "No tests yet" + pip install -e ".[dev]" + + - name: Run tests with coverage + run: | + pytest -v --cov=meshguard --cov-report=xml --cov-report=term-missing + + - name: Upload coverage to Codecov + if: matrix.python-version == '3.12' + uses: codecov/codecov-action@v4 + with: + files: ./coverage.xml + fail_ci_if_error: false + verbose: true lint: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: python-version: "3.12" + - name: Install linting tools run: | python -m pip install --upgrade pip - pip install ruff mypy - - name: Type check - run: mypy src/ --ignore-missing-imports || true - - name: Lint - run: ruff check src/ || true + pip install ruff mypy httpx + + - name: Lint with Ruff + run: ruff check meshguard/ + + - name: Type check with mypy + run: mypy meshguard/ --ignore-missing-imports build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: python-version: "3.12" + - name: Build package run: | python -m pip install --upgrade pip build python -m build + - name: Check package run: | pip install twine diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c08cd0c --- /dev/null +++ b/.gitignore @@ -0,0 +1,78 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Ruff +.ruff_cache/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# macOS +.DS_Store diff --git a/README.md b/README.md index f9c4091..3f74740 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,6 @@ # MeshGuard Python SDK +[![CI](https://github.com/meshguard/meshguard-python/actions/workflows/ci.yml/badge.svg)](https://github.com/meshguard/meshguard-python/actions/workflows/ci.yml) [![PyPI version](https://badge.fury.io/py/meshguard.svg)](https://pypi.org/project/meshguard/) [![Python 3.9+](https://img.shields.io/badge/python-3.9+-blue.svg)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..255ee87 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# MeshGuard Python SDK Tests diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..6b603ff --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,54 @@ +""" +Pytest configuration and fixtures for MeshGuard tests. +""" + +import pytest +import httpx +from unittest.mock import MagicMock + + +@pytest.fixture +def mock_client(): + """Create a mock httpx client for testing.""" + return MagicMock(spec=httpx.Client) + + +@pytest.fixture +def gateway_url(): + """Default gateway URL for tests.""" + return "https://test.meshguard.app" + + +@pytest.fixture +def agent_token(): + """Test agent token.""" + return "test-agent-token-12345" + + +@pytest.fixture +def admin_token(): + """Test admin token.""" + return "test-admin-token-67890" + + +@pytest.fixture +def mock_response(): + """Factory for creating mock responses.""" + def _make_response( + status_code: int = 200, + json_data: dict = None, + content: bytes = None, + ): + response = MagicMock(spec=httpx.Response) + response.status_code = status_code + response.content = content or (json_data and b"{}") or b"" + response.text = str(json_data) if json_data else "" + + if json_data is not None: + response.json.return_value = json_data + else: + response.json.side_effect = ValueError("No JSON content") + + return response + + return _make_response diff --git a/tests/test_admin.py b/tests/test_admin.py new file mode 100644 index 0000000..4e63b04 --- /dev/null +++ b/tests/test_admin.py @@ -0,0 +1,298 @@ +""" +Tests for admin operations (agents, policies, audit). +""" + +import pytest +from unittest.mock import patch, MagicMock +import httpx + +from meshguard import MeshGuardClient, AuthenticationError +from meshguard.client import Agent + + +class TestListAgents: + """Test agent listing functionality.""" + + def test_list_agents_success(self, gateway_url, admin_token, mock_response): + """List agents returns Agent objects.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response( + status_code=200, + json_data={ + "agents": [ + { + "id": "agent-1", + "name": "test-agent-1", + "trustTier": "verified", + "tags": ["prod"], + "orgId": "org-1", + }, + { + "id": "agent-2", + "name": "test-agent-2", + "trustTier": "untrusted", + "tags": [], + }, + ] + }, + ) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + agents = client.list_agents() + + assert len(agents) == 2 + assert all(isinstance(a, Agent) for a in agents) + + assert agents[0].id == "agent-1" + assert agents[0].name == "test-agent-1" + assert agents[0].trust_tier == "verified" + assert agents[0].tags == ["prod"] + assert agents[0].org_id == "org-1" + + assert agents[1].id == "agent-2" + assert agents[1].trust_tier == "untrusted" + assert agents[1].org_id is None + + # Verify admin headers were used + call_args = mock_get.call_args + headers = call_args.kwargs["headers"] + assert headers["X-Admin-Token"] == admin_token + + client.close() + + def test_list_agents_empty(self, gateway_url, admin_token, mock_response): + """List agents handles empty list.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"agents": []}) + + with patch.object(client._client, "get", return_value=mock_resp): + agents = client.list_agents() + + assert agents == [] + client.close() + + def test_list_agents_requires_admin_token(self, gateway_url): + """List agents fails without admin token.""" + client = MeshGuardClient(gateway_url=gateway_url) + + with pytest.raises(AuthenticationError, match="Admin token required"): + client.list_agents() + + client.close() + + +class TestCreateAgent: + """Test agent creation functionality.""" + + def test_create_agent_success(self, gateway_url, admin_token, mock_response): + """Create agent returns agent details.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response( + status_code=201, + json_data={ + "id": "new-agent-id", + "name": "new-agent", + "token": "new-agent-token", + "trustTier": "verified", + }, + ) + + with patch.object(client._client, "post", return_value=mock_resp) as mock_post: + result = client.create_agent( + name="new-agent", + trust_tier="verified", + tags=["production", "api"], + ) + + assert result["id"] == "new-agent-id" + assert result["token"] == "new-agent-token" + + # Verify request body + call_args = mock_post.call_args + body = call_args.kwargs["json"] + assert body["name"] == "new-agent" + assert body["trustTier"] == "verified" + assert body["tags"] == ["production", "api"] + + client.close() + + def test_create_agent_defaults(self, gateway_url, admin_token, mock_response): + """Create agent uses default values.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=201, json_data={"id": "agent"}) + + with patch.object(client._client, "post", return_value=mock_resp) as mock_post: + client.create_agent(name="minimal-agent") + + body = mock_post.call_args.kwargs["json"] + assert body["trustTier"] == "verified" + assert body["tags"] == [] + + client.close() + + +class TestRevokeAgent: + """Test agent revocation functionality.""" + + def test_revoke_agent_success(self, gateway_url, admin_token, mock_response): + """Revoke agent completes successfully.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=204) + + with patch.object(client._client, "delete", return_value=mock_resp) as mock_delete: + client.revoke_agent("agent-to-revoke") + + # Verify correct URL + call_args = mock_delete.call_args + url = call_args.args[0] + assert "/admin/agents/agent-to-revoke" in url + + client.close() + + +class TestListPolicies: + """Test policy listing functionality.""" + + def test_list_policies_success(self, gateway_url, admin_token, mock_response): + """List policies returns policy data.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response( + status_code=200, + json_data={ + "policies": [ + { + "id": "policy-1", + "name": "default", + "rules": [{"action": "read:*", "effect": "allow"}], + }, + { + "id": "policy-2", + "name": "strict", + "rules": [{"action": "*", "effect": "deny"}], + }, + ] + }, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + policies = client.list_policies() + + assert len(policies) == 2 + assert policies[0]["name"] == "default" + assert policies[1]["name"] == "strict" + + client.close() + + def test_list_policies_empty(self, gateway_url, admin_token, mock_response): + """List policies handles empty list.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"policies": []}) + + with patch.object(client._client, "get", return_value=mock_resp): + policies = client.list_policies() + + assert policies == [] + client.close() + + +class TestGetAuditLog: + """Test audit log retrieval functionality.""" + + def test_get_audit_log_success(self, gateway_url, admin_token, mock_response): + """Get audit log returns entries.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response( + status_code=200, + json_data={ + "entries": [ + { + "id": "entry-1", + "action": "read:contacts", + "decision": "allow", + "timestamp": "2024-01-01T00:00:00Z", + }, + { + "id": "entry-2", + "action": "write:email", + "decision": "deny", + "timestamp": "2024-01-01T00:01:00Z", + }, + ] + }, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + entries = client.get_audit_log() + + assert len(entries) == 2 + assert entries[0]["decision"] == "allow" + assert entries[1]["decision"] == "deny" + + client.close() + + def test_get_audit_log_with_params(self, gateway_url, admin_token, mock_response): + """Get audit log passes query parameters.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"entries": []}) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + client.get_audit_log(limit=25, decision="deny") + + params = mock_get.call_args.kwargs["params"] + assert params["limit"] == 25 + assert params["decision"] == "deny" + + client.close() + + def test_get_audit_log_default_limit(self, gateway_url, admin_token, mock_response): + """Get audit log uses default limit.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"entries": []}) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + client.get_audit_log() + + params = mock_get.call_args.kwargs["params"] + assert params["limit"] == 50 + + client.close() diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..d292fd1 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,277 @@ +""" +Tests for MeshGuardClient initialization and core functionality. +""" + +import os +import pytest +from unittest.mock import patch, MagicMock +import httpx + +from meshguard import MeshGuardClient +from meshguard.client import PolicyDecision, Agent + + +class TestClientInitialization: + """Test MeshGuardClient initialization.""" + + def test_init_with_explicit_params(self, gateway_url, agent_token, admin_token): + """Client initializes with explicit parameters.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + admin_token=admin_token, + timeout=60.0, + ) + + assert client.gateway_url == gateway_url + assert client.agent_token == agent_token + assert client.admin_token == admin_token + assert client.timeout == 60.0 + client.close() + + def test_init_with_env_vars(self): + """Client initializes from environment variables.""" + with patch.dict(os.environ, { + "MESHGUARD_GATEWAY_URL": "https://env.meshguard.app", + "MESHGUARD_AGENT_TOKEN": "env-agent-token", + "MESHGUARD_ADMIN_TOKEN": "env-admin-token", + }): + client = MeshGuardClient() + + assert client.gateway_url == "https://env.meshguard.app" + assert client.agent_token == "env-agent-token" + assert client.admin_token == "env-admin-token" + client.close() + + def test_init_default_gateway_url(self): + """Client uses default gateway URL when not specified.""" + with patch.dict(os.environ, {}, clear=True): + # Clear relevant env vars + for key in ["MESHGUARD_GATEWAY_URL", "MESHGUARD_AGENT_TOKEN", "MESHGUARD_ADMIN_TOKEN"]: + os.environ.pop(key, None) + + client = MeshGuardClient() + assert client.gateway_url == "https://dashboard.meshguard.app" + client.close() + + def test_init_strips_trailing_slash(self): + """Gateway URL has trailing slash stripped.""" + client = MeshGuardClient(gateway_url="https://test.meshguard.app/") + assert client.gateway_url == "https://test.meshguard.app" + client.close() + + def test_init_generates_trace_id(self): + """Client generates trace ID if not provided.""" + client = MeshGuardClient() + assert client.trace_id is not None + assert len(client.trace_id) > 0 + client.close() + + def test_init_uses_provided_trace_id(self): + """Client uses provided trace ID.""" + client = MeshGuardClient(trace_id="custom-trace-123") + assert client.trace_id == "custom-trace-123" + client.close() + + +class TestHeaders: + """Test header generation.""" + + def test_headers_with_auth(self, gateway_url, agent_token): + """Headers include auth token when available.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + headers = client._headers() + + assert headers["Authorization"] == f"Bearer {agent_token}" + assert "X-MeshGuard-Trace-ID" in headers + client.close() + + def test_headers_without_auth(self, gateway_url): + """Headers can exclude auth when specified.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token="some-token", + ) + + headers = client._headers(include_auth=False) + + assert "Authorization" not in headers + assert "X-MeshGuard-Trace-ID" in headers + client.close() + + def test_headers_no_token(self, gateway_url): + """Headers without token don't include Authorization.""" + client = MeshGuardClient(gateway_url=gateway_url) + + headers = client._headers() + + assert "Authorization" not in headers + client.close() + + def test_admin_headers(self, gateway_url, admin_token): + """Admin headers include admin token.""" + client = MeshGuardClient( + gateway_url=gateway_url, + admin_token=admin_token, + ) + + headers = client._admin_headers() + + assert headers["X-Admin-Token"] == admin_token + assert "X-MeshGuard-Trace-ID" in headers + client.close() + + def test_admin_headers_raises_without_token(self, gateway_url): + """Admin headers raise error without admin token.""" + client = MeshGuardClient(gateway_url=gateway_url) + + from meshguard import AuthenticationError + with pytest.raises(AuthenticationError, match="Admin token required"): + client._admin_headers() + + client.close() + + +class TestContextManager: + """Test context manager functionality.""" + + def test_context_manager_opens_and_closes(self, gateway_url): + """Context manager properly opens and closes client.""" + with MeshGuardClient(gateway_url=gateway_url) as client: + assert client._client is not None + + def test_close_method(self, gateway_url): + """Close method closes the HTTP client.""" + client = MeshGuardClient(gateway_url=gateway_url) + client.close() + # Should not raise even if called multiple times + client.close() + + +class TestHealthCheck: + """Test health check functionality.""" + + def test_health_returns_data(self, gateway_url, mock_response): + """Health check returns gateway health data.""" + client = MeshGuardClient(gateway_url=gateway_url) + + mock_resp = mock_response( + status_code=200, + json_data={"status": "healthy", "version": "1.0.0"}, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + health = client.health() + + assert health["status"] == "healthy" + assert health["version"] == "1.0.0" + client.close() + + def test_is_healthy_true(self, gateway_url, mock_response): + """is_healthy returns True when gateway is healthy.""" + client = MeshGuardClient(gateway_url=gateway_url) + + mock_resp = mock_response( + status_code=200, + json_data={"status": "healthy"}, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + assert client.is_healthy() is True + + client.close() + + def test_is_healthy_false_on_unhealthy(self, gateway_url, mock_response): + """is_healthy returns False when gateway reports unhealthy.""" + client = MeshGuardClient(gateway_url=gateway_url) + + mock_resp = mock_response( + status_code=200, + json_data={"status": "unhealthy"}, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + assert client.is_healthy() is False + + client.close() + + def test_is_healthy_false_on_error(self, gateway_url): + """is_healthy returns False on connection error.""" + client = MeshGuardClient(gateway_url=gateway_url) + + with patch.object( + client._client, "get", + side_effect=httpx.ConnectError("Connection failed"), + ): + assert client.is_healthy() is False + + client.close() + + +class TestDataclasses: + """Test dataclass structures.""" + + def test_policy_decision_fields(self): + """PolicyDecision has expected fields.""" + decision = PolicyDecision( + allowed=True, + action="read:contacts", + decision="allow", + policy="default", + rule="rule-1", + reason="Allowed by policy", + trace_id="trace-123", + ) + + assert decision.allowed is True + assert decision.action == "read:contacts" + assert decision.decision == "allow" + assert decision.policy == "default" + assert decision.rule == "rule-1" + assert decision.reason == "Allowed by policy" + assert decision.trace_id == "trace-123" + + def test_policy_decision_optional_fields(self): + """PolicyDecision works with minimal fields.""" + decision = PolicyDecision( + allowed=False, + action="write:email", + decision="deny", + ) + + assert decision.allowed is False + assert decision.policy is None + assert decision.rule is None + assert decision.reason is None + assert decision.trace_id is None + + def test_agent_fields(self): + """Agent has expected fields.""" + agent = Agent( + id="agent-123", + name="test-agent", + trust_tier="verified", + tags=["production", "web"], + org_id="org-456", + ) + + assert agent.id == "agent-123" + assert agent.name == "test-agent" + assert agent.trust_tier == "verified" + assert agent.tags == ["production", "web"] + assert agent.org_id == "org-456" + + def test_agent_default_tags(self): + """Agent has empty list as default tags.""" + agent = Agent( + id="agent-123", + name="test-agent", + trust_tier="verified", + ) + + assert agent.tags == [] + assert agent.org_id is None diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..60038e6 --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,234 @@ +""" +Tests for error handling and exception classes. +""" + +import pytest +from unittest.mock import patch, MagicMock +import httpx + +from meshguard import ( + MeshGuardClient, + MeshGuardError, + AuthenticationError, + PolicyDeniedError, + RateLimitError, +) + + +class TestExceptionClasses: + """Test exception class structures.""" + + def test_meshguard_error_base(self): + """MeshGuardError is the base exception.""" + error = MeshGuardError("Test error") + assert str(error) == "Test error" + assert isinstance(error, Exception) + + def test_authentication_error(self): + """AuthenticationError inherits from MeshGuardError.""" + error = AuthenticationError("Invalid token") + assert str(error) == "Invalid token" + assert isinstance(error, MeshGuardError) + + def test_rate_limit_error(self): + """RateLimitError inherits from MeshGuardError.""" + error = RateLimitError("Too many requests") + assert str(error) == "Too many requests" + assert isinstance(error, MeshGuardError) + + def test_policy_denied_error_minimal(self): + """PolicyDeniedError with minimal args.""" + error = PolicyDeniedError(action="read:contacts") + + assert error.action == "read:contacts" + assert error.policy is None + assert error.rule is None + assert error.reason == "Access denied by policy" + assert "Action 'read:contacts' denied" in str(error) + + def test_policy_denied_error_full(self): + """PolicyDeniedError with all args.""" + error = PolicyDeniedError( + action="delete:database", + policy="strict-policy", + rule="no-delete-rule", + reason="Delete operations are prohibited", + ) + + assert error.action == "delete:database" + assert error.policy == "strict-policy" + assert error.rule == "no-delete-rule" + assert error.reason == "Delete operations are prohibited" + + message = str(error) + assert "Action 'delete:database' denied" in message + assert "policy 'strict-policy'" in message + assert "rule: no-delete-rule" in message + assert "Delete operations are prohibited" in message + + def test_policy_denied_error_is_meshguard_error(self): + """PolicyDeniedError inherits from MeshGuardError.""" + error = PolicyDeniedError(action="test") + assert isinstance(error, MeshGuardError) + + +class TestResponseErrorHandling: + """Test error handling from HTTP responses.""" + + def test_401_raises_authentication_error(self, gateway_url, agent_token): + """401 response raises AuthenticationError.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 401 + mock_resp.content = b"{}" + + with pytest.raises(AuthenticationError, match="Invalid or expired token"): + client._handle_response(mock_resp) + + client.close() + + def test_403_raises_policy_denied_error(self, gateway_url, agent_token): + """403 response raises PolicyDeniedError.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 403 + mock_resp.content = b'{"action": "write:email", "policy": "read-only", "message": "Write denied"}' + mock_resp.json.return_value = { + "action": "write:email", + "policy": "read-only", + "message": "Write denied", + } + + with pytest.raises(PolicyDeniedError) as exc_info: + client._handle_response(mock_resp) + + error = exc_info.value + assert error.action == "write:email" + assert error.policy == "read-only" + assert error.reason == "Write denied" + + client.close() + + def test_403_with_empty_body(self, gateway_url, agent_token): + """403 response with empty body uses defaults.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 403 + mock_resp.content = b"" + + with pytest.raises(PolicyDeniedError) as exc_info: + client._handle_response(mock_resp) + + error = exc_info.value + assert error.action == "unknown" + assert error.reason == "Access denied by policy" + + client.close() + + def test_429_raises_rate_limit_error(self, gateway_url, agent_token): + """429 response raises RateLimitError.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 429 + + with pytest.raises(RateLimitError, match="Rate limit exceeded"): + client._handle_response(mock_resp) + + client.close() + + def test_500_raises_meshguard_error(self, gateway_url, agent_token): + """500 response raises generic MeshGuardError.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 500 + mock_resp.text = "Internal Server Error" + + with pytest.raises(MeshGuardError, match="Request failed: 500"): + client._handle_response(mock_resp) + + client.close() + + def test_400_raises_meshguard_error(self, gateway_url, agent_token): + """400 response raises generic MeshGuardError.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 400 + mock_resp.text = "Bad Request" + + with pytest.raises(MeshGuardError, match="Request failed: 400"): + client._handle_response(mock_resp) + + client.close() + + def test_200_returns_json(self, gateway_url, agent_token): + """200 response returns parsed JSON.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.content = b'{"key": "value"}' + mock_resp.json.return_value = {"key": "value"} + + result = client._handle_response(mock_resp) + assert result == {"key": "value"} + + client.close() + + def test_200_empty_body_returns_empty_dict(self, gateway_url, agent_token): + """200 response with empty body returns empty dict.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.content = b"" + + result = client._handle_response(mock_resp) + assert result == {} + + client.close() + + def test_204_no_content(self, gateway_url, agent_token): + """204 response returns empty dict.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 204 + mock_resp.content = b"" + + result = client._handle_response(mock_resp) + assert result == {} + + client.close() diff --git a/tests/test_langchain.py b/tests/test_langchain.py new file mode 100644 index 0000000..90e2f78 --- /dev/null +++ b/tests/test_langchain.py @@ -0,0 +1,413 @@ +""" +Tests for LangChain integration. +""" + +import pytest +from unittest.mock import patch, MagicMock, AsyncMock +import httpx + +from meshguard import MeshGuardClient, PolicyDeniedError +from meshguard.langchain import ( + governed_tool, + GovernedTool, + GovernedToolkit, +) + + +class TestGovernedToolDecorator: + """Test the @governed_tool decorator.""" + + def test_governed_tool_allows_execution(self, gateway_url, agent_token, mock_response): + """Decorated function executes when policy allows.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + @governed_tool("read:contacts", client=client) + def fetch_contacts(query: str) -> str: + return f"Found contacts for: {query}" + + with patch.object(client._client, "get", return_value=mock_resp): + result = fetch_contacts("John") + + assert result == "Found contacts for: John" + client.close() + + def test_governed_tool_denies_execution(self, gateway_url, agent_token, mock_response): + """Decorated function raises when policy denies.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"policy": "strict", "message": "Not allowed"}, + ) + + @governed_tool("delete:contacts", client=client) + def delete_contacts() -> str: + return "Deleted" + + with patch.object(client._client, "get", return_value=mock_resp): + with pytest.raises(PolicyDeniedError): + delete_contacts() + + client.close() + + def test_governed_tool_on_deny_callback(self, gateway_url, agent_token, mock_response): + """on_deny callback is called when policy denies.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"message": "Denied"}, + ) + + def handle_denial(error, *args, **kwargs): + return f"Denied: {error.reason}" + + @governed_tool("write:email", client=client, on_deny=handle_denial) + def send_email(to: str) -> str: + return f"Sent to {to}" + + with patch.object(client._client, "get", return_value=mock_resp): + result = send_email("test@example.com") + + assert "Denied" in result + client.close() + + def test_governed_tool_preserves_metadata(self, gateway_url, agent_token): + """Decorator preserves function metadata.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + @governed_tool("read:data", client=client) + def my_function(): + """My docstring.""" + pass + + assert my_function.__name__ == "my_function" + assert my_function.__doc__ == "My docstring." + assert my_function._meshguard_action == "read:data" + + client.close() + + def test_governed_tool_creates_client_if_not_provided(self, mock_response): + """Decorator creates client from env vars if not provided.""" + @governed_tool("read:test") + def test_func(): + return "executed" + + # Mock the MeshGuardClient that gets created inside + with patch("meshguard.langchain.MeshGuardClient") as MockClient: + mock_client = MagicMock() + mock_decision = MagicMock() + mock_decision.allowed = True + mock_client.enforce.return_value = mock_decision + MockClient.return_value = mock_client + + result = test_func() + + assert result == "executed" + MockClient.assert_called_once() + + +class TestGovernedTool: + """Test the GovernedTool wrapper class.""" + + def test_governed_tool_run(self, gateway_url, agent_token, mock_response): + """GovernedTool.run() works correctly.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + # Create a mock tool + mock_tool = MagicMock() + mock_tool.name = "search" + mock_tool.description = "Search the web" + mock_tool.run.return_value = "Search results" + + governed = GovernedTool( + tool=mock_tool, + action="read:search", + client=client, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + result = governed.run("test query") + + assert result == "Search results" + mock_tool.run.assert_called_once_with("test query") + client.close() + + def test_governed_tool_run_denied(self, gateway_url, agent_token, mock_response): + """GovernedTool.run() raises when denied.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"message": "Denied"}, + ) + + mock_tool = MagicMock() + governed = GovernedTool( + tool=mock_tool, + action="write:dangerous", + client=client, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + with pytest.raises(PolicyDeniedError): + governed.run("dangerous operation") + + mock_tool.run.assert_not_called() + client.close() + + def test_governed_tool_on_deny(self, gateway_url, agent_token, mock_response): + """GovernedTool uses on_deny callback.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"message": "Denied"}, + ) + + mock_tool = MagicMock() + + def deny_handler(error, *args, **kwargs): + return "Access denied" + + governed = GovernedTool( + tool=mock_tool, + action="write:test", + client=client, + on_deny=deny_handler, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + result = governed.run("test") + + assert result == "Access denied" + client.close() + + def test_governed_tool_callable(self, gateway_url, agent_token, mock_response): + """GovernedTool is callable.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + mock_tool = MagicMock() + mock_tool.run.return_value = "result" + + governed = GovernedTool( + tool=mock_tool, + action="read:test", + client=client, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + result = governed("test arg") + + assert result == "result" + client.close() + + def test_governed_tool_copies_attributes(self, gateway_url, agent_token): + """GovernedTool copies name and description from wrapped tool.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_tool = MagicMock() + mock_tool.name = "custom_tool" + mock_tool.description = "A custom tool description" + + governed = GovernedTool( + tool=mock_tool, + action="read:custom", + client=client, + ) + + assert governed.name == "custom_tool" + assert governed.description == "A custom tool description" + client.close() + + @pytest.mark.asyncio + async def test_governed_tool_arun(self, gateway_url, agent_token, mock_response): + """GovernedTool.arun() works correctly.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + mock_tool = MagicMock() + mock_tool.arun = AsyncMock(return_value="async result") + + governed = GovernedTool( + tool=mock_tool, + action="read:async", + client=client, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + result = await governed.arun("async query") + + assert result == "async result" + mock_tool.arun.assert_called_once_with("async query") + client.close() + + +class TestGovernedToolkit: + """Test the GovernedToolkit class.""" + + def test_toolkit_get_tools(self, gateway_url, agent_token): + """Toolkit returns governed tools.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + tool1 = MagicMock() + tool1.name = "search" + + tool2 = MagicMock() + tool2.name = "calculator" + + toolkit = GovernedToolkit( + tools=[tool1, tool2], + client=client, + action_map={ + "search": "read:web_search", + "calculator": "execute:math", + }, + ) + + governed_tools = toolkit.get_tools() + + assert len(governed_tools) == 2 + assert all(isinstance(t, GovernedTool) for t in governed_tools) + + client.close() + + def test_toolkit_action_map(self, gateway_url, agent_token): + """Toolkit uses action_map for tool actions.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + tool = MagicMock() + tool.name = "search" + + toolkit = GovernedToolkit( + tools=[tool], + client=client, + action_map={"search": "custom:action"}, + ) + + governed_tools = toolkit.get_tools() + assert governed_tools[0].action == "custom:action" + + client.close() + + def test_toolkit_default_action(self, gateway_url, agent_token): + """Toolkit uses default_action for unmapped tools.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + tool = MagicMock() + tool.name = "unknown_tool" + + toolkit = GovernedToolkit( + tools=[tool], + client=client, + action_map={}, + default_action="execute:unknown", + ) + + governed_tools = toolkit.get_tools() + assert governed_tools[0].action == "execute:unknown" + + client.close() + + def test_toolkit_get_action(self, gateway_url, agent_token): + """Toolkit.get_action returns correct action.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + toolkit = GovernedToolkit( + tools=[], + client=client, + action_map={"search": "read:search"}, + default_action="execute:default", + ) + + tool_with_name = MagicMock() + tool_with_name.name = "search" + + tool_without_name = MagicMock() + tool_without_name.name = "other" + + assert toolkit.get_action(tool_with_name) == "read:search" + assert toolkit.get_action(tool_without_name) == "execute:default" + + client.close() + + def test_toolkit_on_deny_propagates(self, gateway_url, agent_token, mock_response): + """Toolkit propagates on_deny to governed tools.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"message": "Denied"}, + ) + + tool = MagicMock() + tool.name = "test" + + def deny_handler(error, *args, **kwargs): + return "Handled denial" + + toolkit = GovernedToolkit( + tools=[tool], + client=client, + on_deny=deny_handler, + ) + + governed_tools = toolkit.get_tools() + + with patch.object(client._client, "get", return_value=mock_resp): + result = governed_tools[0].run() + + assert result == "Handled denial" + client.close() diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..5ffe25c --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,264 @@ +""" +Tests for policy evaluation functionality (check, enforce, govern). +""" + +import pytest +from unittest.mock import patch, MagicMock +import httpx + +from meshguard import MeshGuardClient, PolicyDeniedError +from meshguard.client import PolicyDecision, GovernedContext + + +class TestPolicyCheck: + """Test policy check functionality.""" + + def test_check_allowed(self, gateway_url, agent_token, mock_response): + """Check returns allowed=True when policy permits.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=200, + json_data={"policy": "default"}, + ) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + decision = client.check("read:contacts") + + assert decision.allowed is True + assert decision.action == "read:contacts" + assert decision.decision == "allow" + assert decision.policy == "default" + + # Verify correct headers were sent + call_args = mock_get.call_args + headers = call_args.kwargs["headers"] + assert headers["X-MeshGuard-Action"] == "read:contacts" + + client.close() + + def test_check_denied(self, gateway_url, agent_token, mock_response): + """Check returns allowed=False when policy denies.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={ + "policy": "strict", + "rule": "no-delete", + "message": "Delete operations not allowed", + }, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + decision = client.check("delete:database") + + assert decision.allowed is False + assert decision.action == "delete:database" + assert decision.decision == "deny" + assert decision.policy == "strict" + assert decision.rule == "no-delete" + assert decision.reason == "Delete operations not allowed" + + client.close() + + def test_check_with_resource(self, gateway_url, agent_token, mock_response): + """Check includes resource in headers.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + client.check("read:contacts", resource="contacts/123") + + headers = mock_get.call_args.kwargs["headers"] + assert headers["X-MeshGuard-Resource"] == "contacts/123" + + client.close() + + def test_check_includes_trace_id(self, gateway_url, agent_token, mock_response): + """Check includes trace ID in response.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + trace_id="test-trace-id", + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "get", return_value=mock_resp): + decision = client.check("read:contacts") + + assert decision.trace_id == "test-trace-id" + + client.close() + + def test_check_handles_empty_response(self, gateway_url, agent_token, mock_response): + """Check handles empty response body.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.content = b"" + mock_resp.json.return_value = {} + + with patch.object(client._client, "get", return_value=mock_resp): + decision = client.check("read:contacts") + + assert decision.allowed is True + + client.close() + + +class TestPolicyEnforce: + """Test policy enforcement functionality.""" + + def test_enforce_returns_decision_when_allowed(self, gateway_url, agent_token, mock_response): + """Enforce returns PolicyDecision when allowed.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"policy": "default"}) + + with patch.object(client._client, "get", return_value=mock_resp): + decision = client.enforce("read:contacts") + + assert decision.allowed is True + assert isinstance(decision, PolicyDecision) + + client.close() + + def test_enforce_raises_when_denied(self, gateway_url, agent_token, mock_response): + """Enforce raises PolicyDeniedError when denied.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={ + "policy": "strict", + "rule": "no-delete", + "message": "Not allowed", + }, + ) + + with patch.object(client._client, "get", return_value=mock_resp): + with pytest.raises(PolicyDeniedError) as exc_info: + client.enforce("delete:database") + + error = exc_info.value + assert error.action == "delete:database" + assert error.policy == "strict" + assert error.rule == "no-delete" + + client.close() + + def test_enforce_with_resource(self, gateway_url, agent_token, mock_response): + """Enforce passes resource to check.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + client.enforce("read:contacts", resource="contacts/456") + + headers = mock_get.call_args.kwargs["headers"] + assert headers["X-MeshGuard-Resource"] == "contacts/456" + + client.close() + + +class TestGovernContext: + """Test governed context manager.""" + + def test_govern_allows_execution(self, gateway_url, agent_token, mock_response): + """Govern context allows code execution when permitted.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"policy": "default"}) + + executed = False + with patch.object(client._client, "get", return_value=mock_resp): + with client.govern("read:contacts") as decision: + executed = True + assert decision.allowed is True + + assert executed is True + client.close() + + def test_govern_prevents_execution_when_denied(self, gateway_url, agent_token, mock_response): + """Govern context raises and prevents execution when denied.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response( + status_code=403, + json_data={"policy": "strict", "message": "Denied"}, + ) + + executed = False + with patch.object(client._client, "get", return_value=mock_resp): + with pytest.raises(PolicyDeniedError): + with client.govern("delete:database"): + executed = True + + assert executed is False + client.close() + + def test_govern_with_resource(self, gateway_url, agent_token, mock_response): + """Govern context passes resource.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "get", return_value=mock_resp) as mock_get: + with client.govern("read:contacts", resource="contacts/789"): + pass + + headers = mock_get.call_args.kwargs["headers"] + assert headers["X-MeshGuard-Resource"] == "contacts/789" + + client.close() + + def test_governed_context_class(self, gateway_url, agent_token): + """GovernedContext class holds correct attributes.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + ctx = GovernedContext(client, "test:action", "test-resource") + + assert ctx.client is client + assert ctx.action == "test:action" + assert ctx.resource == "test-resource" + assert ctx.decision is None + + client.close() diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 0000000..c51479c --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,175 @@ +""" +Tests for proxy request functionality. +""" + +import pytest +from unittest.mock import patch, MagicMock +import httpx + +from meshguard import MeshGuardClient + + +class TestProxyRequests: + """Test proxy request methods.""" + + def test_request_method(self, gateway_url, agent_token, mock_response): + """Generic request method works correctly.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"result": "success"}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + response = client.request( + method="POST", + path="/api/data", + action="write:data", + json={"key": "value"}, + ) + + # Verify request parameters + call_args = mock_req.call_args + assert call_args.args[0] == "POST" + assert f"{gateway_url}/proxy/api/data" in call_args.args[1] + + headers = call_args.kwargs["headers"] + assert headers["X-MeshGuard-Action"] == "write:data" + assert "Authorization" in headers + + client.close() + + def test_request_strips_leading_slash(self, gateway_url, agent_token, mock_response): + """Request path has leading slash handled correctly.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + client.request("GET", "/api/test", action="read:test") + + url = mock_req.call_args.args[1] + assert "/proxy/api/test" in url + assert "/proxy//api" not in url + + client.close() + + def test_get_shorthand(self, gateway_url, agent_token, mock_response): + """GET shorthand method works.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"items": []}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + response = client.get("/api/items", action="read:items") + + assert mock_req.call_args.args[0] == "GET" + client.close() + + def test_post_shorthand(self, gateway_url, agent_token, mock_response): + """POST shorthand method works.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=201, json_data={"id": "new-item"}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + response = client.post( + "/api/items", + action="write:items", + json={"name": "test"}, + ) + + assert mock_req.call_args.args[0] == "POST" + client.close() + + def test_put_shorthand(self, gateway_url, agent_token, mock_response): + """PUT shorthand method works.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={"updated": True}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + response = client.put( + "/api/items/123", + action="write:items", + json={"name": "updated"}, + ) + + assert mock_req.call_args.args[0] == "PUT" + client.close() + + def test_delete_shorthand(self, gateway_url, agent_token, mock_response): + """DELETE shorthand method works.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=204) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + response = client.delete("/api/items/123", action="delete:items") + + assert mock_req.call_args.args[0] == "DELETE" + client.close() + + def test_request_merges_headers(self, gateway_url, agent_token, mock_response): + """Request merges custom headers with auth headers.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + client.request( + "GET", + "/api/test", + action="read:test", + headers={"X-Custom-Header": "custom-value"}, + ) + + headers = mock_req.call_args.kwargs["headers"] + assert headers["X-Custom-Header"] == "custom-value" + assert headers["X-MeshGuard-Action"] == "read:test" + assert "Authorization" in headers + + client.close() + + def test_request_passes_kwargs(self, gateway_url, agent_token, mock_response): + """Request passes additional kwargs to httpx.""" + client = MeshGuardClient( + gateway_url=gateway_url, + agent_token=agent_token, + ) + + mock_resp = mock_response(status_code=200, json_data={}) + + with patch.object(client._client, "request", return_value=mock_resp) as mock_req: + client.request( + "POST", + "/api/upload", + action="write:upload", + data=b"file content", + params={"version": "2"}, + ) + + kwargs = mock_req.call_args.kwargs + assert kwargs["data"] == b"file content" + assert kwargs["params"] == {"version": "2"} + + client.close()