jpmschweitzerandClaude Opus 4.5
a22e168666
Add comprehensive test suites for: - NPM client (27 tests) - Ollama client (16 tests) - AI client and controller (34 tests) - Static controller (8 tests) - Tools controller DNS lookup (9 tests) - OIDC authentication (10 tests) - Housekeeping endpoints (28 tests) - Infrastructure endpoints (15 tests) - Health endpoints (12 tests) - Portainer client (12 tests) - Home Assistant client (24 tests) Total: 285 tests passing with 65% code coverage. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
399 lines
15 KiB
Python
399 lines
15 KiB
Python
"""Tests for NPM client."""
|
|
import pytest
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from datetime import datetime, timedelta
|
|
|
|
from src.clients.npm_client import NPMClient, get_npm_client
|
|
|
|
|
|
class TestNPMClientInit:
|
|
"""Test NPMClient initialization."""
|
|
|
|
@patch("src.clients.npm_client.settings")
|
|
def test_uses_settings_defaults(self, mock_settings):
|
|
"""Client should use settings for defaults."""
|
|
mock_settings.npm_url = "http://npm:81"
|
|
mock_settings.npm_email = "admin@example.com"
|
|
mock_settings.npm_password = "password123"
|
|
|
|
client = NPMClient()
|
|
|
|
assert client.base_url == "http://npm:81"
|
|
assert client.email == "admin@example.com"
|
|
assert client.password == "password123"
|
|
|
|
def test_accepts_custom_credentials(self):
|
|
"""Client should accept custom credentials."""
|
|
client = NPMClient(
|
|
base_url="http://custom:81",
|
|
email="custom@example.com",
|
|
password="custom_pass"
|
|
)
|
|
|
|
assert client.base_url == "http://custom:81"
|
|
assert client.email == "custom@example.com"
|
|
assert client.password == "custom_pass"
|
|
|
|
def test_strips_trailing_slash_from_url(self):
|
|
"""Client should strip trailing slash from URL."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81/",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
|
|
assert client.base_url == "http://npm:81"
|
|
|
|
@patch("src.clients.npm_client.logger")
|
|
@patch("src.clients.npm_client.settings")
|
|
def test_warns_when_credentials_missing(self, mock_settings, mock_logger):
|
|
"""Client should warn when credentials are not configured."""
|
|
mock_settings.npm_url = "http://npm:81"
|
|
mock_settings.npm_email = ""
|
|
mock_settings.npm_password = ""
|
|
|
|
NPMClient()
|
|
|
|
mock_logger.warning.assert_called_once()
|
|
|
|
def test_initializes_token_as_none(self):
|
|
"""Client should initialize token as None."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
|
|
assert client._token is None
|
|
assert client._token_expires is None
|
|
|
|
|
|
class TestNPMClientHeaders:
|
|
"""Test header generation."""
|
|
|
|
def test_get_headers_raises_without_token(self):
|
|
"""Headers should raise if no token available."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="No NPM token available"):
|
|
client._get_headers()
|
|
|
|
def test_get_headers_includes_bearer_token(self):
|
|
"""Headers should include Bearer token when available."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
client._token = "test_token_123"
|
|
|
|
headers = client._get_headers()
|
|
|
|
assert headers["Authorization"] == "Bearer test_token_123"
|
|
assert headers["Content-Type"] == "application/json"
|
|
|
|
|
|
class TestNPMClientToken:
|
|
"""Test token management."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_stores_token(self):
|
|
"""_refresh_token should store token from response."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"token": "new_token_abc"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
await client._refresh_token()
|
|
|
|
assert client._token == "new_token_abc"
|
|
assert client._token_expires is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_token_refreshes_when_none(self):
|
|
"""_ensure_token should refresh when no token."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
|
|
with patch.object(client, "_refresh_token", new_callable=AsyncMock) as mock_refresh:
|
|
await client._ensure_token()
|
|
mock_refresh.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_token_skips_refresh_when_valid(self):
|
|
"""_ensure_token should skip refresh when token is valid."""
|
|
client = NPMClient(
|
|
base_url="http://npm:81",
|
|
email="test@test.com",
|
|
password="pass"
|
|
)
|
|
client._token = "valid_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
with patch.object(client, "_refresh_token", new_callable=AsyncMock) as mock_refresh:
|
|
await client._ensure_token()
|
|
mock_refresh.assert_not_called()
|
|
|
|
|
|
class TestNPMClientHealthCheck:
|
|
"""Test health check functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_returns_true_on_200(self):
|
|
"""Health check should return True when NPM responds 200."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.health_check()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_returns_true_on_redirect(self):
|
|
"""Health check should return True on redirect (3xx)."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 302
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.health_check()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_returns_false_on_error(self):
|
|
"""Health check should return False on connection error."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_client = AsyncMock()
|
|
mock_client.get.side_effect = Exception("Connection refused")
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.health_check()
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestNPMClientProxyHosts:
|
|
"""Test proxy host operations."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_proxy_hosts_returns_list(self):
|
|
"""get_proxy_hosts should return list of proxy hosts."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
proxy_hosts = [
|
|
{"id": 1, "domain_names": ["example.com"]},
|
|
{"id": 2, "domain_names": ["test.com"]}
|
|
]
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = proxy_hosts
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.get_proxy_hosts()
|
|
|
|
assert result == proxy_hosts
|
|
assert len(result) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_proxy_host_returns_single_host(self):
|
|
"""get_proxy_host should return single proxy host."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
proxy_host = {"id": 1, "domain_names": ["example.com"], "forward_host": "app"}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = proxy_host
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.get_proxy_host(1)
|
|
|
|
assert result == proxy_host
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_proxy_host_posts_correct_data(self):
|
|
"""create_proxy_host should POST with correct payload."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"id": 1, "domain_names": ["new.com"]}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
await client.create_proxy_host(
|
|
domain_names=["new.com"],
|
|
forward_host="backend",
|
|
forward_port=8080
|
|
)
|
|
|
|
call_args = mock_client.post.call_args
|
|
assert call_args[1]["json"]["domain_names"] == ["new.com"]
|
|
assert call_args[1]["json"]["forward_host"] == "backend"
|
|
assert call_args[1]["json"]["forward_port"] == 8080
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_proxy_host_puts_correct_data(self):
|
|
"""update_proxy_host should PUT with correct payload."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
config = {"domain_names": ["updated.com"], "forward_host": "new-backend"}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.is_success = True
|
|
mock_response.json.return_value = config
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.put.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.update_proxy_host(1, config)
|
|
|
|
assert result == config
|
|
|
|
|
|
class TestNPMClientCertificates:
|
|
"""Test certificate operations."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_certificates_returns_list(self):
|
|
"""get_certificates should return list of certificates."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
certificates = [
|
|
{"id": 1, "domain_names": ["example.com"]},
|
|
{"id": 2, "domain_names": ["test.com"]}
|
|
]
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = certificates
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
result = await client.get_certificates()
|
|
|
|
assert result == certificates
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_certificate_posts_correct_data(self):
|
|
"""create_certificate should POST with correct payload."""
|
|
client = NPMClient(base_url="http://npm:81", email="test@test.com", password="pass")
|
|
client._token = "test_token"
|
|
client._token_expires = datetime.now() + timedelta(hours=12)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"id": 1, "domain_names": ["secure.com"]}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = mock_response
|
|
mock_client.__aenter__.return_value = mock_client
|
|
mock_client.__aexit__.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
|
|
await client.create_certificate(domain_names=["secure.com"])
|
|
|
|
call_args = mock_client.post.call_args
|
|
assert call_args[1]["json"]["domain_names"] == ["secure.com"]
|
|
assert call_args[1]["json"]["provider"] == "letsencrypt"
|
|
|
|
|
|
class TestNPMClientSingleton:
|
|
"""Test singleton pattern."""
|
|
|
|
def test_returns_same_instance(self):
|
|
"""get_npm_client should return singleton."""
|
|
import src.clients.npm_client as module
|
|
module._npm_client = None
|
|
|
|
client1 = get_npm_client()
|
|
client2 = get_npm_client()
|
|
|
|
assert client1 is client2
|