refactor: improve backend type safety and import organization
- Add TYPE_CHECKING guards for runtime-expensive imports (aiohttp, aiosqlite) - Reorganize imports to follow PEP 8 conventions - Convert TypeAlias to modern PEP 695 type syntax (where appropriate) - Use Sequence/Mapping from collections.abc for type hints (covariant) - Replace string literals with cast() for improved type inference - Fix casting of Fail2BanResponse and TypedDict patterns - Add IpLookupResult TypedDict for precise return type annotation - Reformat overlong lines for readability (120 char limit) - Add asyncio_mode and filterwarnings to pytest config - Update test fixtures with improved type hints This improves mypy type checking and makes type relationships explicit.
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
@@ -157,12 +158,12 @@ class TestRequireAuthSessionCache:
|
||||
"""In-memory session token cache inside ``require_auth``."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_cache(self) -> None: # type: ignore[misc]
|
||||
def reset_cache(self) -> Generator[None, None, None]:
|
||||
"""Flush the session cache before and after every test in this class."""
|
||||
from app import dependencies
|
||||
|
||||
dependencies.clear_session_cache()
|
||||
yield # type: ignore[misc]
|
||||
yield
|
||||
dependencies.clear_session_cache()
|
||||
|
||||
async def test_second_request_skips_db(self, client: AsyncClient) -> None:
|
||||
|
||||
@@ -70,7 +70,7 @@ class TestGeoLookup:
|
||||
async def test_200_with_geo_info(self, geo_client: AsyncClient) -> None:
|
||||
"""GET /api/geo/lookup/{ip} returns 200 with enriched result."""
|
||||
geo = GeoInfo(country_code="DE", country_name="Germany", asn="12345", org="Acme")
|
||||
result = {
|
||||
result: dict[str, object] = {
|
||||
"ip": "1.2.3.4",
|
||||
"currently_banned_in": ["sshd"],
|
||||
"geo": geo,
|
||||
@@ -92,7 +92,7 @@ class TestGeoLookup:
|
||||
|
||||
async def test_200_when_not_banned(self, geo_client: AsyncClient) -> None:
|
||||
"""GET /api/geo/lookup/{ip} returns empty list when IP is not banned anywhere."""
|
||||
result = {
|
||||
result: dict[str, object] = {
|
||||
"ip": "8.8.8.8",
|
||||
"currently_banned_in": [],
|
||||
"geo": GeoInfo(country_code="US", country_name="United States", asn=None, org=None),
|
||||
@@ -108,7 +108,7 @@ class TestGeoLookup:
|
||||
|
||||
async def test_200_with_no_geo(self, geo_client: AsyncClient) -> None:
|
||||
"""GET /api/geo/lookup/{ip} returns null geo when enricher fails."""
|
||||
result = {
|
||||
result: dict[str, object] = {
|
||||
"ip": "1.2.3.4",
|
||||
"currently_banned_in": [],
|
||||
"geo": None,
|
||||
@@ -144,7 +144,7 @@ class TestGeoLookup:
|
||||
|
||||
async def test_ipv6_address(self, geo_client: AsyncClient) -> None:
|
||||
"""GET /api/geo/lookup/{ip} handles IPv6 addresses."""
|
||||
result = {
|
||||
result: dict[str, object] = {
|
||||
"ip": "2001:db8::1",
|
||||
"currently_banned_in": [],
|
||||
"geo": None,
|
||||
|
||||
@@ -12,6 +12,7 @@ from httpx import ASGITransport, AsyncClient
|
||||
from app.config import Settings
|
||||
from app.db import init_db
|
||||
from app.main import create_app
|
||||
from app.models.ban import JailBannedIpsResponse
|
||||
from app.models.jail import Jail, JailDetailResponse, JailListResponse, JailStatus, JailSummary
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -801,17 +802,17 @@ class TestGetJailBannedIps:
|
||||
def _mock_response(
|
||||
self,
|
||||
*,
|
||||
items: list[dict] | None = None,
|
||||
items: list[dict[str, str | None]] | None = None,
|
||||
total: int = 2,
|
||||
page: int = 1,
|
||||
page_size: int = 25,
|
||||
) -> "JailBannedIpsResponse": # type: ignore[name-defined]
|
||||
) -> JailBannedIpsResponse:
|
||||
from app.models.ban import ActiveBan, JailBannedIpsResponse
|
||||
|
||||
ban_items = (
|
||||
[
|
||||
ActiveBan(
|
||||
ip=item.get("ip", "1.2.3.4"),
|
||||
ip=item.get("ip") or "1.2.3.4",
|
||||
jail="sshd",
|
||||
banned_at=item.get("banned_at", "2025-01-01T10:00:00+00:00"),
|
||||
expires_at=item.get("expires_at", "2025-01-01T10:10:00+00:00"),
|
||||
|
||||
@@ -247,9 +247,9 @@ class TestSetupCompleteCaching:
|
||||
assert not getattr(app.state, "_setup_complete_cached", False)
|
||||
|
||||
# First non-exempt request — middleware queries DB and sets the flag.
|
||||
await client.post("/api/auth/login", json={"password": _SETUP_PAYLOAD["master_password"]}) # type: ignore[call-overload]
|
||||
await client.post("/api/auth/login", json={"password": _SETUP_PAYLOAD["master_password"]})
|
||||
|
||||
assert app.state._setup_complete_cached is True # type: ignore[attr-defined]
|
||||
assert app.state._setup_complete_cached is True
|
||||
|
||||
async def test_cached_path_skips_is_setup_complete(
|
||||
self,
|
||||
@@ -267,12 +267,12 @@ class TestSetupCompleteCaching:
|
||||
|
||||
# Do setup and warm the cache.
|
||||
await client.post("/api/setup", json=_SETUP_PAYLOAD)
|
||||
await client.post("/api/auth/login", json={"password": _SETUP_PAYLOAD["master_password"]}) # type: ignore[call-overload]
|
||||
assert app.state._setup_complete_cached is True # type: ignore[attr-defined]
|
||||
await client.post("/api/auth/login", json={"password": _SETUP_PAYLOAD["master_password"]})
|
||||
assert app.state._setup_complete_cached is True
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def _counting(db): # type: ignore[no-untyped-def]
|
||||
async def _counting(db: aiosqlite.Connection) -> bool:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user