refactoring-backend #3
@@ -89,7 +89,7 @@ def since_unix(range_: str) -> int:
|
||||
Unix timestamp (seconds since epoch) representing the start of the
|
||||
time window: *now − range_ − slack*.
|
||||
"""
|
||||
from app.models.ban import TIME_RANGE_SECONDS # noqa: F401
|
||||
from app.models._common import TIME_RANGE_SECONDS
|
||||
from app.utils.constants import TIME_RANGE_SLACK_SECONDS
|
||||
|
||||
seconds: int = TIME_RANGE_SECONDS[range_]
|
||||
|
||||
@@ -143,9 +143,9 @@ class TestDashboardBanListMapper:
|
||||
|
||||
result = map_domain_dashboard_ban_list_to_response(domain_list)
|
||||
|
||||
assert result.total == 100
|
||||
assert result.page == 2
|
||||
assert result.page_size == 50
|
||||
assert result.pagination.total == 100
|
||||
assert result.pagination.page == 2
|
||||
assert result.pagination.page_size == 50
|
||||
assert len(result.items) == 1
|
||||
assert result.items[0].ip == "1.1.1.1"
|
||||
|
||||
|
||||
@@ -625,9 +625,9 @@ class TestImportLogPagination:
|
||||
db, source_id=None, page=1, page_size=10
|
||||
)
|
||||
assert resp.items == []
|
||||
assert resp.total == 0
|
||||
assert resp.page == 1
|
||||
assert resp.page_size == 10
|
||||
assert resp.pagination.total == 0
|
||||
assert resp.pagination.page == 1
|
||||
assert resp.pagination.page_size == 10
|
||||
|
||||
async def test_list_import_logs_paginates(self, db: aiosqlite.Connection) -> None:
|
||||
"""list_import_logs computes total pages and returns the correct subset."""
|
||||
@@ -646,8 +646,8 @@ class TestImportLogPagination:
|
||||
resp = await blocklist_service.list_import_logs(
|
||||
db, source_id=None, page=2, page_size=2
|
||||
)
|
||||
assert resp.total == 3
|
||||
assert resp.page == 2
|
||||
assert resp.page_size == 2
|
||||
assert resp.pagination.total == 3
|
||||
assert resp.pagination.page == 2
|
||||
assert resp.pagination.page_size == 2
|
||||
assert len(resp.items) == 1
|
||||
assert resp.items[0].source_url == "https://example0.test/ips.txt"
|
||||
|
||||
@@ -138,7 +138,7 @@ class TestListHistory:
|
||||
new=AsyncMock(return_value=f2b_db_path),
|
||||
):
|
||||
result = await history_service.list_history("fake_socket")
|
||||
assert result.total == 4
|
||||
assert result.pagination.total == 4
|
||||
assert len(result.items) == 4
|
||||
|
||||
async def test_time_range_filter_excludes_old_bans(
|
||||
@@ -153,7 +153,7 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", range_="24h"
|
||||
)
|
||||
assert result.total == 2
|
||||
assert result.pagination.total == 2
|
||||
|
||||
async def test_jail_filter(self, f2b_db_path: str) -> None:
|
||||
"""Jail filter restricts results to bans from that jail."""
|
||||
@@ -162,7 +162,7 @@ class TestListHistory:
|
||||
new=AsyncMock(return_value=f2b_db_path),
|
||||
):
|
||||
result = await history_service.list_history("fake_socket", jail="nginx")
|
||||
assert result.total == 1
|
||||
assert result.pagination.total == 1
|
||||
assert result.items[0].jail == "nginx"
|
||||
|
||||
async def test_ip_prefix_filter(self, f2b_db_path: str) -> None:
|
||||
@@ -174,7 +174,7 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", ip_filter="1.2.3"
|
||||
)
|
||||
assert result.total == 2
|
||||
assert result.pagination.total == 2
|
||||
for item in result.items:
|
||||
assert item.ip.startswith("1.2.3")
|
||||
|
||||
@@ -188,7 +188,7 @@ class TestListHistory:
|
||||
"fake_socket", jail="sshd", ip_filter="1.2.3.4"
|
||||
)
|
||||
# 2 sshd bans for 1.2.3.4
|
||||
assert result.total == 2
|
||||
assert result.pagination.total == 2
|
||||
|
||||
async def test_origin_filter_selfblock(self, f2b_db_path: str) -> None:
|
||||
"""Origin filter should include only selfblock entries."""
|
||||
@@ -200,7 +200,7 @@ class TestListHistory:
|
||||
"fake_socket", origin="selfblock"
|
||||
)
|
||||
|
||||
assert result.total == 4
|
||||
assert result.pagination.total == 4
|
||||
assert all(item.jail != "blocklist-import" for item in result.items)
|
||||
|
||||
async def test_unknown_ip_returns_empty(self, f2b_db_path: str) -> None:
|
||||
@@ -212,7 +212,7 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", ip_filter="99.99.99.99"
|
||||
)
|
||||
assert result.total == 0
|
||||
assert result.pagination.total == 0
|
||||
assert result.items == []
|
||||
|
||||
async def test_failures_extracted_from_data(
|
||||
@@ -226,7 +226,7 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", ip_filter="5.6.7.8"
|
||||
)
|
||||
assert result.total == 1
|
||||
assert result.pagination.total == 1
|
||||
assert result.items[0].failures == 3
|
||||
|
||||
async def test_matches_extracted_from_data(
|
||||
@@ -287,7 +287,7 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", ip_filter="9.0.0.1"
|
||||
)
|
||||
assert result.total == 1
|
||||
assert result.pagination.total == 1
|
||||
item = result.items[0]
|
||||
assert item.failures == 0
|
||||
assert item.matches == []
|
||||
@@ -301,10 +301,10 @@ class TestListHistory:
|
||||
result = await history_service.list_history(
|
||||
"fake_socket", page=1, page_size=2
|
||||
)
|
||||
assert result.total == 4
|
||||
assert result.pagination.total == 4
|
||||
assert len(result.items) == 2
|
||||
assert result.page == 1
|
||||
assert result.page_size == 2
|
||||
assert result.pagination.page == 1
|
||||
assert result.pagination.page_size == 2
|
||||
|
||||
async def test_source_archive_reads_from_archive(self, f2b_db_path: str, tmp_path: Path) -> None:
|
||||
"""Using source='archive' reads from the BanGUI archive table."""
|
||||
@@ -328,7 +328,7 @@ class TestListHistory:
|
||||
db=db,
|
||||
)
|
||||
|
||||
assert result.total == 1
|
||||
assert result.pagination.total == 1
|
||||
assert result.items[0].ip == "10.0.0.1"
|
||||
|
||||
|
||||
@@ -363,8 +363,8 @@ class TestGetIpDetail:
|
||||
|
||||
assert result is not None
|
||||
assert result.ip == "1.2.3.4"
|
||||
assert result.total_bans == 2
|
||||
assert result.total_failures == 10 # 5 + 5
|
||||
assert result.pagination.total_bans == 2
|
||||
assert result.pagination.total_failures == 10 # 5 + 5
|
||||
|
||||
async def test_timeline_ordered_newest_first(
|
||||
self, f2b_db_path: str
|
||||
|
||||
Reference in New Issue
Block a user