From 44de453bb2bd24bcd893143b049763ed571d35dc Mon Sep 17 00:00:00 2001 From: anti Date: Thu, 9 Apr 2026 12:32:31 -0400 Subject: [PATCH] refactor: modularize API tests to match router structure --- .claude/settings.local.json | 5 +- .hypothesis/constants/3b152726a666601e | 4 + .hypothesis/constants/b807ea9189944fb3 | 4 + .hypothesis/constants/ff35158fdfe08acb | 4 + .../unicode_data/16.0.0/codec-utf-8.json.gz | Bin 60 -> 60 bytes test_api_decnet.db-shm | Bin 0 -> 32768 bytes test_api_decnet.db-wal | 0 test_bounty_decnet.db-shm | Bin 32768 -> 32768 bytes test_bounty_decnet.db-wal | Bin 28872 -> 28872 bytes test_decnet.db-shm | Bin 32768 -> 32768 bytes test_decnet.db-wal | Bin 28872 -> 28872 bytes test_fleet_decnet.db-shm | Bin 32768 -> 32768 bytes test_fleet_decnet.db-wal | Bin 28872 -> 28872 bytes test_fuzz_decnet.db-shm | Bin 32768 -> 32768 bytes test_fuzz_decnet.db-wal | Bin 28872 -> 28872 bytes tests/test_bounty.py | 41 ------ tests/test_fleet_api.py | 99 ------------- tests/test_web_api.py | 135 ------------------ tests/test_web_api_fuzz.py | 110 -------------- 19 files changed, 16 insertions(+), 386 deletions(-) create mode 100644 .hypothesis/constants/3b152726a666601e create mode 100644 .hypothesis/constants/b807ea9189944fb3 create mode 100644 .hypothesis/constants/ff35158fdfe08acb create mode 100644 test_api_decnet.db-shm create mode 100644 test_api_decnet.db-wal delete mode 100644 tests/test_bounty.py delete mode 100644 tests/test_fleet_api.py delete mode 100644 tests/test_web_api.py delete mode 100644 tests/test_web_api_fuzz.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 02b56d2..6053293 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -1,7 +1,10 @@ { "permissions": { "allow": [ - "mcp__plugin_context-mode_context-mode__ctx_batch_execute" + "mcp__plugin_context-mode_context-mode__ctx_batch_execute", + "mcp__plugin_context-mode_context-mode__ctx_search", + "Bash(grep:*)", + "Bash(python -m pytest --tb=short -q)" ] } } diff --git a/.hypothesis/constants/3b152726a666601e b/.hypothesis/constants/3b152726a666601e new file mode 100644 index 0000000..e6e1485 --- /dev/null +++ b/.hypothesis/constants/3b152726a666601e @@ -0,0 +1,4 @@ +# file: /home/anti/Tools/DECNET/decnet/web/sqlite_repository.py +# hypothesis_version: 6.151.11 + +[0.1, ' AND ', ' WHERE ', ':', 'BEGIN IMMEDIATE', 'COMMIT', 'ROLLBACK', '[^a-zA-Z0-9_]', 'active_deckies', 'admin', 'attacker', 'attacker-ip', 'attacker_ip', 'bounty_type', 'bounty_type = ?', 'bucket_time', 'count', 'decky', 'decnet.db', 'deployed_deckies', 'event', 'event_type', 'fields', 'id > ?', 'max_id', 'msg', 'must_change_password', 'password_hash', 'payload', 'raw_line', 'role', 'service', 'time', 'timestamp', 'timestamp <= ?', 'timestamp >= ?', 'total', 'total_logs', 'unique_attackers', 'username', 'uuid'] \ No newline at end of file diff --git a/.hypothesis/constants/b807ea9189944fb3 b/.hypothesis/constants/b807ea9189944fb3 new file mode 100644 index 0000000..04c6133 --- /dev/null +++ b/.hypothesis/constants/b807ea9189944fb3 @@ -0,0 +1,4 @@ +# file: /home/anti/Tools/DECNET/decnet/web/api.py +# hypothesis_version: 6.151.11 + +[0.5, '/api/v1', '/docs', '/openapi.json', '/redoc', '1.0.0', 'Authorization', 'Content-Type', 'DELETE', 'GET', 'OPTIONS', 'POST', 'PUT'] \ No newline at end of file diff --git a/.hypothesis/constants/ff35158fdfe08acb b/.hypothesis/constants/ff35158fdfe08acb new file mode 100644 index 0000000..86a9d99 --- /dev/null +++ b/.hypothesis/constants/ff35158fdfe08acb @@ -0,0 +1,4 @@ +# file: /home/anti/Tools/DECNET/decnet/env.py +# hypothesis_version: 6.151.11 + +[',', '.env', '.env.local', '0.0.0.0', '8000', '8080', 'DECNET_ADMIN_USER', 'DECNET_API_HOST', 'DECNET_API_PORT', 'DECNET_CORS_ORIGINS', 'DECNET_DEVELOPER', 'DECNET_JWT_SECRET', 'DECNET_WEB_HOST', 'DECNET_WEB_PORT', 'False', 'admin', 'changeme', 'password', 'secret', 'true'] \ No newline at end of file diff --git a/.hypothesis/unicode_data/16.0.0/codec-utf-8.json.gz b/.hypothesis/unicode_data/16.0.0/codec-utf-8.json.gz index 864dfa3e7e14d1371a694c45dd7aee28c1868a39..c11a20d5e21389599dd57cac1610fc0a00f96abc 100644 GIT binary patch delta 27 icmcDq5tZ-e;9z86U|{-Rl3S2mR%~8rSe!XgR2=|VlLv7C delta 27 icmcDq5tZ-e;9z86U|{-Rl3P$%oST+wn3XY6R2=|W%m820NVB-B>(^b diff --git a/test_bounty_decnet.db-wal b/test_bounty_decnet.db-wal index f5ce014a5d7698aee29c9f009593a9df8cbd36f0..090940310c790558db02042b9457b1ac2fb7eca1 100644 GIT binary patch delta 362 zcmX@{knzMrMho+Lwk8JMM-mJS0t_JVh&kZm?2TW{mYK7Cohm%R6ez@qEN&chEWy%l zUF${*4gmp}Vx=w1g`WKuIk7oF07;aanbqYs&x`3mQDy;A7GzUS+|P(QZxi@#bA!P& z{`xdilO)45OB3B>3-c6RljP)7T?^x616^awq(n=Lw6w%DlhnkN+{`=`qa+nWBNczI zyb!m5V(09vz@VzCfPfsc0*`b()AZ8ByrLZQP{Wb{1DC)u=aMSFtW5oIOJfUvugqYu z0YEpfBD+KReeAR?%n`Z_n`b!O;kO~iAwcbH$VOOfdt^Fk`u(`g8v>C0K6T23(`UcG K(On!MAOHY!iGcM0 delta 362 zcmX@{knzMrMho+Lwk8JMM-mJS0t_HfyyDuSnCE+Rrte+dR+E&~4isWU7Wehb{kz-2 z>-t6u4gmp};swvTWt;yKoZ^a^4Q90T+E(8QDy;A7GzV{r}4;m?6=q0++Z+` zzuwrw*u>1z&_LHR+1Nza#KO{8H_^htT-Vav$jrph+`!z}GBGhFH#1MgC`rZ8NW~|` zE4acdLq9A(*e};3$uK{%s-m(qG`y_LFWo07ST83a#5=^yw=g6sJv%+9!lk&%$;TON z0MHGr$nH42qe!2T+3Z#R<{1um_-%-B2v9p4vJn;wA8)$8gm2^K4FO1gkNaL%^Pg|B J$l?G20RZ{&f1>~Z diff --git a/test_decnet.db-shm b/test_decnet.db-shm index a1988f3964c65e831cc1340d3a216fb5a4ca07ae..ca4a95daff183e2572e4ce6480bbec5308306fb3 100644 GIT binary patch delta 64 wcmZo@U}|V!njj(3I)*))E<_*CfBvTafOx8sr8n6RK803gmHmjD0& delta 64 vcmZo@U}|V!njj%@JgAp%&QFu(nT|)hh18my9A4;by8F!J79OdM2@C20h!q{L diff --git a/test_decnet.db-wal b/test_decnet.db-wal index bf38d6393fc6eaccad911820b6e056eb08934f95..714b9c6f98d644d40b6a33ff7d89ae1ea9a8657e 100644 GIT binary patch delta 362 zcmX@{knzMrMho+Lwk8JMM-mJS0t_HfX8*m*rt#y&_! z$x+3X0R@gOzNN1DQHGfr*?}3U?uI$p*-1eb74F7SQJxmYnMt1RCHdh_`9@{tkx38( zfNo$#b_bK$x(_p%l_wi-p5btZ--Z~60JXCr8{wn=^}Wd9*twfG1R(i+Rn^2+hodWA JEDjJ5006RPgh>DZ delta 362 zcmX@{knzMrMho+Lwk8JMM-mJS0t_H9)A4AxkXn;d!Wx$~(PB=?Kp{pHam_2QzooA% z-e|!gAOKS=xv+R)--42`%>e>PqW_vMySFPhEd`1)3y88Ho3dl^f-OfgA6IQ|Fqp<) zZ;@zZnqqEgu4`^+nWSrCkZ7u#WSN$xn_^^?W@?^lX`Gm7l9-a4nWtiuq+)2K5*gxR zQ5KMso^0%q6BJaQW}X=6pR4DdYM58$?P{6kpA_mJnU)mnUuhX?R#6;i99CjsWCS(< z=mu6~cWn5(`jHOv(#7hVXE@y9w;{$MK<#YEMzlv#S03^Tj^#&bpHu*Vc Iae#mT0GE4q$N&HU diff --git a/test_fleet_decnet.db-shm b/test_fleet_decnet.db-shm index 1b7110124758857b321d43423abe95af8441997f..771848a15a332e2231cabe75a1319430986bb531 100644 GIT binary patch delta 64 vcmZo@U}|V!njj(3?t5iI$n?H6=9|hn&R6oy9|;v#X+&`T#Ur&bVL?3rPJSB! delta 64 wcmZo@U}|V!njj%jrJDIsHQ-X+-{%tVBUJn}CU;%EwrPXXVLVbB6Bg710I(t-2><{9 diff --git a/test_fleet_decnet.db-wal b/test_fleet_decnet.db-wal index c9dd302f0cb7909ac0ce9a65e2ae875f7aa11f4d..8805b8f4ec8743c715ed89c3877fbaaac2a051b1 100644 GIT binary patch delta 360 zcmX@{knzMrMho+Lwk8JMM-mJS0t_I)d{a5c`AWXIBi9`x~l4%q&bzbWMznlXXqZlFW52lFco34bvSjYWVY-sz&a(8cnwvKSAo+a472o!>zUd*0 H0|W#Bdy#K> delta 360 zcmX@{knzMrMho+Lwk8JMM-mJS0t_JV_qoLT2o*n#!~b6kEHQn~3lw5R7JnEqYyHE# zgRL7aI0OV>inqNy%^ICG`NQS_0VL74E*9OBv>v4cMVSReS&&Vsa@;cE$(75zn;Q(K z^VcVv8<-}inCKcOrJ3uRm|L3YCMFvj>zW#wT393|StgrVBqyfiX6C6FC8-!1sRU#@ z>br)RW{0{K8R_SQx)l~#dglA*=!a!lMumImyO|f{8>a>8xdw;p1(Z8`8iXYI>zjcM z06Kvc*%{7v)q35(Gcmu|Jj3BGzcmr=0P1E#HbG0yv1R(kRW2S0V-K7B4 I#Q_2W0AJI3?*IS* diff --git a/test_fuzz_decnet.db-shm b/test_fuzz_decnet.db-shm index ab3ab375339dac434b9bb4f7741709cb488a86af..5bfb1ff4b7b61b9dc9a0746f6343b084dfb5deb2 100644 GIT binary patch delta 64 vcmZo@U}|V!njj(JYh+z06nAR-25Z%6Mv*3~ZnD;Gn^awW7mw7&ga!2gC2AV^ delta 64 wcmZo@U}|V!njj(Z`q{3&i4)AHZ#&SY@#GNW^I2*>$x3gTkKmEon6RK80P-^*vj6}9 diff --git a/test_fuzz_decnet.db-wal b/test_fuzz_decnet.db-wal index 7a014696c61d34e834b41424e00b15f307039d50..8e4ab42f72f3fe64e90f0ddb6b072d49ccfca30d 100644 GIT binary patch delta 362 zcmX@{knzMrMho+Lwk8JMM-mJS0t_Iq!CG~iQKZSLY#zJAXEpL?0fiWm#nYZggE6eBvGk5+lyGPF4q8xG7E^ZAe)l+^@DiK7B7d*4F=Qr z>kSi=3=>UL5_OXe%uIAm(ku;h6Voisbd!xzOp`5=Qc@GmjT2LHGxJo8l2iiVyj?e)DvftKA$RfFyc*_O$@M!+aZnqRaxKEXbz(XDEy6VVCCI++Z+` zzuwr$)HEs8#8THHCB;nF#K=5Z*TN#jQrFNZ*)%o9A~D4v(K0b5H#1MgC`rZ8NX6eH z%`nuc)V$Q$D=)y!J143vC@9iDI6N}P!oxq;UthmC$vn%YEW*d#J1^9&A}QFUAgL5= z0MHGr$nFTf`TW}t=1XT Generator[None, None, None]: - repo.db_path = "test_bounty_decnet.db" - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - repo.reinitialize() - yield - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - -@pytest.fixture -def auth_token(): - 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"] - -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 repository if we want, or mock a log line that triggers it. - # For now, let's 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) - -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 diff --git a/tests/test_fleet_api.py b/tests/test_fleet_api.py deleted file mode 100644 index cb65881..0000000 --- a/tests/test_fleet_api.py +++ /dev/null @@ -1,99 +0,0 @@ -import json -import pytest -from fastapi.testclient import TestClient -from decnet.web.api import app -import decnet.config -from pathlib import Path -from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD -from decnet.web.dependencies import repo - -@pytest.fixture(autouse=True) -def setup_db(): - repo.db_path = "test_fleet_decnet.db" - import os - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - repo.reinitialize() - yield - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - -TEST_STATE_FILE = Path("test-decnet-state.json") - -@pytest.fixture(autouse=True) -def patch_state_file(monkeypatch): - # Patch the global STATE_FILE variable in the config module - monkeypatch.setattr(decnet.config, "STATE_FILE", TEST_STATE_FILE) - -@pytest.fixture -def mock_state_file(): - # Create a dummy state file for testing - _test_state = { - "config": { - "mode": "unihost", - "interface": "eth0", - "subnet": "192.168.1.0/24", - "gateway": "192.168.1.1", - "deckies": [ - { - "name": "test-decky-1", - "ip": "192.168.1.10", - "services": ["ssh"], - "distro": "debian", - "base_image": "debian", - "hostname": "test-host-1", - "service_config": {"ssh": {"banner": "SSH-2.0-OpenSSH_8.9"}}, - "archetype": "deaddeck", - "nmap_os": "linux", - "build_base": "debian:bookworm-slim" - }, - { - "name": "test-decky-2", - "ip": "192.168.1.11", - "services": ["http"], - "distro": "ubuntu", - "base_image": "ubuntu", - "hostname": "test-host-2", - "service_config": {}, - "archetype": None, - "nmap_os": "linux", - "build_base": "debian:bookworm-slim" - } - ], - "log_target": None, - "log_file": "test.log", - "ipvlan": False - }, - "compose_path": "test-compose.yml" - } - TEST_STATE_FILE.write_text(json.dumps(_test_state)) - - yield _test_state - - # Cleanup - if TEST_STATE_FILE.exists(): - TEST_STATE_FILE.unlink() - -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" - -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 diff --git a/tests/test_web_api.py b/tests/test_web_api.py deleted file mode 100644 index 0e1651f..0000000 --- a/tests/test_web_api.py +++ /dev/null @@ -1,135 +0,0 @@ -import os -from typing import Generator - -import pytest -from fastapi.testclient import TestClient - -from decnet.web.api import app -from decnet.web.dependencies import repo -from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD - - -@pytest.fixture(autouse=True) -def setup_db() -> Generator[None, None, None]: - repo.db_path = "test_decnet.db" - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - - repo.reinitialize() - - # Yield control to the test function - yield - - # Teardown - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - - -def test_login_success() -> None: - with TestClient(app) as client: - # The TestClient context manager triggers startup/shutdown events - 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 - - -def test_login_failure() -> None: - with TestClient(app) as client: - response = client.post( - "/api/v1/auth/login", - json={"username": "admin", "password": "wrongpassword"} - ) - assert response.status_code == 401 - - response = client.post( - "/api/v1/auth/login", - json={"username": "nonexistent", "password": "wrongpassword"} - ) - assert response.status_code == 401 - - -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"] - - # 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 - - # 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 - - # 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 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 - - -def test_get_logs_unauthorized() -> None: - with TestClient(app) as client: - response = 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) - -def test_get_stats_unauthorized() -> None: - with TestClient(app) as client: - response = 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 diff --git a/tests/test_web_api_fuzz.py b/tests/test_web_api_fuzz.py deleted file mode 100644 index f14caeb..0000000 --- a/tests/test_web_api_fuzz.py +++ /dev/null @@ -1,110 +0,0 @@ -import os -import pytest -import json -from typing import Generator, Any, Optional -from fastapi.testclient import TestClient -from hypothesis import given, strategies as st, settings, HealthCheck -import httpx - -from decnet.web.api import app -from decnet.web.dependencies import repo -from decnet.env import DECNET_ADMIN_USER, DECNET_ADMIN_PASSWORD - -# Re-use setup from test_web_api -@pytest.fixture(scope="function", autouse=True) -def setup_db() -> Generator[None, None, None]: - repo.db_path = "test_fuzz_decnet.db" - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - - repo.reinitialize() - yield - if os.path.exists(repo.db_path): - os.remove(repo.db_path) - -# bcrypt is intentionally slow, so we disable/extend the deadline -_FUZZ_SETTINGS: dict[str, Any] = { - "max_examples": 50, - "deadline": None, # bcrypt hashing takes >200ms - "suppress_health_check": [HealthCheck.function_scoped_fixture] -} - -@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: - """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) - # 200, 401, or 422 are acceptable. 500 is a failure. - assert _response.status_code in (200, 401, 422) - except (UnicodeEncodeError, json.JSONDecodeError): - pass - -@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: - """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 - -@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: - """Fuzz the logs pagination and search.""" - 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"] - - _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) - -@settings(**_FUZZ_SETTINGS) -@given( - token=st.text(min_size=0, max_size=4096) -) -def test_fuzz_auth_header(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