90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
"""Tests for the Pi-hole API client."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestV5Client:
|
|
def _client(self):
|
|
from app.services.pihole import PiholeClient
|
|
return PiholeClient(url="http://pi.hole", api_key="testkey", version="v5")
|
|
|
|
def test_block_calls_v5_get(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v5_get") as mock_get:
|
|
client.block("samsungads.com")
|
|
mock_get.assert_called_once_with("black", "add", "samsungads.com")
|
|
|
|
def test_unblock_calls_v5_get_sub(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v5_get") as mock_get:
|
|
client.unblock("samsungads.com")
|
|
mock_get.assert_called_once_with("black", "sub", "samsungads.com")
|
|
|
|
def test_test_connection_returns_ok(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v5_test", return_value={"ok": True, "version": "v5", "domain_count": 42, "error": None}):
|
|
result = client.test_connection()
|
|
assert result["ok"] is True
|
|
assert result["domain_count"] == 42
|
|
|
|
def test_test_connection_catches_error(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v5_test", side_effect=Exception("connection refused")):
|
|
result = client.test_connection()
|
|
assert result["ok"] is False
|
|
assert "connection refused" in result["error"]
|
|
|
|
|
|
class TestV6Client:
|
|
def _client(self):
|
|
from app.services.pihole import PiholeClient
|
|
return PiholeClient(url="http://pi.hole", api_key="apppassword", version="v6")
|
|
|
|
def test_block_auths_then_posts(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v6_auth", return_value="test-sid") as mock_auth, \
|
|
patch.object(client, "_v6_post_domain") as mock_post:
|
|
client.block("samsungads.com", "test comment")
|
|
mock_auth.assert_called_once()
|
|
mock_post.assert_called_once_with("test-sid", "samsungads.com", "test comment")
|
|
|
|
def test_unblock_auths_then_deletes(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v6_auth", return_value="sid123") as mock_auth, \
|
|
patch.object(client, "_v6_delete_domain") as mock_del:
|
|
client.unblock("samsungads.com")
|
|
mock_auth.assert_called_once()
|
|
mock_del.assert_called_once_with("sid123", "samsungads.com")
|
|
|
|
def test_test_connection_returns_ok(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v6_test", return_value={"ok": True, "version": "v6", "domain_count": 5, "error": None}):
|
|
result = client.test_connection()
|
|
assert result["ok"] is True
|
|
|
|
def test_test_connection_catches_error(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v6_test", side_effect=Exception("timeout")):
|
|
result = client.test_connection()
|
|
assert result["ok"] is False
|
|
assert "timeout" in result["error"]
|
|
|
|
def test_block_default_comment(self):
|
|
client = self._client()
|
|
with patch.object(client, "_v6_auth", return_value="sid"), \
|
|
patch.object(client, "_v6_post_domain") as mock_post:
|
|
client.block("samsungads.com")
|
|
_, _, comment = mock_post.call_args.args
|
|
assert comment == "Turnstone block"
|
|
|
|
def test_auth_raises_on_invalid_session(self):
|
|
client = self._client()
|
|
invalid_resp = {"session": {"valid": False, "message": "wrong password"}}
|
|
with patch("httpx.Client") as mock_cls:
|
|
mock_c = mock_cls.return_value.__enter__.return_value
|
|
mock_c.post.return_value.raise_for_status.return_value = None
|
|
mock_c.post.return_value.json.return_value = invalid_resp
|
|
with pytest.raises(ValueError, match="Pi-hole v6 auth failed"):
|
|
client._v6_auth()
|