fix(tests): repoint security tests + add CI

.github/workflows/ci.yml

@@ -0,0 +1,33 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - run: pip install -e . -r requirements.txt
+
+      - run: pytest -q
+
+      - run: ruff check .
+
+      - name: mypy (advisory, not gating)
+        run: mypy velocix --ignore-missing-imports || true

examples/openapi_example.py

@@ -9,9 +9,9 @@
 Following FastAPI's approach for automatic OpenAPI generation.
 """
 from velocix import Velocix
-from velocix.validation import Struct, field
 from velocix.core.depends import Depends
 from velocix.openapi import enable_auto_docs
+from velocix.validation import Struct
 
 
 # Data models using msgspec Struct (Velocix's validation system)

pyproject.toml

@@ -67,7 +67,10 @@ target-version = "py311"
 
 [tool.ruff.lint]
 select = ["E", "F", "I", "UP", "B"]
-ignore = ["E501"]
+# E501: line length handled by formatting, not worth failing CI over.
+# B008: Depends()/Query()/Header() etc. as default-arg values is velocix's
+# own FastAPI-style DI pattern, used throughout its public API — not a bug.
+ignore = ["E501", "B008"]
 
 [tool.mypy]
 python_version = "3.11"

requirements.txt

@@ -1,19 +1,16 @@
 granian>=2.5.6
 orjson>=3.11.3
-httptools>=0.6.4
-fast-query-parsers>=2.0.0
+fast-query-parsers>=1.0.0
 python-multipart>=0.0.9
 msgspec>=0.19.0
 argon2-cffi>=25.1.0
 pyjwt>=2.10.1
 cryptography>=42.0.0
-websockets>=15.0.1
 uvloop>=0.21.0; sys_platform != 'win32'
 zstandard>=0.25.0
 brotli>=1.1.0
 click>=8.1.8
 xxhash>=3.6.0
-regex>=2024.11.6
 itsdangerous>=2.0.0
 nh3>=0.2.0
 pytest>=7.4.0

setup.py

@@ -1,9 +1,9 @@
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
-with open("README.md", "r", encoding="utf-8") as f:
+with open("README.md", encoding="utf-8") as f:
     long_description = f.read()
 
-with open("requirements.txt", "r", encoding="utf-8") as f:
+with open("requirements.txt", encoding="utf-8") as f:
     requirements = [line.strip() for line in f if line.strip() and not line.startswith("#")]
 
 setup(

tests/test_edge_cases_dependency.py

@@ -1,7 +1,7 @@
 import asyncio
 from typing import Annotated
 
-from velocix import Cookie, Header, Query, TestClient, Velocix
+from velocix import Query, TestClient, Velocix
 from velocix.core.depends import Depends
 from velocix.core.exceptions import HTTPException
 

tests/test_edge_cases_error_middleware.py

@@ -1,10 +1,8 @@
 import asyncio
 from functools import partial
 
-import msgspec
-
 from velocix import CORSMiddleware, TestClient, Velocix
-from velocix.core.exceptions import HTTPException, NotFound
+from velocix.core.exceptions import HTTPException
 from velocix.core.middleware import BaseHTTPMiddleware
 from velocix.core.response import JSONResponse, Response
 

tests/test_security_base.py

@@ -5,7 +5,6 @@
 """
 
 import asyncio
-import time
 
 from velocix.security.base import (
     HookManager,
@@ -207,7 +206,7 @@ async def scenario():
         async_backend = MemoryBackend()
         sync_backend = MemoryBackend()
 
-        for i in range(5):
+        for _i in range(5):
             async_result = await async_backend.incr("key1", window=60.0)
             sync_result = sync_backend.incr_sync("key1", window=60.0)
             assert async_result == sync_result

tests/test_security_brute_force.py

@@ -1,11 +1,10 @@
-"""Tests for brute force protection middleware and standalone utility.
+"""Tests for brute force protection middleware.
 
 Covers: record_failure, mark_success, is_locked, get_retry_after,
-middleware lockout, create() factory, separate key tracking.
+middleware lockout, separate key tracking.
 """
 
 import asyncio
-import time
 from functools import partial
 
 from velocix import TestClient, Velocix
@@ -17,35 +16,32 @@ def _run(coro):
     return asyncio.run(coro)
 
 
-# ---------------------------------------------------------------------------
-# BruteForceProtection standalone — create() factory
-# ---------------------------------------------------------------------------
+async def _passthrough_app(request):
+    from velocix.core.response import Response
 
+    return Response(b"ok", status_code=200)
+
+
+def _make_bf(**kwargs):
+    """Build a BruteForceProtection instance for direct method testing,
+    without going through a Velocix app or the request-handling path."""
+    return BruteForceProtection(_passthrough_app, **kwargs)
 
-def test_create_factory():
-    bf = BruteForceProtection.create(
-        max_attempts=3,
-        window_seconds=60,
-        lockout_seconds=120,
-    )
-    assert bf._max_attempts == 3
-    assert bf._window_seconds == 60
-    assert bf._lockout_seconds == 120
+
+# ---------------------------------------------------------------------------
+# BruteForceProtection — record_failure / is_locked / mark_success
+# ---------------------------------------------------------------------------
 
 
 def test_record_failure_increments():
-    bf = BruteForceProtection.create(max_attempts=5, window_seconds=60)
+    bf = _make_bf(max_attempts=5, window_seconds=60)
     assert bf.record_failure("user:1.2.3.4") == 1
     assert bf.record_failure("user:1.2.3.4") == 2
     assert bf.record_failure("user:1.2.3.4") == 3
 
 
 def test_is_locked_after_threshold():
-    bf = BruteForceProtection.create(
-        max_attempts=3,
-        window_seconds=60,
-        lockout_seconds=120,
-    )
+    bf = _make_bf(max_attempts=3, window_seconds=60, lockout_seconds=120)
     bf.record_failure("user:1")
     bf.record_failure("user:1")
     assert bf.is_locked("user:1") is False  # 2 < 3
@@ -54,16 +50,12 @@ def test_is_locked_after_threshold():
 
 
 def test_is_locked_returns_false_for_unknown_key():
-    bf = BruteForceProtection.create(max_attempts=3, window_seconds=60)
+    bf = _make_bf(max_attempts=3, window_seconds=60)
     assert bf.is_locked("unknown") is False
 
 
 def test_mark_success_resets():
-    bf = BruteForceProtection.create(
-        max_attempts=3,
-        window_seconds=60,
-        lockout_seconds=120,
-    )
+    bf = _make_bf(max_attempts=3, window_seconds=60, lockout_seconds=120)
     bf.record_failure("user:1")
     bf.record_failure("user:1")
     bf.record_failure("user:1")
@@ -73,11 +65,7 @@ def test_mark_success_resets():
 
 
 def test_mark_success_clears_counter():
-    bf = BruteForceProtection.create(
-        max_attempts=3,
-        window_seconds=60,
-        lockout_seconds=120,
-    )
+    bf = _make_bf(max_attempts=3, window_seconds=60, lockout_seconds=120)
     bf.record_failure("user:1")
     bf.record_failure("user:1")
     bf.mark_success("user:1")
@@ -90,42 +78,29 @@ def test_mark_success_clears_counter():
 
 
 def test_get_retry_after():
-    bf = BruteForceProtection.create(
-        max_attempts=2,
-        window_seconds=60,
-        lockout_seconds=300,
-    )
+    bf = _make_bf(max_attempts=2, window_seconds=60, lockout_seconds=300)
     assert bf.get_retry_after("user:1") == 0  # not locked
     bf.record_failure("user:1")
     bf.record_failure("user:1")
     assert bf.get_retry_after("user:1") > 0
 
 
 def test_separate_keys_independent():
-    bf = BruteForceProtection.create(
-        max_attempts=2,
-        window_seconds=60,
-        lockout_seconds=120,
-    )
+    bf = _make_bf(max_attempts=2, window_seconds=60, lockout_seconds=120)
     bf.record_failure("user:A")
     bf.record_failure("user:A")
     assert bf.is_locked("user:A") is True
     assert bf.is_locked("user:B") is False
 
 
 # ---------------------------------------------------------------------------
-# BruteForceProtection standalone — custom backend
+# BruteForceProtection — custom backend
 # ---------------------------------------------------------------------------
 
 
-def test_create_with_custom_backend():
+def test_custom_backend():
     backend = MemoryBackend()
-    bf = BruteForceProtection.create(
-        max_attempts=2,
-        window_seconds=60,
-        lockout_seconds=60,
-        backend=backend,
-    )
+    bf = _make_bf(max_attempts=2, window_seconds=60, lockout_seconds=60, backend=backend)
     bf.record_failure("test")
     bf.record_failure("test")
     assert bf.is_locked("test") is True
@@ -165,16 +140,35 @@ async def scenario():
 
 
 def test_middleware_blocks_locked_ip():
-    app = _app_with_brute_force(max_attempts=2, window_seconds=60, lockout_seconds=120)
-
     async def scenario():
-        async with TestClient(app) as client:
-            # The test client has a fixed IP, so all requests share the same key
-            # We need to trigger lockout via the utility methods
-            # But middleware runs on every request... let's just test the utility
-            # and verify middleware doesn't interfere with clean requests
-            resp = await client.get("/ping")
-            assert resp.status_code == 200
+        from velocix.core.request import Request
+
+        middleware = BruteForceProtection(
+            _passthrough_app, max_attempts=2, window_seconds=60, lockout_seconds=120
+        )
+
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "path": "/ping",
+            "query_string": b"",
+            "headers": [],
+            "server": ("test", 80),
+            "client": ("testclient", 50000),
+        }
+        request = Request(scope, receive=None)
+
+        # Lock the key directly (this is the same IP TestClient/the request
+        # scope above resolves to via the default IP-based key_func)
+        middleware.record_failure("testclient")
+        middleware.record_failure("testclient")
+
+        resp = await middleware(request)
+        assert resp.status_code == 429
+        import orjson
+
+        body = orjson.loads(resp.body)
+        assert body["error"]["code"] == "BRUTE_FORCE_LOCKED"
 
     _run(scenario())