generated from coulomb/repo-seed
Bernd Worsch
a7eaf59ced
Full TDD implementation of the `bridge` CLI tool covering all phases from BRIDGE-WP-0001: project scaffolding, config loading, state management, audit logging, health checks, tunnel lifecycle manager, and all CLI commands (up/down/restart/status/logs). 77 tests, all green. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Tests for health checking."""
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
from bridge.health import HealthChecker, HealthResult
|
|
|
|
|
|
class TestHealthResult:
|
|
def test_ok(self):
|
|
r = HealthResult(ok=True, status_code=200)
|
|
assert r.ok
|
|
assert r.status_code == 200
|
|
assert r.error is None
|
|
|
|
def test_failure(self):
|
|
r = HealthResult(ok=False, error="connection refused")
|
|
assert not r.ok
|
|
assert r.error == "connection refused"
|
|
|
|
|
|
class TestHealthChecker:
|
|
@pytest.mark.asyncio
|
|
async def test_check_ok(self):
|
|
checker = HealthChecker(url="http://127.0.0.1:18000/health", timeout_seconds=5)
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await checker.check()
|
|
|
|
assert result.ok
|
|
assert result.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_connection_error(self):
|
|
import httpx
|
|
checker = HealthChecker(url="http://127.0.0.1:19999/health", timeout_seconds=1)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client.get = AsyncMock(side_effect=httpx.ConnectError("refused"))
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await checker.check()
|
|
|
|
assert not result.ok
|
|
assert result.error is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_http_error(self):
|
|
import httpx
|
|
checker = HealthChecker(url="http://127.0.0.1:18000/health", timeout_seconds=5)
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 503
|
|
mock_response.raise_for_status = MagicMock(
|
|
side_effect=httpx.HTTPStatusError("503", request=MagicMock(), response=mock_response)
|
|
)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await checker.check()
|
|
|
|
assert not result.ok
|
|
assert result.status_code == 503
|