diff --git a/backend/tests/integration/api/v1/api_keys/__init__.py b/backend/tests/integration/api/v1/api_keys/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/integration/api/v1/api_keys/test_read.py b/backend/tests/integration/api/v1/api_keys/test_read.py new file mode 100644 index 00000000..210a2600 --- /dev/null +++ b/backend/tests/integration/api/v1/api_keys/test_read.py @@ -0,0 +1,40 @@ +"""Read routes for API keys rely on the global domain-error handlers. + +These pin the generic 404 and 403 bodies (no raw exception text, no echo of the +identifier) that the route modules previously produced themselves. +""" + +import pytest +from httpx import AsyncClient + +from src.infrastructure.auth.dependencies import get_current_user +from src.interfaces.main import app + +pytestmark = pytest.mark.asyncio + + +async def test_a_missing_api_key_returns_the_generic_not_found(auth_client: AsyncClient): + response = await auth_client.get("/api/v1/api-keys/999999") + + assert response.status_code == 404 + body = response.json() + assert body["detail"] == "The requested resource was not found." + assert body["support_id"] + assert "999999" not in body["detail"] + + +async def test_another_users_api_key_is_forbidden_generically( + auth_client: AsyncClient, test_user: dict, test_user_2: dict +): + created = await auth_client.post("/api/v1/api-keys/", json={"name": "Cross User Key"}) + assert created.status_code == 201 + key_id = created.json()["id"] + + app.dependency_overrides[get_current_user] = lambda: test_user_2 + response = await auth_client.get(f"/api/v1/api-keys/{key_id}") + + assert response.status_code == 403 + body = response.json() + assert body["detail"] == "You don't have permission for this action." + assert body["support_id"] + assert "Cross User Key" not in response.text diff --git a/backend/tests/integration/api/v1/users/test_create.py b/backend/tests/integration/api/v1/users/test_create.py index 323f8104..fd9ce702 100644 --- a/backend/tests/integration/api/v1/users/test_create.py +++ b/backend/tests/integration/api/v1/users/test_create.py @@ -135,3 +135,12 @@ async def test_signup_accepts_a_non_latin_password(client: AsyncClient, db_sessi response = await client.post("/api/v1/users/", json={**generate_unique_user_data(), "password": "Пароль1!"}) assert response.status_code == 201 + + +async def test_signup_names_every_missing_class_at_once(client: AsyncClient, db_session: AsyncSession): + """A password missing several classes reports all of them, not just the first.""" + response = await client.post("/api/v1/users/", json={**generate_unique_user_data(), "password": "password"}) + + assert response.status_code == 422 + requirements = {error["ctx"]["requirement"] for error in response.json()["detail"]} + assert {"uppercase", "digit", "special"} <= requirements diff --git a/backend/tests/integration/auth/test_endpoints.py b/backend/tests/integration/auth/test_endpoints.py index f3fc2b02..ad843d00 100644 --- a/backend/tests/integration/auth/test_endpoints.py +++ b/backend/tests/integration/auth/test_endpoints.py @@ -13,6 +13,7 @@ from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession +from src.infrastructure.auth import routes from src.infrastructure.auth.dependencies import get_optional_principal from src.infrastructure.auth.setup import auth as crud_auth from src.interfaces.main import app @@ -350,3 +351,24 @@ def recording_hashpw(password, salt): assert response.status_code == 201 assert threads assert threading.main_thread().name not in threads + + +@pytest.mark.asyncio +async def test_login_finishes_through_the_session_transport(client: AsyncClient, test_user: dict): + """The route must create the session through SessionTransport.complete_login (cookies + hooks).""" + captured: dict = {} + original = routes.session_transport.complete_login + + async def spy(request, response, user, options): + captured["options"] = options + return await original(request, response, user, options) + + with patch.object(routes.session_transport, "complete_login", spy): + response = await client.post( + "/api/v1/auth/login", + data={**_credentials(test_user), "remember_me": "true"}, + ) + + assert response.status_code == 200 + assert captured["options"]["remember_me"] is True + assert captured["options"]["metadata"]["login_type"] == "password" diff --git a/backend/tests/integration/auth/test_oauth.py b/backend/tests/integration/auth/test_oauth.py index 95e7fe8b..0b5478cc 100644 --- a/backend/tests/integration/auth/test_oauth.py +++ b/backend/tests/integration/auth/test_oauth.py @@ -150,6 +150,22 @@ async def test_an_offsite_redirect_target_falls_back_to_the_app(client: AsyncCli assert response.headers["location"] == BASE +async def test_an_absolute_same_origin_redirect_is_refused(client: AsyncClient, monkeypatch): + """Only relative paths are safe to hand back, even when the host is the app's own.""" + _stub_google( + monkeypatch, + {"sub": "google-321", "email": "absolute@example.com", "email_verified": True, "name": "Absolute Target"}, + ) + state = await _start(client, redirect_to=f"{BASE}/dashboard") + + response = await client.get( + "/api/v1/auth/oauth/callback/google", params={"code": "the-code", "state": state}, follow_redirects=False + ) + + assert response.status_code == 307 + assert response.headers["location"] == BASE + + async def test_a_state_this_browser_never_started_is_refused(client: AsyncClient): """A state without its browser-bound cookie may be a login-CSRF attempt, so no session.""" response = await client.get( @@ -170,3 +186,19 @@ async def test_a_provider_error_sends_the_browser_back_with_an_error(client: Asy location = urlparse(response.headers["location"]) assert f"{location.scheme}://{location.netloc}" == BASE assert parse_qs(location.query)["error"] + + +async def test_the_auth_paths_keep_their_existing_contract(): + """The migration to crudauth must not move the URLs clients already call.""" + paths = {route.path for route in app.routes} + + for path in ( + "/api/v1/auth/login", + "/api/v1/auth/logout", + "/api/v1/auth/logout-all", + "/api/v1/auth/refresh-csrf", + "/api/v1/auth/check-auth", + "/api/v1/auth/oauth/{provider}", + "/api/v1/auth/oauth/callback/{provider}", + ): + assert path in paths diff --git a/backend/tests/unit/infrastructure/auth/test_setup.py b/backend/tests/unit/infrastructure/auth/test_setup.py index 48ed3d77..75cae3b6 100644 --- a/backend/tests/unit/infrastructure/auth/test_setup.py +++ b/backend/tests/unit/infrastructure/auth/test_setup.py @@ -1,11 +1,16 @@ """Tests for the crudauth composition root wiring.""" +from types import SimpleNamespace + import pytest -from crudauth import Principal +from crudauth import NewUserContext, Principal from starlette.requests import Request from src.infrastructure.auth import setup from src.infrastructure.config.settings import settings +from src.infrastructure.database.session import async_session +from src.modules.rate_limit.crud import crud_rate_limits +from src.modules.user.constants import NAME_MAX_LENGTH def _request(path: str, client_host: str = "203.0.113.7") -> Request: @@ -84,9 +89,177 @@ def test_different_paths_get_different_budgets(self): _request("/api/v1/rate-limits/"), principal ) + def test_ipv6_callers_are_keyed_by_their_network(self, monkeypatch): + """Rotating addresses inside one /64 must not mint fresh budgets.""" + monkeypatch.setattr(settings, "TRUSTED_PROXY_HOPS", 0) + + key = setup.api_rate_limit_key(_request("/api/v1/tiers/", "2001:db8:1:2:3:4:5:6"), None) + + assert key == "ip:2001:db8:1:2::/64:/api/v1/tiers/" + class TestOAuthWiring: """The callback URI crudauth sends to the provider matches the route that serves it.""" def test_the_callback_lives_under_the_api_prefix(self): assert setup.OAUTH_PREFIX == "/api/v1/auth/oauth" + + +class TestRateLimiterBackendIndependence: + """RATE_LIMITER_BACKEND and SESSION_BACKEND are chosen independently.""" + + def test_rate_limiter_does_not_follow_the_session_backend(self, monkeypatch): + monkeypatch.setattr(settings, "SESSION_BACKEND", "memory") + monkeypatch.setattr(settings, "RATE_LIMITER_BACKEND", "redis") + + assert setup._rate_limiter() is not None + + def test_sessions_do_not_follow_the_rate_limiter_backend(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_BACKEND", "redis") + monkeypatch.setattr(settings, "SESSION_BACKEND", "memory") + + assert setup._session_transport().redis_url is None + + +_SENTINEL_DB = object() + + +class TestResolveApiRateLimit: + """The tier row is read through the request's own database dependency.""" + + async def test_the_tier_row_comes_from_the_session_override(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_ENABLED", True) + entered: list[bool] = [] + + async def override_session(): + entered.append(True) + yield _SENTINEL_DB + + request = SimpleNamespace( + url=SimpleNamespace(path="/api/v1/tiers/"), + app=SimpleNamespace(dependency_overrides={async_session: override_session}), + ) + principal = Principal(user_id=1, user=SimpleNamespace(tier_id=7), transport="session") + seen: dict[str, object] = {} + + async def fake_get(db, **kwargs): + seen["db"] = db + return {"limit": 2, "period": 3600} + + monkeypatch.setattr(crud_rate_limits, "get", fake_get) + + result = await setup.resolve_api_rate_limit(request, principal) + + assert entered == [True] + assert seen["db"] is _SENTINEL_DB + assert (result.times, result.seconds) == (2, 3600) + + async def test_a_caller_without_a_tier_gets_the_default_limit(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_ENABLED", True) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_LIMIT", 11) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_PERIOD", 99) + + result = await setup.resolve_api_rate_limit(SimpleNamespace(url=None, app=None), None) + + assert (result.times, result.seconds) == (11, 99) + + async def test_a_disabled_limiter_returns_no_limit(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_ENABLED", False) + + assert await setup.resolve_api_rate_limit(SimpleNamespace(url=None, app=None), None) is None + + async def test_a_tier_without_a_row_for_the_path_gets_the_default_limit(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_ENABLED", True) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_LIMIT", 11) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_PERIOD", 99) + + async def override_session(): + yield _SENTINEL_DB + + request = SimpleNamespace( + url=SimpleNamespace(path="/api/v1/tiers/"), + app=SimpleNamespace(dependency_overrides={async_session: override_session}), + ) + principal = Principal(user_id=1, user=SimpleNamespace(tier_id=7), transport="session") + + async def fake_get(db, **kwargs): + return None + + monkeypatch.setattr(crud_rate_limits, "get", fake_get) + + result = await setup.resolve_api_rate_limit(request, principal) + + assert (result.times, result.seconds) == (11, 99) + + async def test_a_principal_without_a_loaded_user_gets_the_default_limit(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_ENABLED", True) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_LIMIT", 11) + monkeypatch.setattr(settings, "DEFAULT_RATE_LIMIT_PERIOD", 99) + + principal = Principal(user_id=1, user=None, transport="session") + + result = await setup.resolve_api_rate_limit(SimpleNamespace(url=None, app=None), principal) + + assert (result.times, result.seconds) == (11, 99) + + +class TestOAuthProviderSelection: + """Only a fully configured Google is wired; the boilerplate has no other provider route.""" + + def test_google_is_wired_when_both_credentials_are_set(self, monkeypatch): + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_ID", "client-id") + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_SECRET", "client-secret") + + providers = setup._oauth_providers() + + assert set(providers) == {"google"} + assert providers["google"].client_id == "client-id" + + def test_google_is_dropped_when_a_credential_is_missing(self, monkeypatch): + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_ID", "client-id") + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_SECRET", "") + + assert setup._oauth_providers() == {} + + def test_github_credentials_do_not_add_an_unrouted_provider(self, monkeypatch): + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_ID", "") + monkeypatch.setattr(settings, "OAUTH_GOOGLE_CLIENT_SECRET", "") + monkeypatch.setattr(settings, "OAUTH_GITHUB_CLIENT_ID", "gh-id") + monkeypatch.setattr(settings, "OAUTH_GITHUB_CLIENT_SECRET", "gh-secret") + + assert setup._oauth_providers() == {} + + +class TestNewUserFields: + """crudauth creates the account; the boilerplate supplies the required display name.""" + + def test_the_display_name_is_filled_and_bounded(self): + context = NewUserContext( + email="a" * 40 + "@example.com", + username="auser", + source="register", + db=None, # type: ignore[arg-type] + ) + + fields = setup._new_user_fields(context) + + assert fields["name"] == "a" * NAME_MAX_LENGTH + assert len(fields["name"]) == NAME_MAX_LENGTH + + +class TestSessionTransportWiring: + """The session settings reach the transport instead of the library defaults.""" + + def test_the_session_settings_reach_the_transport(self, monkeypatch): + monkeypatch.setattr(settings, "SESSION_BACKEND", "memory") + monkeypatch.setattr(settings, "CSRF_ENABLED", False) + monkeypatch.setattr(settings, "MAX_SESSIONS_PER_USER", 2) + monkeypatch.setattr(settings, "SESSION_TIMEOUT_MINUTES", 7) + monkeypatch.setattr(settings, "SESSION_CLEANUP_INTERVAL_MINUTES", 3) + + transport = setup._session_transport() + + assert transport.csrf_enabled is False + assert transport.max_sessions_per_user == 2 + assert transport.session_timeout_minutes == 7 + assert transport.cleanup_interval_minutes == 3 diff --git a/backend/tests/unit/infrastructure/config/test_settings.py b/backend/tests/unit/infrastructure/config/test_settings.py index df8d62e2..9165ff3a 100644 --- a/backend/tests/unit/infrastructure/config/test_settings.py +++ b/backend/tests/unit/infrastructure/config/test_settings.py @@ -5,6 +5,7 @@ import pytest +from src.infrastructure.config.enums import RateLimiterBackend, SessionBackend from src.infrastructure.config.settings import Settings, get_settings @@ -267,3 +268,43 @@ def test_taskiq_required_settings_exist(self): for attr in required_attrs: assert hasattr(settings, attr), f"Missing required Taskiq setting: {attr}" + + +class TestRateLimiterSettings: + """The limiter moved to crudauth; the memcached and fail-open knobs went with it.""" + + def test_the_removed_memcached_and_fail_open_settings_are_gone(self): + settings = get_settings() + + for name in ( + "RATE_LIMITER_FAIL_OPEN", + "RATE_LIMITER_MEMCACHED_HOST", + "RATE_LIMITER_MEMCACHED_PORT", + "RATE_LIMITER_MEMCACHED_POOL_SIZE", + "RATE_LIMITER_MEMCACHED_CONNECT_TIMEOUT", + ): + assert not hasattr(settings, name), name + + def test_the_rate_limiter_backend_names_a_supported_backend(self): + assert RateLimiterBackend(get_settings().RATE_LIMITER_BACKEND) + + def test_the_backend_enums_accept_only_redis_and_memory(self): + with pytest.raises(ValueError): + SessionBackend("memcached") + with pytest.raises(ValueError): + RateLimiterBackend("memcached") + + +class TestCORSSettings: + """Credentialed requests reject a wildcard origin, so the default must not be one.""" + + def test_the_default_origins_are_explicit_not_a_wildcard(self): + origins = Settings().CORS_ORIGINS_LIST + + assert origins + assert "*" not in origins + assert all(origin.startswith("http") for origin in origins) + + @patch.dict(os.environ, {"CORS_ORIGINS": "http://a.test, http://b.test ,"}) + def test_the_origin_list_strips_whitespace_and_drops_empties(self): + assert Settings().CORS_ORIGINS_LIST == ["http://a.test", "http://b.test"] diff --git a/backend/tests/unit/infrastructure/database/test_metadata.py b/backend/tests/unit/infrastructure/database/test_metadata.py new file mode 100644 index 00000000..5f99a877 --- /dev/null +++ b/backend/tests/unit/infrastructure/database/test_metadata.py @@ -0,0 +1,9 @@ +"""The shared metadata uses the standard constraint naming convention.""" + +from src.infrastructure.database.session import NAMING_CONVENTION, Base + + +def test_metadata_uses_the_standard_naming_convention(): + """Stable names keep Alembic autogenerate from renaming constraints on every run.""" + assert dict(Base.metadata.naming_convention) == NAMING_CONVENTION + assert set(NAMING_CONVENTION) == {"ix", "uq", "ck", "fk", "pk"} diff --git a/backend/tests/unit/infrastructure/test_app_factory.py b/backend/tests/unit/infrastructure/test_app_factory.py index 04f1efc9..64e88a8a 100644 --- a/backend/tests/unit/infrastructure/test_app_factory.py +++ b/backend/tests/unit/infrastructure/test_app_factory.py @@ -145,9 +145,13 @@ def _create_app(environment: EnvironmentOption, enable_docs_in_production: bool ) -async def _docs_statuses(app: FastAPI) -> list[int]: +async def _statuses(app: FastAPI, paths: tuple[str, ...]) -> list[int]: async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: - return [(await client.get(path)).status_code for path in DOCS_PATHS] + return [(await client.get(path)).status_code for path in paths] + + +async def _docs_statuses(app: FastAPI) -> list[int]: + return await _statuses(app, DOCS_PATHS) @pytest.mark.asyncio @@ -177,3 +181,34 @@ async def test_gated_docs_are_served_to_superusers(environment, enable_docs_in_p app.dependency_overrides[get_current_superuser] = lambda: {"id": 1, "is_superuser": True} assert await _docs_statuses(app) == [200] * len(DOCS_PATHS) + + +@pytest.mark.asyncio +async def test_gated_docs_use_the_configured_paths(): + """The protected docs router must serve at the configured URLs, not hardcoded ones.""" + custom_paths = ("/internal/docs", "/internal/redoc", "/internal/openapi.json") + custom = Settings( + ENVIRONMENT=EnvironmentOption.STAGING, + DOCS_URL=custom_paths[0], + REDOC_URL=custom_paths[1], + OPENAPI_URL=custom_paths[2], + ) + app = app_factory.create_application(router=APIRouter(), settings=custom) + + assert await _statuses(app, custom_paths) == [401, 401, 401] + assert await _docs_statuses(app) == [404, 404, 404] + + app.dependency_overrides[get_current_superuser] = lambda: {"id": 1, "is_superuser": True} + + assert await _statuses(app, custom_paths) == [200, 200, 200] + + +class TestLifespanAuth: + """crudauth is initialized on startup, after every connection is ready.""" + + async def test_initializes_crudauth_on_startup(self, lifespan_settings, patched_lifespan): + mocks, _ = patched_lifespan + lifespan = app_factory.lifespan_factory(lifespan_settings) + + async with lifespan(FastAPI()): + mocks["auth"].initialize.assert_awaited_once() diff --git a/backend/tests/unit/infrastructure/test_redis.py b/backend/tests/unit/infrastructure/test_redis.py new file mode 100644 index 00000000..c193fb9e --- /dev/null +++ b/backend/tests/unit/infrastructure/test_redis.py @@ -0,0 +1,63 @@ +"""The shared Redis clients, and where they are injected.""" + +from src.infrastructure import redis +from src.infrastructure.auth import setup +from src.infrastructure.cache import initialize +from src.infrastructure.config.settings import settings + + +def _connection(client): + return client.connection_pool + + +class TestClientSettings: + """Each client follows its own settings, not the other service's.""" + + def test_cache_client_uses_the_cache_settings(self): + pool = _connection(redis.cache_redis_client) + kwargs = pool.connection_kwargs + + assert kwargs["host"] == settings.CACHE_REDIS_HOST + assert kwargs["port"] == settings.CACHE_REDIS_PORT + assert kwargs["db"] == settings.CACHE_REDIS_DB + assert kwargs["socket_timeout"] == settings.CACHE_REDIS_CONNECT_TIMEOUT + assert kwargs["decode_responses"] is False + assert pool.max_connections == settings.CACHE_REDIS_POOL_SIZE + + def test_rate_limiter_client_uses_the_rate_limiter_settings(self): + pool = _connection(redis.rate_limiter_redis_client) + kwargs = pool.connection_kwargs + + assert kwargs["host"] == settings.RATE_LIMITER_REDIS_HOST + assert kwargs["port"] == settings.RATE_LIMITER_REDIS_PORT + assert kwargs["db"] == settings.RATE_LIMITER_REDIS_DB + assert kwargs["socket_timeout"] == settings.RATE_LIMITER_REDIS_CONNECT_TIMEOUT + assert kwargs["decode_responses"] is False + assert pool.max_connections == settings.RATE_LIMITER_REDIS_POOL_SIZE + + +class TestInjection: + """The shared clients are the ones crudauth and the cache actually use.""" + + def test_the_rate_limiter_reuses_the_shared_client(self, monkeypatch): + monkeypatch.setattr(settings, "RATE_LIMITER_BACKEND", "redis") + + backend = setup._rate_limiter() + + assert backend.client is redis.rate_limiter_redis_client + assert backend._owns_client is False + + async def test_the_cache_backend_reuses_the_shared_client(self, monkeypatch): + monkeypatch.setattr(settings, "CACHE_BACKEND", "redis") + captured: dict[str, object] = {} + + class RecordingBackend: + def __init__(self, settings, client): + captured["client"] = client + + monkeypatch.setattr(initialize, "RedisBackend", RecordingBackend) + monkeypatch.setattr(initialize.cache_provider, "register_backend", lambda *args, **kwargs: None) + + await initialize.initialize_cache() + + assert captured["client"] is redis.cache_redis_client diff --git a/backend/tests/unit/interfaces/admin/test_users_view.py b/backend/tests/unit/interfaces/admin/test_users_view.py new file mode 100644 index 00000000..c6b00f4c --- /dev/null +++ b/backend/tests/unit/interfaces/admin/test_users_view.py @@ -0,0 +1,42 @@ +"""Tests for the User admin view's password handling.""" + +import threading +from unittest.mock import patch + +import bcrypt +import pytest +from crudauth.exceptions import PasswordPolicyException + +from src.interfaces.admin.views.users import UserAdmin + + +async def test_the_admin_form_hashes_the_password_off_the_event_loop(): + """bcrypt is deliberately slow; on the loop thread it would stall every other request.""" + real_hashpw = bcrypt.hashpw + threads: list[str] = [] + + def recording_hashpw(password, salt): + threads.append(threading.current_thread().name) + return real_hashpw(password, salt) + + data = {"hashed_password": "Str1ngst!"} + with patch.object(bcrypt, "hashpw", recording_hashpw): + await UserAdmin().on_model_change(data, model=None, is_created=True, request=None) + + assert data["hashed_password"] != "Str1ngst!" + assert data["hashed_password"] + assert threads + assert threading.main_thread().name not in threads + + +async def test_the_admin_form_refuses_a_password_that_breaks_the_policy(): + with pytest.raises(PasswordPolicyException): + await UserAdmin().on_model_change({"hashed_password": "weak"}, model=None, is_created=True, request=None) + + +async def test_the_admin_form_turns_a_blank_oauth_provider_into_none(): + data = {"oauth_provider": ""} + + await UserAdmin().on_model_change(data, model=None, is_created=False, request=None) + + assert data["oauth_provider"] is None diff --git a/backend/tests/unit/modules/api_keys/test_indexes.py b/backend/tests/unit/modules/api_keys/test_indexes.py new file mode 100644 index 00000000..f934d36a --- /dev/null +++ b/backend/tests/unit/modules/api_keys/test_indexes.py @@ -0,0 +1,27 @@ +"""The api_keys tables must not declare the same index twice.""" + +import src.modules.api_keys.models # noqa: F401 (registers the tables) +from src.infrastructure.database.session import Base + +API_KEYS_TABLES = ("api_keys", "key_usage", "key_permissions") + + +def _is_implicit(index) -> bool: + """True for an index SQLAlchemy creates from a column's ``index=True``.""" + return bool(getattr(index, "_column_flag", False)) or index.name.startswith("ix_") + + +def test_no_column_index_is_duplicated_by_an_explicit_index(): + """A column cannot carry both ``index=True`` and its own explicit single-column Index.""" + duplicates: list[tuple[str, str, str]] = [] + for table_name in API_KEYS_TABLES: + table = Base.metadata.tables[table_name] + flagged = {column.name for column in table.columns if column.index} + for index in table.indexes: + if _is_implicit(index): + continue + columns = [expression.name for expression in index.expressions if hasattr(expression, "name")] + if len(columns) == 1 and columns[0] in flagged: + duplicates.append((table_name, index.name, columns[0])) + + assert duplicates == [] diff --git a/backend/tests/unit/modules/user/test_schemas.py b/backend/tests/unit/modules/user/test_schemas.py index 42736d5f..a0e8d73e 100644 --- a/backend/tests/unit/modules/user/test_schemas.py +++ b/backend/tests/unit/modules/user/test_schemas.py @@ -1,6 +1,9 @@ """Unit tests for the User schemas.""" +from src.infrastructure.auth.password_policy import password_policy +from src.infrastructure.auth.setup import auth from src.infrastructure.config.settings import settings +from src.modules.user import schemas as user_schemas from src.modules.user.schemas import UserCreate @@ -17,3 +20,24 @@ def test_the_schema_leaves_enforcement_to_the_policy(): user = UserCreate(name="Test User", username="testuser", email="user.userson@example.com", password="weak") assert user.password == "weak" + + +def test_the_request_schema_and_crudauth_share_one_password_policy(): + """A second policy instance could document or accept rules crudauth doesn't enforce.""" + assert user_schemas.password_policy is password_policy + assert auth.password_policy is password_policy + + +def test_the_password_policy_is_built_from_the_password_settings(): + assert password_policy.min_length == settings.PASSWORD_MIN_LENGTH + assert password_policy.require_uppercase == settings.PASSWORD_REQUIRE_UPPERCASE + assert password_policy.require_lowercase == settings.PASSWORD_REQUIRE_LOWERCASE + assert password_policy.require_digit == settings.PASSWORD_REQUIRE_DIGIT + assert password_policy.require_special == settings.PASSWORD_REQUIRE_SPECIAL + + +def test_the_password_field_is_driven_by_the_policy(): + field = UserCreate.model_json_schema()["properties"]["password"] + + assert field["minLength"] == password_policy.min_length + assert field["description"] == password_policy.description