From 6fc1a2a3ea8b7d2f6cc616af890513464d05d63c Mon Sep 17 00:00:00 2001 From: anti Date: Thu, 9 Apr 2026 16:43:49 -0400 Subject: [PATCH] test: refactor suite to use AsyncClient, in-memory DBs, and parallel coverage --- pyproject.toml | 3 + tests/api/auth/test_change_pass.py | 87 +++++++++++++-------------- tests/api/auth/test_login.py | 69 +++++++++++---------- tests/api/bounty/test_get_bounties.py | 34 +++++------ tests/api/conftest.py | 55 +++++++++++------ tests/api/fleet/test_get_deckies.py | 24 +++----- tests/api/logs/test_get_logs.py | 70 +++++++++------------ tests/api/logs/test_histogram.py | 7 ++- tests/api/stats/test_get_stats.py | 82 +++++++++++-------------- tests/api/test_repository.py | 12 +++- tests/api/test_schemathesis.py | 2 + 11 files changed, 230 insertions(+), 215 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 88708b1..b3e39cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ "bcrypt>=4.1.0", "psutil>=5.9.0", "python-dotenv>=1.0.0", + "sqlmodel>=0.0.16", ] [project.optional-dependencies] @@ -34,6 +35,7 @@ dev = [ "pytest-asyncio>=1.0", "freezegun>=1.5", "schemathesis>=4.0", + "pytest-xdist>=3.8.0", ] [project.scripts] @@ -49,6 +51,7 @@ filterwarnings = [ [tool.coverage.run] source = ["decnet"] omit = ["*/tests/*", "templates/*"] +parallel = true [tool.coverage.report] show_missing = true diff --git a/tests/api/auth/test_change_pass.py b/tests/api/auth/test_change_pass.py index 2830d74..0940410 100644 --- a/tests/api/auth/test_change_pass.py +++ b/tests/api/auth/test_change_pass.py @@ -1,61 +1,60 @@ import json -from fastapi.testclient import TestClient -from decnet.web.api import app +import pytest from hypothesis import given, strategies as st, settings import httpx from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD from ..conftest import _FUZZ_SETTINGS -def test_change_password() -> None: - with TestClient(app) as client: - # First login to get token - login_resp = client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - token = login_resp.json()["access_token"] +@pytest.mark.anyio +async def test_change_password(client: httpx.AsyncClient) -> None: + # First login to get token + login_resp = await client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) + token = login_resp.json()["access_token"] - # Try changing password with wrong old password - resp1 = client.post( - "/api/v1/auth/change-password", - json={"old_password": "wrong", "new_password": "new_secure_password"}, - headers={"Authorization": f"Bearer {token}"} - ) - assert resp1.status_code == 401 + # Try changing password with wrong old password + resp1 = await client.post( + "/api/v1/auth/change-password", + json={"old_password": "wrong", "new_password": "new_secure_password"}, + headers={"Authorization": f"Bearer {token}"} + ) + assert resp1.status_code == 401 - # Change password successfully - resp2 = client.post( - "/api/v1/auth/change-password", - json={"old_password": DECNET_ADMIN_PASSWORD, "new_password": "new_secure_password"}, - headers={"Authorization": f"Bearer {token}"} - ) - assert resp2.status_code == 200 + # Change password successfully + resp2 = await client.post( + "/api/v1/auth/change-password", + json={"old_password": DECNET_ADMIN_PASSWORD, "new_password": "new_secure_password"}, + headers={"Authorization": f"Bearer {token}"} + ) + assert resp2.status_code == 200 - # Verify old password no longer works - resp3 = client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - assert resp3.status_code == 401 + # Verify old password no longer works + resp3 = await client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) + assert resp3.status_code == 401 - # Verify new password works and must_change_password is False - resp4 = client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": "new_secure_password"}) - assert resp4.status_code == 200 - assert resp4.json()["must_change_password"] is False + # Verify new password works and must_change_password is False + resp4 = await client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": "new_secure_password"}) + assert resp4.status_code == 200 + assert resp4.json()["must_change_password"] is False +@pytest.mark.anyio @settings(**_FUZZ_SETTINGS) @given( old_password=st.text(min_size=0, max_size=2048), new_password=st.text(min_size=0, max_size=2048) ) -def test_fuzz_change_password(old_password: str, new_password: str) -> None: +async def test_fuzz_change_password(client: httpx.AsyncClient, old_password: str, new_password: str) -> None: """Fuzz the change-password endpoint with random strings.""" - with TestClient(app) as _client: - # Get valid token first - _login_resp: httpx.Response = _client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - _token: str = _login_resp.json()["access_token"] - - _payload: dict[str, str] = {"old_password": old_password, "new_password": new_password} - try: - _response: httpx.Response = _client.post( - "/api/v1/auth/change-password", - json=_payload, - headers={"Authorization": f"Bearer {_token}"} - ) - assert _response.status_code in (200, 401, 422) - except (UnicodeEncodeError, json.JSONDecodeError): - pass + # Get valid token first + _login_resp: httpx.Response = await client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) + _token: str = _login_resp.json()["access_token"] + + _payload: dict[str, str] = {"old_password": old_password, "new_password": new_password} + try: + _response: httpx.Response = await client.post( + "/api/v1/auth/change-password", + json=_payload, + headers={"Authorization": f"Bearer {_token}"} + ) + assert _response.status_code in (200, 401, 422) + except (UnicodeEncodeError, json.JSONDecodeError): + pass diff --git a/tests/api/auth/test_login.py b/tests/api/auth/test_login.py index 220b7b4..a5a512c 100644 --- a/tests/api/auth/test_login.py +++ b/tests/api/auth/test_login.py @@ -1,4 +1,5 @@ import json +import pytest from fastapi.testclient import TestClient from decnet.web.api import app from hypothesis import given, strategies as st, settings @@ -6,44 +7,46 @@ import httpx from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD from ..conftest import _FUZZ_SETTINGS -def test_login_success() -> None: - with TestClient(app) as client: - response = client.post( - "/api/v1/auth/login", - json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD} - ) - assert response.status_code == 200 - data = response.json() - assert "access_token" in data - assert data["token_type"] == "bearer" - assert "must_change_password" in data - assert data["must_change_password"] is True +@pytest.mark.anyio +async def test_login_success(client: httpx.AsyncClient) -> None: + response = await client.post( + "/api/v1/auth/login", + json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD} + ) + assert response.status_code == 200 + data = response.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + assert "must_change_password" in data + assert data["must_change_password"] is True -def test_login_failure() -> None: - with TestClient(app) as client: - response = client.post( - "/api/v1/auth/login", - json={"username": DECNET_ADMIN_USER, "password": "wrongpassword"} - ) - assert response.status_code == 401 - - response = client.post( - "/api/v1/auth/login", - json={"username": "nonexistent", "password": "wrongpassword"} - ) - assert response.status_code == 401 +@pytest.mark.anyio +async def test_login_failure(client: httpx.AsyncClient) -> None: + response = await client.post( + "/api/v1/auth/login", + json={"username": DECNET_ADMIN_USER, "password": "wrongpassword"} + ) + assert response.status_code == 401 + response = await client.post( + "/api/v1/auth/login", + json={"username": "nonexistent", "password": "wrongpassword"} + ) + assert response.status_code == 401 + +@pytest.mark.anyio @settings(**_FUZZ_SETTINGS) @given( username=st.text(min_size=0, max_size=2048), password=st.text(min_size=0, max_size=2048) ) -def test_fuzz_login(username: str, password: str) -> None: + +@pytest.mark.anyio +async def test_fuzz_login(client: httpx.AsyncClient, username: str, password: str) -> None: """Fuzz the login endpoint with random strings (including non-ASCII).""" - with TestClient(app) as _client: - _payload: dict[str, str] = {"username": username, "password": password} - try: - _response: httpx.Response = _client.post("/api/v1/auth/login", json=_payload) - assert _response.status_code in (200, 401, 422) - except (UnicodeEncodeError, json.JSONDecodeError): - pass + _payload: dict[str, str] = {"username": username, "password": password} + try: + _response: httpx.Response = await client.post("/api/v1/auth/login", json=_payload) + assert _response.status_code in (200, 401, 422) + except (UnicodeEncodeError, json.JSONDecodeError): + pass diff --git a/tests/api/bounty/test_get_bounties.py b/tests/api/bounty/test_get_bounties.py index 11a331f..0c21c4a 100644 --- a/tests/api/bounty/test_get_bounties.py +++ b/tests/api/bounty/test_get_bounties.py @@ -1,19 +1,19 @@ -from fastapi.testclient import TestClient -from decnet.web.api import app +import pytest +import httpx -def test_add_and_get_bounty(auth_token): - with TestClient(app) as client: - # We can't directly call add_bounty from API yet (it's internal to ingester) - # But we can test the endpoint returns 200 even if empty. - resp = client.get("/api/v1/bounty", headers={"Authorization": f"Bearer {auth_token}"}) - assert resp.status_code == 200 - data = resp.json() - assert "total" in data - assert "data" in data - assert isinstance(data["data"], list) +@pytest.mark.anyio +async def test_add_and_get_bounty(client: httpx.AsyncClient, auth_token: str): + # We can't directly call add_bounty from API yet (it's internal to ingester) + # But we can test the endpoint returns 200 even if empty. + resp = await client.get("/api/v1/bounty", headers={"Authorization": f"Bearer {auth_token}"}) + assert resp.status_code == 200 + data = resp.json() + assert "total" in data + assert "data" in data + assert isinstance(data["data"], list) -def test_bounty_pagination(auth_token): - with TestClient(app) as client: - resp = client.get("/api/v1/bounty?limit=1&offset=0", headers={"Authorization": f"Bearer {auth_token}"}) - assert resp.status_code == 200 - assert resp.json()["limit"] == 1 +@pytest.mark.anyio +async def test_bounty_pagination(client: httpx.AsyncClient, auth_token: str): + resp = await client.get("/api/v1/bounty?limit=1&offset=0", headers={"Authorization": f"Bearer {auth_token}"}) + assert resp.status_code == 200 + assert resp.json()["limit"] == 1 diff --git a/tests/api/conftest.py b/tests/api/conftest.py index 195c155..38ee02c 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -1,41 +1,58 @@ import os import json import pytest -from typing import Generator, Any +from typing import Generator, Any, AsyncGenerator from pathlib import Path -from fastapi.testclient import TestClient +import httpx from hypothesis import HealthCheck +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +# Ensure required env vars are set to non-bad values for tests before anything imports decnet.env +os.environ["DECNET_JWT_SECRET"] = "test-secret-key-at-least-32-chars-long!!" +os.environ["DECNET_ADMIN_PASSWORD"] = "test-password-123" from decnet.web.api import app from decnet.web.dependencies import repo +from decnet.web.db.sqlite.database import get_async_engine from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD import decnet.config TEST_STATE_FILE = Path("test-decnet-state.json") @pytest.fixture(scope="function", autouse=True) -def setup_db() -> Generator[None, None, None]: - # Use a unique DB for each test process/thread if possible, but for now just one - repo.db_path = "test_api_decnet.db" - if os.path.exists(repo.db_path): - try: - os.remove(repo.db_path) - except OSError: - pass +async def setup_db(worker_id, monkeypatch) -> AsyncGenerator[None, None]: + import uuid + # Use worker-specific in-memory DB with shared cache for maximum speed + unique_id = uuid.uuid4().hex + db_path = f"file:memdb_{worker_id}_{unique_id}?mode=memory&cache=shared" + # Patch the global repo singleton + monkeypatch.setattr(repo, "db_path", db_path) + + engine = get_async_engine(db_path) + session_factory = async_sessionmaker( + engine, class_=AsyncSession, expire_on_commit=False + ) + + monkeypatch.setattr(repo, "engine", engine) + monkeypatch.setattr(repo, "session_factory", session_factory) + + # Initialize the in-memory DB (tables + admin) repo.reinitialize() + yield - if os.path.exists(repo.db_path): - try: - os.remove(repo.db_path) - except OSError: - pass + + await engine.dispose() @pytest.fixture -def auth_token() -> str: - with TestClient(app) as client: - resp = client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - return resp.json()["access_token"] +async def client() -> AsyncGenerator[httpx.AsyncClient, None]: + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as ac: + yield ac + +@pytest.fixture +async def auth_token(client: httpx.AsyncClient) -> str: + resp = await client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) + return resp.json()["access_token"] @pytest.fixture(autouse=True) def patch_state_file(monkeypatch): diff --git a/tests/api/fleet/test_get_deckies.py b/tests/api/fleet/test_get_deckies.py index a7d6189..036a9fc 100644 --- a/tests/api/fleet/test_get_deckies.py +++ b/tests/api/fleet/test_get_deckies.py @@ -1,16 +1,12 @@ -from fastapi.testclient import TestClient -from decnet.web.api import app +import pytest +import httpx from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD -def test_get_deckies_endpoint(mock_state_file): - with TestClient(app) as _client: - # Login to get token - _login_resp = _client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - _token = _login_resp.json()["access_token"] - - _response = _client.get("/api/v1/deckies", headers={"Authorization": f"Bearer {_token}"}) - assert _response.status_code == 200 - _data = _response.json() - assert len(_data) == 2 - assert _data[0]["name"] == "test-decky-1" - assert _data[0]["service_config"]["ssh"]["banner"] == "SSH-2.0-OpenSSH_8.9" +@pytest.mark.anyio +async def test_get_deckies_endpoint(mock_state_file, client: httpx.AsyncClient, auth_token: str): + _response = await client.get("/api/v1/deckies", headers={"Authorization": f"Bearer {auth_token}"}) + assert _response.status_code == 200 + _data = _response.json() + assert len(_data) == 2 + assert _data[0]["name"] == "test-decky-1" + assert _data[0]["service_config"]["ssh"]["banner"] == "SSH-2.0-OpenSSH_8.9" diff --git a/tests/api/logs/test_get_logs.py b/tests/api/logs/test_get_logs.py index 12471bb..4952b19 100644 --- a/tests/api/logs/test_get_logs.py +++ b/tests/api/logs/test_get_logs.py @@ -1,53 +1,43 @@ -from typing import Any, Optional -from fastapi.testclient import TestClient -from decnet.web.api import app -from hypothesis import given, strategies as st, settings +import pytest import httpx +from typing import Any, Optional from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD from ..conftest import _FUZZ_SETTINGS +from hypothesis import given, strategies as st, settings -def test_get_logs_unauthorized() -> None: - with TestClient(app) as client: - response = client.get("/api/v1/logs") - assert response.status_code == 401 +@pytest.mark.anyio +async def test_get_logs_unauthorized(client: httpx.AsyncClient) -> None: + response = await client.get("/api/v1/logs") + assert response.status_code == 401 -def test_get_logs_success() -> None: - with TestClient(app) as client: - login_response = client.post( - "/api/v1/auth/login", - json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD} - ) - token = login_response.json()["access_token"] - - response = client.get( - "/api/v1/logs", - headers={"Authorization": f"Bearer {token}"} - ) - assert response.status_code == 200 - data = response.json() - assert "data" in data - assert data["total"] >= 0 - assert isinstance(data["data"], list) +@pytest.mark.anyio +async def test_get_logs_success(client: httpx.AsyncClient, auth_token: str) -> None: + response = await client.get( + "/api/v1/logs", + headers={"Authorization": f"Bearer {auth_token}"} + ) + assert response.status_code == 200 + data = response.json() + assert "data" in data + assert data["total"] >= 0 + assert isinstance(data["data"], list) +@pytest.mark.anyio @settings(**_FUZZ_SETTINGS) @given( limit=st.integers(min_value=-2000, max_value=5000), offset=st.integers(min_value=-2000, max_value=5000), search=st.one_of(st.none(), st.text(max_size=2048)) ) -def test_fuzz_get_logs(limit: int, offset: int, search: Optional[str]) -> None: - with TestClient(app) as _client: - _login_resp: httpx.Response = _client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - _token: str = _login_resp.json()["access_token"] +async def test_fuzz_get_logs(client: httpx.AsyncClient, auth_token: str, limit: int, offset: int, search: Optional[str]) -> None: + _params: dict[str, Any] = {"limit": limit, "offset": offset} + if search is not None: + _params["search"] = search - _params: dict[str, Any] = {"limit": limit, "offset": offset} - if search is not None: - _params["search"] = search - - _response: httpx.Response = _client.get( - "/api/v1/logs", - params=_params, - headers={"Authorization": f"Bearer {_token}"} - ) - - assert _response.status_code in (200, 422) + _response: httpx.Response = await client.get( + "/api/v1/logs", + params=_params, + headers={"Authorization": f"Bearer {auth_token}"} + ) + + assert _response.status_code in (200, 422) diff --git a/tests/api/logs/test_histogram.py b/tests/api/logs/test_histogram.py index 608c13e..ae8a984 100644 --- a/tests/api/logs/test_histogram.py +++ b/tests/api/logs/test_histogram.py @@ -9,7 +9,7 @@ import json import pytest from datetime import datetime, timedelta from freezegun import freeze_time -from decnet.web.sqlite_repository import SQLiteRepository +from decnet.web.db.sqlite.repository import SQLiteRepository @pytest.fixture @@ -30,11 +30,13 @@ def _log(decky="d", service="ssh", ip="1.2.3.4", timestamp=None): } +@pytest.mark.anyio async def test_histogram_empty_db(repo): result = await repo.get_log_histogram() assert result == [] +@pytest.mark.anyio @freeze_time("2026-04-09 12:00:00") async def test_histogram_single_bucket(repo): now = datetime.now() @@ -48,6 +50,7 @@ async def test_histogram_single_bucket(repo): assert result[0]["count"] == 5 +@pytest.mark.anyio @freeze_time("2026-04-09 12:00:00") async def test_histogram_two_buckets(repo): now = datetime.now() @@ -65,6 +68,7 @@ async def test_histogram_two_buckets(repo): assert counts == {3, 7} +@pytest.mark.anyio @freeze_time("2026-04-09 12:00:00") async def test_histogram_respects_start_end_filter(repo): now = datetime.now() @@ -82,6 +86,7 @@ async def test_histogram_respects_start_end_filter(repo): assert total == 1 +@pytest.mark.anyio @freeze_time("2026-04-09 12:00:00") async def test_histogram_search_filter(repo): now = datetime.now() diff --git a/tests/api/stats/test_get_stats.py b/tests/api/stats/test_get_stats.py index 7849530..de39644 100644 --- a/tests/api/stats/test_get_stats.py +++ b/tests/api/stats/test_get_stats.py @@ -1,58 +1,48 @@ -from typing import Any -from fastapi.testclient import TestClient -from decnet.web.api import app -from hypothesis import given, strategies as st, settings +import pytest import httpx +from typing import Any from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD from ..conftest import _FUZZ_SETTINGS +from hypothesis import given, strategies as st, settings -def test_get_stats_unauthorized() -> None: - with TestClient(app) as client: - response = client.get("/api/v1/stats") - assert response.status_code == 401 +@pytest.mark.anyio +async def test_get_stats_unauthorized(client: httpx.AsyncClient) -> None: + response = await client.get("/api/v1/stats") + assert response.status_code == 401 -def test_get_stats_success() -> None: - with TestClient(app) as client: - login_response = client.post( - "/api/v1/auth/login", - json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD} - ) - token = login_response.json()["access_token"] - - response = client.get( - "/api/v1/stats", - headers={"Authorization": f"Bearer {token}"} - ) - assert response.status_code == 200 - data = response.json() - assert "total_logs" in data - assert "unique_attackers" in data - assert "active_deckies" in data +@pytest.mark.anyio +async def test_get_stats_success(client: httpx.AsyncClient, auth_token: str) -> None: + response = await client.get( + "/api/v1/stats", + headers={"Authorization": f"Bearer {auth_token}"} + ) + assert response.status_code == 200 + data = response.json() + assert "total_logs" in data + assert "unique_attackers" in data + assert "active_deckies" in data -def test_stats_includes_deployed_count(mock_state_file): - with TestClient(app) as _client: - _login_resp = _client.post("/api/v1/auth/login", json={"username": DECNET_ADMIN_USER, "password": DECNET_ADMIN_PASSWORD}) - _token = _login_resp.json()["access_token"] - - _response = _client.get("/api/v1/stats", headers={"Authorization": f"Bearer {_token}"}) - assert _response.status_code == 200 - _data = _response.json() - assert "deployed_deckies" in _data - assert _data["deployed_deckies"] == 2 +@pytest.mark.anyio +async def test_stats_includes_deployed_count(mock_state_file, client: httpx.AsyncClient, auth_token: str): + _response = await client.get("/api/v1/stats", headers={"Authorization": f"Bearer {auth_token}"}) + assert _response.status_code == 200 + _data = _response.json() + assert "deployed_deckies" in _data + assert _data["deployed_deckies"] == 2 +@pytest.mark.anyio @settings(**_FUZZ_SETTINGS) @given( token=st.text(min_size=0, max_size=4096) ) -def test_fuzz_auth_header(token: str) -> None: +async def test_fuzz_auth_header(client: httpx.AsyncClient, token: str) -> None: """Fuzz the Authorization header with full unicode noise.""" - with TestClient(app) as _client: - try: - _response: httpx.Response = _client.get( - "/api/v1/stats", - headers={"Authorization": f"Bearer {token}"} - ) - assert _response.status_code in (401, 422) - except (UnicodeEncodeError, httpx.InvalidURL, httpx.CookieConflict): - # Expected client-side rejection of invalid header characters - pass + try: + _response: httpx.Response = await client.get( + "/api/v1/stats", + headers={"Authorization": f"Bearer {token}"} + ) + assert _response.status_code in (401, 422) + except (UnicodeEncodeError, httpx.InvalidURL, httpx.CookieConflict): + # Expected client-side rejection of invalid header characters + pass diff --git a/tests/api/test_repository.py b/tests/api/test_repository.py index f05a244..12d7bb8 100644 --- a/tests/api/test_repository.py +++ b/tests/api/test_repository.py @@ -5,7 +5,7 @@ covering DEBT-006 (zero test coverage on the database layer). """ import json import pytest -from decnet.web.sqlite_repository import SQLiteRepository +from decnet.web.db.sqlite.repository import SQLiteRepository @pytest.fixture @@ -13,6 +13,7 @@ def repo(tmp_path): return SQLiteRepository(db_path=str(tmp_path / "test.db")) +@pytest.mark.anyio async def test_add_and_get_log(repo): await repo.add_log({ "decky": "decky-01", @@ -30,6 +31,7 @@ async def test_add_and_get_log(repo): assert logs[0]["attacker_ip"] == "10.0.0.1" +@pytest.mark.anyio async def test_get_total_logs(repo): for i in range(5): await repo.add_log({ @@ -45,6 +47,7 @@ async def test_get_total_logs(repo): assert total == 5 +@pytest.mark.anyio async def test_search_filter_by_decky(repo): await repo.add_log({"decky": "target", "service": "ssh", "event_type": "connect", "attacker_ip": "1.1.1.1", "raw_line": "x", "fields": "{}", "msg": ""}) @@ -56,6 +59,7 @@ async def test_search_filter_by_decky(repo): assert logs[0]["decky"] == "target" +@pytest.mark.anyio async def test_search_filter_by_service(repo): await repo.add_log({"decky": "d1", "service": "rdp", "event_type": "connect", "attacker_ip": "1.1.1.1", "raw_line": "x", "fields": "{}", "msg": ""}) @@ -67,6 +71,7 @@ async def test_search_filter_by_service(repo): assert logs[0]["service"] == "rdp" +@pytest.mark.anyio async def test_search_filter_by_json_field(repo): await repo.add_log({"decky": "d1", "service": "ssh", "event_type": "connect", "attacker_ip": "1.1.1.1", "raw_line": "x", @@ -80,6 +85,7 @@ async def test_search_filter_by_json_field(repo): assert json.loads(logs[0]["fields"])["username"] == "root" +@pytest.mark.anyio async def test_get_logs_after_id(repo): for i in range(4): await repo.add_log({"decky": "d", "service": "ssh", "event_type": "connect", @@ -97,6 +103,7 @@ async def test_get_logs_after_id(repo): assert len(new_logs) == 1 +@pytest.mark.anyio async def test_full_text_search(repo): await repo.add_log({"decky": "d1", "service": "ssh", "event_type": "connect", "attacker_ip": "1.1.1.1", "raw_line": "supersecretstring", @@ -109,6 +116,7 @@ async def test_full_text_search(repo): assert len(logs) == 1 +@pytest.mark.anyio async def test_pagination(repo): for i in range(10): await repo.add_log({"decky": "d", "service": "ssh", "event_type": "connect", @@ -128,6 +136,7 @@ async def test_pagination(repo): assert ids1.isdisjoint(ids2) +@pytest.mark.anyio async def test_add_and_get_bounty(repo): await repo.add_bounty({ "decky": "decky-01", @@ -142,6 +151,7 @@ async def test_add_and_get_bounty(repo): assert bounties[0]["bounty_type"] == "credentials" +@pytest.mark.anyio async def test_user_lifecycle(repo): import uuid uid = str(uuid.uuid4()) diff --git a/tests/api/test_schemathesis.py b/tests/api/test_schemathesis.py index d29063d..310f3da 100644 --- a/tests/api/test_schemathesis.py +++ b/tests/api/test_schemathesis.py @@ -11,6 +11,7 @@ replace the checks list with the default (remove the argument) for full complian Requires DECNET_DEVELOPER=true (set in tests/conftest.py) to expose /openapi.json. """ import schemathesis +from hypothesis import settings from schemathesis.checks import not_a_server_error from decnet.web.api import app @@ -18,5 +19,6 @@ schema = schemathesis.openapi.from_asgi("/openapi.json", app) @schemathesis.pytest.parametrize(api=schema) +@settings(max_examples=5, deadline=None) def test_schema_compliance(case): case.call_and_validate(checks=[not_a_server_error])