85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""Tests for the fail2ban metadata service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.services.fail2ban_metadata_service import Fail2BanMetadataService
|
|
from app.exceptions import Fail2BanConnectionError
|
|
|
|
|
|
async def test_get_db_path_caches_result() -> None:
|
|
"""Same socket path should be resolved only once and reused from cache."""
|
|
service = Fail2BanMetadataService()
|
|
client = AsyncMock()
|
|
client.send = AsyncMock(return_value=(0, "/tmp/fail2ban.sqlite3"))
|
|
client.__aenter__.return_value = client
|
|
client.__aexit__.return_value = None
|
|
|
|
with patch(
|
|
"app.services.fail2ban_metadata_service.Fail2BanClient",
|
|
return_value=client,
|
|
) as mock_client_cls:
|
|
result1 = await service.get_db_path("/tmp/fail2ban.sock")
|
|
result2 = await service.get_db_path("/tmp/fail2ban.sock")
|
|
|
|
assert result1 == "/tmp/fail2ban.sqlite3"
|
|
assert result2 == "/tmp/fail2ban.sqlite3"
|
|
assert mock_client_cls.call_count == 1
|
|
assert client.send.call_count == 1
|
|
|
|
|
|
async def test_get_db_path_uses_cached_path_when_refresh_fails() -> None:
|
|
"""When explicit refresh fails, the previously cached path is returned."""
|
|
service = Fail2BanMetadataService()
|
|
first_client = AsyncMock()
|
|
first_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban.sqlite3"))
|
|
first_client.__aenter__.return_value = first_client
|
|
first_client.__aexit__.return_value = None
|
|
|
|
second_client = AsyncMock()
|
|
second_client.send = AsyncMock(side_effect=Fail2BanConnectionError("socket down", "/tmp/fail2ban.sock"))
|
|
second_client.__aenter__.return_value = second_client
|
|
second_client.__aexit__.return_value = None
|
|
|
|
with patch(
|
|
"app.services.fail2ban_metadata_service.Fail2BanClient",
|
|
side_effect=[first_client, second_client],
|
|
) as mock_client_cls:
|
|
initial_path = await service.get_db_path("/tmp/fail2ban.sock")
|
|
refreshed_path = await service.get_db_path(
|
|
"/tmp/fail2ban.sock",
|
|
force_refresh=True,
|
|
)
|
|
|
|
assert initial_path == "/tmp/fail2ban.sqlite3"
|
|
assert refreshed_path == "/tmp/fail2ban.sqlite3"
|
|
assert mock_client_cls.call_count == 2
|
|
assert second_client.send.call_count == 1
|
|
|
|
|
|
async def test_invalidate_db_path_forces_re_resolution() -> None:
|
|
"""Invalidating the cache forces a new metadata resolution."""
|
|
service = Fail2BanMetadataService()
|
|
first_client = AsyncMock()
|
|
first_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban-v1.sqlite3"))
|
|
first_client.__aenter__.return_value = first_client
|
|
first_client.__aexit__.return_value = None
|
|
|
|
second_client = AsyncMock()
|
|
second_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban-v2.sqlite3"))
|
|
second_client.__aenter__.return_value = second_client
|
|
second_client.__aexit__.return_value = None
|
|
|
|
with patch(
|
|
"app.services.fail2ban_metadata_service.Fail2BanClient",
|
|
side_effect=[first_client, second_client],
|
|
) as mock_client_cls:
|
|
first_path = await service.get_db_path("/tmp/fail2ban.sock")
|
|
service.invalidate_db_path("/tmp/fail2ban.sock")
|
|
second_path = await service.get_db_path("/tmp/fail2ban.sock")
|
|
|
|
assert first_path == "/tmp/fail2ban-v1.sqlite3"
|
|
assert second_path == "/tmp/fail2ban-v2.sqlite3"
|
|
assert mock_client_cls.call_count == 2
|