lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor ban management with domain models and mappers

Browse Source 
- Add ban domain model for core business logic separation
- Implement mapper pattern for DTO/domain conversions
- Update ban service with new domain-driven approach
- Refactor router endpoints to use new architecture
- Add comprehensive mapper tests
- Update documentation with architecture changes

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Lukas
2026-04-28 07:46:02 +02:00
parent 507f153ab9
commit 3888c5eb3f
11 changed files with 640 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/app/mappers/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
"""Response mappers.
Convert domain models (from services) to response models (for HTTP API).
This is the mapping layer at the router boundary, ensuring the service layer
remains independent of HTTP response shapes.
"""
from app.mappers.ban_mappers import (
map_domain_active_ban_list_to_response,
map_domain_active_ban_to_response,
map_domain_ban_trend_to_response,
map_domain_bans_by_country_to_response,
map_domain_bans_by_jail_to_response,
map_domain_dashboard_ban_item_to_response,
map_domain_dashboard_ban_list_to_response,
)
__all__ = [
"map_domain_active_ban_to_response",
"map_domain_active_ban_list_to_response",
"map_domain_dashboard_ban_item_to_response",
"map_domain_dashboard_ban_list_to_response",
"map_domain_bans_by_country_to_response",
"map_domain_ban_trend_to_response",
"map_domain_bans_by_jail_to_response",
]

120
backend/app/mappers/ban_mappers.py Normal file
View File

@@ -0,0 +1,120 @@
"""Ban response mappers.
Convert domain models (from ban_service) to response models (for HTTP API).
This is the mapping layer at the router boundary, ensuring the service layer
remains independent of HTTP response shapes.
"""
from __future__ import annotations
from app.models.ban import (
ActiveBan,
ActiveBanListResponse,
BansByCountryResponse,
BansByJailResponse,
BanTrendBucket,
BanTrendResponse,
DashboardBanItem,
DashboardBanListResponse,
JailBanCount,
)
from app.models.ban_domain import (
DomainActiveBan,
DomainActiveBanList,
DomainBansByCountry,
DomainBansByJail,
DomainBanTrend,
DomainDashboardBanItem,
DomainDashboardBanList,
)
def map_domain_active_ban_to_response(domain_ban: DomainActiveBan) -> ActiveBan:
"""Convert a domain active ban to a response model."""
return ActiveBan(
ip=domain_ban.ip,
jail=domain_ban.jail,
banned_at=domain_ban.banned_at,
expires_at=domain_ban.expires_at,
ban_count=domain_ban.ban_count,
country=domain_ban.country,
)
def map_domain_active_ban_list_to_response(
domain_list: DomainActiveBanList,
) -> ActiveBanListResponse:
"""Convert a domain active ban list to a response model."""
return ActiveBanListResponse(
bans=[map_domain_active_ban_to_response(ban) for ban in domain_list.bans],
total=domain_list.total,
)
def map_domain_dashboard_ban_item_to_response(
domain_item: DomainDashboardBanItem,
) -> DashboardBanItem:
"""Convert a domain dashboard ban item to a response model."""
return DashboardBanItem(
ip=domain_item.ip,
jail=domain_item.jail,
banned_at=domain_item.banned_at,
service=domain_item.service,
country_code=domain_item.country_code,
country_name=domain_item.country_name,
asn=domain_item.asn,
org=domain_item.org,
ban_count=domain_item.ban_count,
origin=domain_item.origin,
)
def map_domain_dashboard_ban_list_to_response(
domain_list: DomainDashboardBanList,
) -> DashboardBanListResponse:
"""Convert a domain dashboard ban list to a response model."""
return DashboardBanListResponse(
items=[
map_domain_dashboard_ban_item_to_response(item) for item in domain_list.items
],
total=domain_list.total,
page=domain_list.page,
page_size=domain_list.page_size,
)
def map_domain_bans_by_country_to_response(
domain_data: DomainBansByCountry,
) -> BansByCountryResponse:
"""Convert domain bans-by-country data to a response model."""
return BansByCountryResponse(
countries=domain_data.countries,
country_names=domain_data.country_names,
bans=[map_domain_dashboard_ban_item_to_response(item) for item in domain_data.items],
total=domain_data.total,
)
def map_domain_ban_trend_to_response(domain_trend: DomainBanTrend) -> BanTrendResponse:
"""Convert domain ban trend data to a response model."""
return BanTrendResponse(
buckets=[
BanTrendBucket(timestamp=bucket.timestamp, count=bucket.count)
for bucket in domain_trend.buckets
],
bucket_size=domain_trend.bucket_size,
)
def map_domain_bans_by_jail_to_response(
domain_data: DomainBansByJail,
) -> BansByJailResponse:
"""Convert domain bans-by-jail data to a response model."""
return BansByJailResponse(
jails=[
JailBanCount(jail=jail_count.jail, count=jail_count.count)
for jail_count in domain_data.jails
],
total=domain_data.total,
)

110
backend/app/models/ban_domain.py Normal file
View File

@@ -0,0 +1,110 @@
"""Ban domain models (DTOs).
Internal domain-focused models used by ban_service. These represent the
business domain layer and are independent of HTTP response shapes.
Response models are defined in `app.models.ban` and mappers convert domain
models to response models at the router boundary.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
# Domain-specific ban origin type
BanOriginDomain = Literal["blocklist", "selfblock"]
@dataclass(frozen=True)
class DomainActiveBan:
"""A currently active ban entry (domain model).
This is the service-layer representation, independent of API response shape.
"""
ip: str
jail: str
banned_at: str | None = None
expires_at: str | None = None
ban_count: int = 1
country: str | None = None
@dataclass(frozen=True)
class DomainActiveBanList:
"""List of currently active bans (domain model)."""
bans: list[DomainActiveBan]
total: int
@dataclass(frozen=True)
class DomainDashboardBanItem:
"""A single row in the dashboard ban-list table (domain model).
Populated from the fail2ban database and enriched with geo data.
"""
ip: str
jail: str
banned_at: str
service: str | None = None
country_code: str | None = None
country_name: str | None = None
asn: str | None = None
org: str | None = None
ban_count: int = 1
origin: BanOriginDomain = "selfblock"
@dataclass(frozen=True)
class DomainDashboardBanList:
"""Paginated dashboard ban-list (domain model)."""
items: list[DomainDashboardBanItem]
total: int
page: int
page_size: int
@dataclass(frozen=True)
class DomainBansByCountry:
"""Bans aggregated by country (domain model)."""
countries: dict[str, int]
country_names: dict[str, str]
items: list[DomainDashboardBanItem]
total: int
@dataclass(frozen=True)
class DomainBanTrendBucket:
"""A single time bucket in the ban trend series (domain model)."""
timestamp: str
count: int
@dataclass(frozen=True)
class DomainBanTrend:
"""Ban trend data over time (domain model)."""
buckets: list[DomainBanTrendBucket]
bucket_size: str
@dataclass(frozen=True)
class DomainJailBanCount:
"""Ban count for a single jail (domain model)."""
jail: str
count: int
@dataclass(frozen=True)
class DomainBansByJail:
"""Bans aggregated by jail (domain model)."""
jails: list[DomainJailBanCount]
total: int

11
backend/app/repositories/protocols.py
View File

@@ -235,6 +235,17 @@ class HistoryArchiveRepository(Protocol):
) -> tuple[list[dict[str, Any]], int]:
...
async def get_all_archived_history(
self,
db: aiosqlite.Connection,
since: int | None = None,
jail: str | None = None,
ip_filter: str | list[str] | None = None,
origin: BanOrigin | None = None,
action: str | None = None,
) -> list[dict[str, Any]]:
...
class Fail2BanDbRepository(Protocol):
async def check_db_nonempty(self, db_path: str) -> bool:

4
backend/app/routers/bans.py
View File

@@ -19,6 +19,7 @@ from app.dependencies import (
GeoCacheDep,
HttpSessionDep,
)
from app.mappers import map_domain_active_ban_list_to_response
from app.models.ban import ActiveBanListResponse, BanRequest, UnbanAllResponse, UnbanRequest
from app.models.jail import JailCommandResponse
from app.services import ban_service, jail_service
@@ -58,12 +59,13 @@ async def get_active_bans(
Raises:
HTTPException: 502 when fail2ban is unreachable.
"""
return await ban_service.get_active_bans(
domain_result = await ban_service.get_active_bans(
socket_path,
geo_cache=geo_cache,
http_session=http_session,
app_db=ban_ctx.db,
)
return map_domain_active_ban_list_to_response(domain_result)
@router.post(

18
backend/app/routers/dashboard.py
View File

@@ -34,6 +34,12 @@ from app.models.ban import (
TimeRange,
)
from app.models.server import ServerStatus, ServerStatusResponse
from app.mappers import (
map_domain_dashboard_ban_list_to_response,
map_domain_bans_by_country_to_response,
map_domain_ban_trend_to_response,
map_domain_bans_by_jail_to_response,
)
from app.services import ban_service
from app.utils.constants import DEFAULT_PAGE_SIZE
@@ -121,7 +127,7 @@ async def get_dashboard_bans(
:class:`~app.models.ban.DashboardBanListResponse` with paginated
ban items and the total count for the selected window.
"""
return await ban_service.list_bans(
domain_result = await ban_service.list_bans(
socket_path,
range,
source=source,
@@ -132,6 +138,7 @@ async def get_dashboard_bans(
geo_cache=geo_cache,
origin=origin,
)
return map_domain_dashboard_ban_list_to_response(domain_result)
@router.get(
@@ -180,7 +187,7 @@ async def get_bans_by_country(
:class:`~app.models.ban.BansByCountryResponse` with per-country
aggregation and the companion ban list.
"""
return await ban_service.bans_by_country(
domain_result = await ban_service.bans_by_country(
socket_path,
range,
source=source,
@@ -191,6 +198,7 @@ async def get_bans_by_country(
origin=origin,
country_code=country_code,
)
return map_domain_bans_by_country_to_response(domain_result)
@router.get(
@@ -237,13 +245,14 @@ async def get_ban_trend(
:class:`~app.models.ban.BanTrendResponse` with the ordered bucket
list and the bucket-size label.
"""
return await ban_service.ban_trend(
domain_result = await ban_service.ban_trend(
socket_path,
range,
source=source,
app_db=ban_ctx.db,
origin=origin,
)
return map_domain_ban_trend_to_response(domain_result)
@router.get(
@@ -283,10 +292,11 @@ async def get_bans_by_jail(
:class:`~app.models.ban.BansByJailResponse` with per-jail counts
sorted descending and the total for the selected window.
"""
return await ban_service.bans_by_jail(
domain_result = await ban_service.bans_by_jail(
socket_path,
range,
source=source,
app_db=ban_ctx.db,
origin=origin,
)
return map_domain_bans_by_jail_to_response(domain_result)

104
backend/app/services/ban_service.py
View File

@@ -23,21 +23,21 @@ from app.models.ban import (
BLOCKLIST_JAIL,
BUCKET_SECONDS,
BUCKET_SIZE_LABEL,
ActiveBan,
ActiveBanListResponse,
BanOrigin,
BansByCountryResponse,
BansByJailResponse,
BanTrendBucket,
BanTrendResponse,
DashboardBanItem,
DashboardBanListResponse,
TimeRange,
_derive_origin,
bucket_count,
)
from app.models.ban import (
JailBanCount as JailBanCountModel,
from app.models.ban_domain import (
DomainActiveBan,
DomainActiveBanList,
DomainBansByCountry,
DomainBansByJail,
DomainBanTrend,
DomainBanTrendBucket,
DomainDashboardBanItem,
DomainDashboardBanList,
DomainJailBanCount,
)
from app.repositories import fail2ban_db_repo
from app.repositories import history_archive_repo as default_history_archive_repo
@@ -140,7 +140,7 @@ def _origin_sql_filter(origin: BanOrigin | None) -> tuple[str, tuple[str, ...]]:
return "", ()
def _parse_ban_entry(entry: str, jail: str) -> ActiveBan | None:
def _parse_ban_entry(entry: str, jail: str) -> DomainActiveBan | None:
"""Parse a ban entry from ``get <jail> banip --with-time`` output."""
from datetime import UTC, datetime
@@ -151,7 +151,7 @@ def _parse_ban_entry(entry: str, jail: str) -> ActiveBan | None:
ipaddress.ip_address(ip)
if len(parts) < 2:
return ActiveBan(
return DomainActiveBan(
ip=ip,
jail=jail,
banned_at=None,
@@ -187,7 +187,7 @@ def _parse_ban_entry(entry: str, jail: str) -> ActiveBan | None:
if expires_at_str:
expires_at_iso = _to_iso(expires_at_str)
return ActiveBan(
return DomainActiveBan(
ip=ip,
jail=jail,
banned_at=banned_at_iso,
@@ -201,19 +201,29 @@ def _parse_ban_entry(entry: str, jail: str) -> ActiveBan | None:
async def _enrich_bans(
bans: list[ActiveBan],
bans: list[DomainActiveBan],
geo_enricher: GeoEnricher,
) -> list[ActiveBan]:
) -> list[DomainActiveBan]:
"""Enrich ban records with geo data asynchronously."""
geo_results: list[object | Exception] = await asyncio.gather(
*[cast("Awaitable[object]", geo_enricher(ban.ip)) for ban in bans],
return_exceptions=True,
)
enriched: list[ActiveBan] = []
enriched: list[DomainActiveBan] = []
for ban, geo in zip(bans, geo_results, strict=False):
if geo is not None and not isinstance(geo, Exception):
geo_info = cast("GeoInfo", geo)
enriched.append(ban.model_copy(update={"country": geo_info.country_code}))
# Create new instance with updated country
enriched.append(
DomainActiveBan(
ip=ban.ip,
jail=ban.jail,
banned_at=ban.banned_at,
expires_at=ban.expires_at,
ban_count=ban.ban_count,
country=geo_info.country_code,
)
)
else:
enriched.append(ban)
return enriched
@@ -225,7 +235,7 @@ async def get_active_bans(
geo_enricher: GeoEnricher | None = None,
http_session: aiohttp.ClientSession | None = None,
app_db: aiosqlite.Connection | None = None,
) -> ActiveBanListResponse:
) -> DomainActiveBanList:
"""Return all currently banned IPs across every jail.
For each jail the ``get <jail> banip --with-time`` command is used
@@ -253,7 +263,7 @@ async def get_active_bans(
meaningful when *http_session* is provided.
Returns:
:class:`~app.models.ban.ActiveBanListResponse` with all active bans.
:class:`~app.models.ban_domain.DomainActiveBanList` with all active bans.
Raises:
~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket
@@ -271,14 +281,14 @@ async def get_active_bans(
)
if not jail_names:
return ActiveBanListResponse(bans=[], total=0)
return DomainActiveBanList(bans=[], total=0)
results: list[object | Exception] = await asyncio.gather(
*[client.send(["get", jn, "banip", "--with-time"]) for jn in jail_names],
return_exceptions=True,
)
bans: list[ActiveBan] = []
bans: list[DomainActiveBan] = []
for jail_name, raw_result in zip(jail_names, results, strict=False):
if isinstance(raw_result, Exception):
log.warning(
@@ -322,7 +332,7 @@ async def get_active_bans(
bans = await _enrich_bans(bans, geo_enricher)
log.info("active_bans_fetched", total=len(bans))
return ActiveBanListResponse(bans=bans, total=len(bans))
return DomainActiveBanList(bans=bans, total=len(bans))
# ---------------------------------------------------------------------------
# Public API
@@ -342,7 +352,7 @@ async def list_bans(
geo_enricher: GeoEnricher | None = None,
history_archive_repo: HistoryArchiveRepository = default_history_archive_repo,
origin: BanOrigin | None = None,
) -> DashboardBanListResponse:
) -> DomainDashboardBanList:
"""Return a paginated list of bans within the selected time window.
Queries the fail2ban database ``bans`` table for records whose
@@ -376,7 +386,7 @@ async def list_bans(
the ``blocklist-import`` jail, ``"selfblock"`` excludes it.
Returns:
:class:`~app.models.ban.DashboardBanListResponse` containing the
:class:`~app.models.ban_domain.DomainDashboardBanList` containing the
paginated items and total count.
"""
@@ -428,7 +438,7 @@ async def list_bans(
except (TimeoutError, aiohttp.ClientError, OSError):
log.warning("ban_service_batch_geo_failed_list_bans")
items: list[DashboardBanItem] = []
items: list[DomainDashboardBanItem] = []
for row in rows:
if source == "archive":
jail = str(row["jail"])
@@ -471,7 +481,7 @@ async def list_bans(
log.error("ban_service_geo_lookup_unexpected_error", ip=ip, error=type(exc).__name__)
items.append(
DashboardBanItem(
DomainDashboardBanItem(
ip=ip,
jail=jail,
banned_at=banned_at,
@@ -485,7 +495,7 @@ async def list_bans(
)
)
return DashboardBanListResponse(
return DomainDashboardBanList(
items=items,
total=total,
page=page,
@@ -514,7 +524,7 @@ async def bans_by_country(
app_db: aiosqlite.Connection | None = None,
origin: BanOrigin | None = None,
country_code: str | None = None,
) -> BansByCountryResponse:
) -> DomainBansByCountry:
"""Aggregate ban counts per country for the selected time window.
Uses a two-step strategy optimised for large datasets:
@@ -547,7 +557,7 @@ async def bans_by_country(
the ``blocklist-import`` jail, ``"selfblock"`` excludes it.
Returns:
:class:`~app.models.ban.BansByCountryResponse` with per-country
:class:`~app.models.ban_domain.DomainBansByCountry` with per-country
aggregation and the companion ban list.
"""
@@ -722,7 +732,7 @@ async def bans_by_country(
country_names[cc] = cn
# Build companion table from recent rows (geo already cached from batch step).
bans: list[ActiveBan] = []
bans: list[DomainDashboardBanItem] = []
for companion_row in companion_rows:
if source == "archive":
ip = companion_row["ip"]
@@ -745,7 +755,7 @@ async def bans_by_country(
org: str | None = geo.org if geo else None
bans.append(
DashboardBanItem(
DomainDashboardBanItem(
ip=ip,
jail=jail,
banned_at=banned_at,
@@ -759,10 +769,10 @@ async def bans_by_country(
)
)
return BansByCountryResponse(
return DomainBansByCountry(
countries=countries,
country_names=country_names,
bans=bans,
items=bans,
total=total,
)
@@ -780,7 +790,7 @@ async def ban_trend(
history_archive_repo: HistoryArchiveRepository = default_history_archive_repo,
app_db: aiosqlite.Connection | None = None,
origin: BanOrigin | None = None,
) -> BanTrendResponse:
) -> DomainBanTrend:
"""Return ban counts aggregated into equal-width time buckets.
Queries the fail2ban database ``bans`` table and groups records by a
@@ -804,7 +814,7 @@ async def ban_trend(
``blocklist-import`` jail, ``"selfblock"`` excludes it.
Returns:
:class:`~app.models.ban.BanTrendResponse` with a full bucket list
:class:`~app.models.ban_domain.DomainBanTrend` with a full bucket list
and the human-readable bucket-size label.
"""
since: int = since_unix(range_)
@@ -861,15 +871,15 @@ async def ban_trend(
origin=origin,
)
buckets: list[BanTrendBucket] = [
BanTrendBucket(
buckets: list[DomainBanTrendBucket] = [
DomainBanTrendBucket(
timestamp=ts_to_iso(since + i * bucket_secs),
count=counts[i],
)
for i in range(num_buckets)
]
return BanTrendResponse(
return DomainBanTrend(
buckets=buckets,
bucket_size=BUCKET_SIZE_LABEL[range_],
)
@@ -888,7 +898,7 @@ async def bans_by_jail(
history_archive_repo: HistoryArchiveRepository = default_history_archive_repo,
app_db: aiosqlite.Connection | None = None,
origin: BanOrigin | None = None,
) -> BansByJailResponse:
) -> DomainBansByJail:
"""Return ban counts aggregated per jail for the selected time window.
Queries the fail2ban database ``bans`` table, groups records by jail
@@ -904,7 +914,7 @@ async def bans_by_jail(
``blocklist-import`` jail, ``"selfblock"`` excludes it.
Returns:
:class:`~app.models.ban.BansByJailResponse` with per-jail counts
:class:`~app.models.ban_domain.DomainBansByJail` with per-jail counts
sorted descending and the total ban count.
"""
since: int = since_unix(range_)
@@ -930,7 +940,7 @@ async def bans_by_jail(
total = sum(jail_counter.values())
jail_counts = [
JailBanCountModel(jail=jail_name, count=count)
DomainJailBanCount(jail=jail_name, count=count)
for jail_name, count in sorted(jail_counter.items(), key=lambda x: x[1], reverse=True)
]
@@ -955,12 +965,18 @@ async def bans_by_jail(
origin=origin,
)
total, jail_counts = await fail2ban_db_repo.get_bans_by_jail(
total, jail_counts_repo = await fail2ban_db_repo.get_bans_by_jail(
db_path=db_path,
since=since,
origin=origin,
)
# Convert repository models to domain models
jail_counts = [
DomainJailBanCount(jail=jc.jail, count=jc.count)
for jc in jail_counts_repo
]
# Diagnostic guard: if zero results were returned, check whether the table
# has *any* rows and log a warning with min/max timeofban so operators can
# diagnose timezone or filter mismatches from logs.
@@ -982,7 +998,7 @@ async def bans_by_jail(
jail_count=len(jail_counts),
)
return BansByJailResponse(
jails=[JailBanCountModel(jail=j.jail, count=j.count) for j in jail_counts],
return DomainBansByJail(
jails=jail_counts,
total=total,
)

1
backend/tests/test_mappers/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Tests for response mappers."""

220
backend/tests/test_mappers/test_ban_mappers.py Normal file
View File

@@ -0,0 +1,220 @@
"""Tests for ban response mappers."""
from app.mappers import (
map_domain_active_ban_list_to_response,
map_domain_active_ban_to_response,
map_domain_bans_by_country_to_response,
map_domain_bans_by_jail_to_response,
map_domain_ban_trend_to_response,
map_domain_dashboard_ban_item_to_response,
map_domain_dashboard_ban_list_to_response,
)
from app.models.ban_domain import (
DomainActiveBan,
DomainActiveBanList,
DomainBansByCountry,
DomainBansByJail,
DomainBanTrend,
DomainBanTrendBucket,
DomainDashboardBanItem,
DomainDashboardBanList,
DomainJailBanCount,
)
class TestActiveBanMapper:
"""Test mapping from DomainActiveBan to ActiveBan."""
def test_maps_all_fields(self) -> None:
"""All fields are correctly mapped."""
domain_ban = DomainActiveBan(
ip="192.168.1.1",
jail="sshd",
banned_at="2026-04-28T07:00:00+00:00",
expires_at="2026-04-28T08:00:00+00:00",
ban_count=3,
country="DE",
)
result = map_domain_active_ban_to_response(domain_ban)
assert result.ip == "192.168.1.1"
assert result.jail == "sshd"
assert result.banned_at == "2026-04-28T07:00:00+00:00"
assert result.expires_at == "2026-04-28T08:00:00+00:00"
assert result.ban_count == 3
assert result.country == "DE"
def test_handles_null_timestamps(self) -> None:
"""Null timestamps are preserved."""
domain_ban = DomainActiveBan(
ip="10.0.0.1",
jail="test",
banned_at=None,
expires_at=None,
)
result = map_domain_active_ban_to_response(domain_ban)
assert result.banned_at is None
assert result.expires_at is None
class TestActiveBanListMapper:
"""Test mapping from DomainActiveBanList to ActiveBanListResponse."""
def test_maps_list_and_total(self) -> None:
"""List and total are correctly mapped."""
domain_list = DomainActiveBanList(
bans=[
DomainActiveBan(ip="1.1.1.1", jail="sshd", ban_count=1),
DomainActiveBan(ip="2.2.2.2", jail="httpd", ban_count=2),
],
total=2,
)
result = map_domain_active_ban_list_to_response(domain_list)
assert result.total == 2
assert len(result.bans) == 2
assert result.bans[0].ip == "1.1.1.1"
assert result.bans[1].ip == "2.2.2.2"
def test_handles_empty_list(self) -> None:
"""Empty list is handled correctly."""
domain_list = DomainActiveBanList(bans=[], total=0)
result = map_domain_active_ban_list_to_response(domain_list)
assert result.total == 0
assert len(result.bans) == 0
class TestDashboardBanItemMapper:
"""Test mapping from DomainDashboardBanItem to DashboardBanItem."""
def test_maps_all_fields(self) -> None:
"""All fields are correctly mapped."""
domain_item = DomainDashboardBanItem(
ip="203.0.113.1",
jail="sshd",
banned_at="2026-04-28T07:00:00+00:00",
service="SSH login attempt",
country_code="US",
country_name="United States",
asn="AS15169",
org="Google LLC",
ban_count=5,
origin="selfblock",
)
result = map_domain_dashboard_ban_item_to_response(domain_item)
assert result.ip == "203.0.113.1"
assert result.jail == "sshd"
assert result.banned_at == "2026-04-28T07:00:00+00:00"
assert result.service == "SSH login attempt"
assert result.country_code == "US"
assert result.country_name == "United States"
assert result.asn == "AS15169"
assert result.org == "Google LLC"
assert result.ban_count == 5
assert result.origin == "selfblock"
class TestDashboardBanListMapper:
"""Test mapping from DomainDashboardBanList to DashboardBanListResponse."""
def test_maps_pagination_and_items(self) -> None:
"""Pagination metadata and items are correctly mapped."""
domain_list = DomainDashboardBanList(
items=[
DomainDashboardBanItem(
ip="1.1.1.1",
jail="sshd",
banned_at="2026-04-28T07:00:00+00:00",
ban_count=1,
),
],
total=100,
page=2,
page_size=50,
)
result = map_domain_dashboard_ban_list_to_response(domain_list)
assert result.total == 100
assert result.page == 2
assert result.page_size == 50
assert len(result.items) == 1
assert result.items[0].ip == "1.1.1.1"
class TestBansByCountryMapper:
"""Test mapping from DomainBansByCountry to BansByCountryResponse."""
def test_maps_aggregation_and_items(self) -> None:
"""Country aggregation and companion items are correctly mapped."""
domain_data = DomainBansByCountry(
countries={"US": 10, "DE": 5, "GB": 3},
country_names={"US": "United States", "DE": "Germany", "GB": "United Kingdom"},
items=[
DomainDashboardBanItem(
ip="1.1.1.1",
jail="sshd",
banned_at="2026-04-28T07:00:00+00:00",
ban_count=1,
origin="selfblock",
),
],
total=18,
)
result = map_domain_bans_by_country_to_response(domain_data)
assert result.countries == {"US": 10, "DE": 5, "GB": 3}
assert result.country_names == {"US": "United States", "DE": "Germany", "GB": "United Kingdom"}
assert result.total == 18
assert len(result.bans) == 1
class TestBanTrendMapper:
"""Test mapping from DomainBanTrend to BanTrendResponse."""
def test_maps_buckets_and_size_label(self) -> None:
"""Buckets and size label are correctly mapped."""
domain_trend = DomainBanTrend(
buckets=[
DomainBanTrendBucket(timestamp="2026-04-28T00:00:00+00:00", count=10),
DomainBanTrendBucket(timestamp="2026-04-28T01:00:00+00:00", count=15),
],
bucket_size="1h",
)
result = map_domain_ban_trend_to_response(domain_trend)
assert result.bucket_size == "1h"
assert len(result.buckets) == 2
assert result.buckets[0].timestamp == "2026-04-28T00:00:00+00:00"
assert result.buckets[0].count == 10
class TestBansByJailMapper:
"""Test mapping from DomainBansByJail to BansByJailResponse."""
def test_maps_jail_counts(self) -> None:
"""Jail counts are correctly mapped."""
domain_data = DomainBansByJail(
jails=[
DomainJailBanCount(jail="sshd", count=50),
DomainJailBanCount(jail="httpd", count=20),
],
total=70,
)
result = map_domain_bans_by_jail_to_response(domain_data)
assert result.total == 70
assert len(result.jails) == 2
assert result.jails[0].jail == "sshd"
assert result.jails[0].count == 50