Initial commit: site monorepo with API, web, and infra.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,63 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.app_settings import (
|
||||
_coerce_value,
|
||||
apply_settings_to_app,
|
||||
bootstrap_settings,
|
||||
env_locks,
|
||||
get_settings_payload,
|
||||
write_settings,
|
||||
)
|
||||
|
||||
|
||||
def test_write_and_bootstrap_settings(tmp_path, monkeypatch):
|
||||
settings_file = tmp_path / "compton_settings.json"
|
||||
monkeypatch.setattr("app.core.app_settings.settings.compton_settings_path", str(settings_file))
|
||||
monkeypatch.delenv("ENABLE_DOCS", raising=False)
|
||||
|
||||
merged = write_settings({"enable_docs": False, "log_level": "INFO"})
|
||||
assert merged["enable_docs"] is False
|
||||
assert settings_file.exists()
|
||||
|
||||
bootstrap_settings()
|
||||
from app.core.config import settings
|
||||
|
||||
assert settings.enable_docs is False
|
||||
|
||||
|
||||
def test_env_locks_skip_locked_keys(tmp_path, monkeypatch):
|
||||
settings_file = tmp_path / "compton_settings.json"
|
||||
monkeypatch.setattr("app.core.app_settings.settings.compton_settings_path", str(settings_file))
|
||||
monkeypatch.setenv("ENABLE_DOCS", "true")
|
||||
monkeypatch.setattr("app.core.app_settings.settings.enable_docs", True)
|
||||
|
||||
write_settings({"enable_docs": False})
|
||||
payload = get_settings_payload()
|
||||
assert payload["locks"]["enable_docs"] is True
|
||||
assert payload["values"]["enable_docs"] is True
|
||||
|
||||
|
||||
def test_apply_settings_coerces_list_and_bool(monkeypatch):
|
||||
apply_settings_to_app({"enable_docs": "false", "cors_origins": "http://a.test,http://b.test"})
|
||||
from app.core.config import settings
|
||||
|
||||
assert settings.enable_docs is False
|
||||
assert settings.cors_origins == ["http://a.test", "http://b.test"]
|
||||
assert isinstance(env_locks(), dict)
|
||||
|
||||
|
||||
def test_read_settings_ignores_invalid_payload(tmp_path, monkeypatch):
|
||||
settings_file = tmp_path / "compton_settings.json"
|
||||
settings_file.write_text("[]", encoding="utf-8")
|
||||
monkeypatch.setattr("app.core.app_settings.settings.compton_settings_path", str(settings_file))
|
||||
from app.core.app_settings import read_settings
|
||||
|
||||
assert read_settings() == {}
|
||||
|
||||
|
||||
def test_coerce_value_rejects_invalid_list():
|
||||
with pytest.raises(ValueError, match="INVALID_LIST_cors_origins"):
|
||||
_coerce_value("cors_origins", 123)
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
from app.core.audit_log import read_audit_events, write_audit_event
|
||||
|
||||
|
||||
def test_audit_log_roundtrip(tmp_path, monkeypatch):
|
||||
audit_file = tmp_path / "admin-audit.jsonl"
|
||||
monkeypatch.setattr("app.core.audit_log.settings.admin_audit_log_path", str(audit_file))
|
||||
write_audit_event("admin.user.patch", "u1", "admin@example.com", {"target": "u2"})
|
||||
events = read_audit_events()
|
||||
assert len(events) == 1
|
||||
assert events[0]["action"] == "admin.user.patch"
|
||||
assert events[0]["details"]["target"] == "u2"
|
||||
|
||||
|
||||
def test_read_audit_events_ignores_invalid_json(tmp_path, monkeypatch):
|
||||
audit_file = tmp_path / "admin-audit.jsonl"
|
||||
audit_file.write_text('{"action":"ok"}\nnot-json\n', encoding="utf-8")
|
||||
monkeypatch.setattr("app.core.audit_log.settings.admin_audit_log_path", str(audit_file))
|
||||
events = read_audit_events()
|
||||
assert len(events) == 1
|
||||
@@ -0,0 +1,68 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi.security import HTTPAuthorizationCredentials
|
||||
|
||||
from app.core.dependencies import get_current_user, require_admin, require_superuser
|
||||
from app.core.security import create_access_token, hash_password
|
||||
from app.modules.users.repository import create_user, update_user
|
||||
|
||||
|
||||
def test_get_current_user_rejects_missing_credentials():
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
get_current_user(None)
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
|
||||
def test_get_current_user_rejects_pending_user():
|
||||
user = create_user("pending-dep@example.com", hash_password("Valid123"), status="pending")
|
||||
token = create_access_token(user.id, user.role, user.is_superuser)
|
||||
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
get_current_user(credentials)
|
||||
assert exc.value.status_code == 403
|
||||
assert exc.value.detail == "EMAIL_NOT_VERIFIED"
|
||||
|
||||
|
||||
def test_get_current_user_rejects_blocked_user():
|
||||
user = create_user("blocked-dep@example.com", hash_password("Valid123"), status="active")
|
||||
user.status = "blocked"
|
||||
update_user(user)
|
||||
token = create_access_token(user.id, user.role, user.is_superuser)
|
||||
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
get_current_user(credentials)
|
||||
assert exc.value.status_code == 403
|
||||
assert exc.value.detail == "ACCOUNT_BLOCKED"
|
||||
|
||||
|
||||
def test_get_current_user_rejects_invalid_token():
|
||||
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="not-a-jwt")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
get_current_user(credentials)
|
||||
assert exc.value.status_code == 401
|
||||
assert exc.value.detail == "INVALID_TOKEN"
|
||||
|
||||
|
||||
def test_get_current_user_rejects_unknown_user():
|
||||
token = create_access_token("missing-user-id", "user", False)
|
||||
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
get_current_user(credentials)
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
|
||||
def test_require_admin_rejects_regular_user():
|
||||
user = create_user("regular-dep@example.com", hash_password("Valid123"), role="user", status="active")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
require_admin(user)
|
||||
assert exc.value.status_code == 403
|
||||
|
||||
|
||||
def test_require_superuser_rejects_non_super_admin():
|
||||
user = create_user("ops-dep@example.com", hash_password("Valid123"), role="admin", status="active")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
require_superuser(user)
|
||||
assert exc.value.status_code == 403
|
||||
assert exc.value.detail == "SUPERUSER_ONLY"
|
||||
@@ -0,0 +1,29 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from app.core.email import SmtpMailer, get_mailer, memory_mailer, send_template_email
|
||||
|
||||
|
||||
def test_memory_mailer_latest_token():
|
||||
memory_mailer.clear()
|
||||
send_template_email(
|
||||
to="user@example.com",
|
||||
template="verify_email",
|
||||
subject="Verify",
|
||||
body="Open link\nTOKEN:abc123\n",
|
||||
)
|
||||
assert memory_mailer.latest_token("user@example.com", "verify_email") == "abc123"
|
||||
|
||||
|
||||
def test_smtp_mailer_sends_message(monkeypatch):
|
||||
monkeypatch.setattr("app.core.email.settings.email_delivery_mode", "smtp")
|
||||
monkeypatch.setattr("app.core.email.settings.smtp_from", "noreply@example.com")
|
||||
monkeypatch.setattr("app.core.email.settings.smtp_host", "localhost")
|
||||
monkeypatch.setattr("app.core.email.settings.smtp_port", 1025)
|
||||
monkeypatch.setattr("app.core.email.settings.smtp_user", "")
|
||||
monkeypatch.setattr("app.core.email.settings.smtp_password", "")
|
||||
|
||||
smtp_instance = MagicMock()
|
||||
with patch("app.core.email.smtplib.SMTP") as smtp_cls:
|
||||
smtp_cls.return_value.__enter__.return_value = smtp_instance
|
||||
get_mailer().send("user@example.com", "Subject", "Body", "verify_email")
|
||||
smtp_instance.send_message.assert_called_once()
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Install secrets must never be poisoned by sqlite test DATABASE_URL."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core import install_secrets as mod
|
||||
|
||||
|
||||
def test_adopt_from_environment_ignores_sqlite(monkeypatch):
|
||||
monkeypatch.setenv("DATABASE_URL", "sqlite+pysqlite:///:memory:")
|
||||
adopted = mod._adopt_from_environment()
|
||||
assert "DATABASE_URL" not in adopted
|
||||
|
||||
|
||||
def test_ensure_install_secrets_rejects_sqlite_install_env(tmp_path, monkeypatch):
|
||||
secrets_dir = tmp_path / "secrets"
|
||||
secrets_dir.mkdir()
|
||||
install_file = secrets_dir / "install.env"
|
||||
install_file.write_text(
|
||||
"DATABASE_URL=sqlite+pysqlite:////tmp/bad.db\n"
|
||||
"JWT_ACCESS_SECRET=x\n"
|
||||
"JWT_REFRESH_PEPPER=y\n"
|
||||
"POSTGRES_PASSWORD=z\n"
|
||||
"SECRETS_LOCKED=true\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("COMPTON_INSTALL_SECRETS_DIR", str(secrets_dir))
|
||||
monkeypatch.setattr(mod, "INSTALL_SECRETS_DIR", secrets_dir)
|
||||
monkeypatch.setattr(mod, "INSTALL_SECRETS_FILE", install_file)
|
||||
with pytest.raises(RuntimeError, match="PostgreSQL"):
|
||||
mod.ensure_install_secrets()
|
||||
|
||||
|
||||
def test_docker_entrypoint_rejects_sqlite(monkeypatch):
|
||||
from scripts import docker_entrypoint
|
||||
|
||||
monkeypatch.setattr(
|
||||
docker_entrypoint.settings,
|
||||
"database_url",
|
||||
"sqlite+pysqlite:////tmp/bad.db",
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="SQLite"):
|
||||
docker_entrypoint.wait_for_database(max_attempts=1, delay_seconds=0)
|
||||
@@ -0,0 +1,119 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core import jwt_denylist as denylist_module
|
||||
from app.core.jwt_denylist import (
|
||||
bump_auth_epoch,
|
||||
deny_jti,
|
||||
ensure_jwt_revocation_backend,
|
||||
get_auth_epoch,
|
||||
is_jti_denied,
|
||||
revoke_access_token,
|
||||
validate_access_claims,
|
||||
)
|
||||
from app.core.security import create_access_token, hash_password
|
||||
from app.modules.users.repository import create_user
|
||||
|
||||
|
||||
def test_auth_epoch_bump_invalidates_token():
|
||||
user = create_user("epoch-user@example.com", hash_password("Valid123"), status="active")
|
||||
token = create_access_token(user.id, user.role, user.is_superuser)
|
||||
from app.core.security import decode_access_token
|
||||
|
||||
payload = decode_access_token(token)
|
||||
validate_access_claims(payload)
|
||||
|
||||
bump_auth_epoch(user.id)
|
||||
with pytest.raises(ValueError, match="TOKEN_REVOKED"):
|
||||
validate_access_claims(payload)
|
||||
|
||||
|
||||
def test_deny_jti_blocks_specific_token():
|
||||
user = create_user("jti-user@example.com", hash_password("Valid123"), status="active")
|
||||
token = create_access_token(user.id, user.role, user.is_superuser)
|
||||
from app.core.security import decode_access_token
|
||||
|
||||
payload = decode_access_token(token)
|
||||
deny_jti(payload["jti"], payload["exp"])
|
||||
assert is_jti_denied(payload["jti"])
|
||||
with pytest.raises(ValueError, match="TOKEN_REVOKED"):
|
||||
validate_access_claims(payload)
|
||||
|
||||
|
||||
def test_revoke_access_token_helper():
|
||||
user = create_user("revoke-user@example.com", hash_password("Valid123"), status="active")
|
||||
token = create_access_token(user.id, user.role, user.is_superuser)
|
||||
revoke_access_token(token)
|
||||
from app.core.security import decode_access_token
|
||||
|
||||
with pytest.raises(ValueError, match="TOKEN_REVOKED"):
|
||||
validate_access_claims(decode_access_token(token))
|
||||
|
||||
|
||||
def test_get_auth_epoch_defaults_to_zero():
|
||||
assert get_auth_epoch("missing-user-id") == 0
|
||||
|
||||
|
||||
def test_validate_access_claims_rejects_missing_sub():
|
||||
with pytest.raises(ValueError, match="INVALID_TOKEN"):
|
||||
validate_access_claims({})
|
||||
|
||||
|
||||
def test_production_is_jti_denied_fail_closed(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module.settings, "app_env", "production")
|
||||
mock_client = MagicMock()
|
||||
mock_client.exists.side_effect = RuntimeError("redis down")
|
||||
with patch("app.core.jwt_denylist.get_redis_client", return_value=mock_client):
|
||||
assert is_jti_denied("any-jti") is True
|
||||
|
||||
|
||||
def test_revoke_access_token_ignores_invalid_token():
|
||||
revoke_access_token("not-a-jwt")
|
||||
|
||||
|
||||
def test_bump_auth_epoch_production_raises_when_redis_fails(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module.settings, "app_env", "production")
|
||||
mock_client = MagicMock()
|
||||
mock_client.incr.side_effect = RuntimeError("redis down")
|
||||
with patch("app.core.jwt_denylist.get_redis_client", return_value=mock_client):
|
||||
with pytest.raises(RuntimeError):
|
||||
bump_auth_epoch("user-x")
|
||||
|
||||
|
||||
def test_empty_jti_is_not_denied():
|
||||
assert is_jti_denied("") is False
|
||||
deny_jti("", 9999999999)
|
||||
|
||||
|
||||
def test_ensure_jwt_revocation_backend_requires_redis_in_production(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module.settings, "app_env", "production")
|
||||
with patch("app.core.jwt_denylist.get_redis_client", return_value=None):
|
||||
with pytest.raises(RuntimeError, match="Redis is required"):
|
||||
ensure_jwt_revocation_backend()
|
||||
|
||||
|
||||
def test_get_auth_epoch_production_raises_when_redis_fails(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module.settings, "app_env", "production")
|
||||
mock_client = MagicMock()
|
||||
mock_client.get.side_effect = RuntimeError("redis down")
|
||||
with patch("app.core.jwt_denylist.get_redis_client", return_value=mock_client):
|
||||
with pytest.raises(RuntimeError):
|
||||
get_auth_epoch("user-y")
|
||||
|
||||
|
||||
def test_deny_jti_production_raises_when_redis_fails(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module.settings, "app_env", "production")
|
||||
mock_client = MagicMock()
|
||||
mock_client.setex.side_effect = RuntimeError("redis down")
|
||||
with patch("app.core.jwt_denylist.get_redis_client", return_value=mock_client):
|
||||
with pytest.raises(RuntimeError):
|
||||
deny_jti("jti-123", int(__import__("time").time()) + 3600)
|
||||
|
||||
|
||||
def test_memory_denylist_purges_expired_jti(monkeypatch):
|
||||
monkeypatch.setattr(denylist_module, "_memory_denied_jti", {"expired-jti": 1.0})
|
||||
assert is_jti_denied("expired-jti") is False
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
from app.main import create_app
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
def test_create_app_includes_test_routes_when_enabled(monkeypatch):
|
||||
monkeypatch.setattr(settings, "enable_test_routes", True)
|
||||
app = create_app()
|
||||
paths = set(app.openapi()["paths"])
|
||||
assert "/api/v1/test/emails/latest-token" in paths
|
||||
@@ -0,0 +1,38 @@
|
||||
from app.core.media_signing import build_signed_media_url, verify_signed_media
|
||||
from app.core.config import settings
|
||||
import time
|
||||
|
||||
|
||||
def test_build_and_verify_signed_media_url():
|
||||
signed = build_signed_media_url("/api/v1/media/files/avatars/user/file.png")
|
||||
assert signed is not None
|
||||
assert "expires=" in signed
|
||||
assert "sig=" in signed
|
||||
|
||||
path = "avatars/user/file.png"
|
||||
query = signed.split("?", 1)[1]
|
||||
params = dict(part.split("=") for part in query.split("&"))
|
||||
assert int(params["expires"]) - int(time.time()) <= settings.media_url_ttl_seconds
|
||||
assert verify_signed_media(path, int(params["expires"]), params["sig"])
|
||||
|
||||
|
||||
def test_build_signed_media_url_none():
|
||||
assert build_signed_media_url(None) is None
|
||||
|
||||
|
||||
def test_verify_signed_media_rejects_expired_signature():
|
||||
signed = build_signed_media_url("/api/v1/media/files/avatars/user/file.png")
|
||||
assert signed is not None
|
||||
path = "avatars/user/file.png"
|
||||
query = signed.split("?", 1)[1]
|
||||
params = dict(part.split("=") for part in query.split("&"))
|
||||
assert not verify_signed_media(path, int(params["expires"]) - 10_000, params["sig"])
|
||||
|
||||
|
||||
def test_verify_signed_media_rejects_tampered_signature():
|
||||
signed = build_signed_media_url("/api/v1/media/files/avatars/user/file.png")
|
||||
assert signed is not None
|
||||
path = "avatars/user/file.png"
|
||||
query = signed.split("?", 1)[1]
|
||||
params = dict(part.split("=") for part in query.split("&"))
|
||||
assert not verify_signed_media(path, int(params["expires"]), "invalid")
|
||||
@@ -0,0 +1,6 @@
|
||||
from app.core.media_signing import build_signed_media_url
|
||||
|
||||
|
||||
def test_build_signed_media_url_passthrough():
|
||||
external = "https://cdn.example.com/avatar.png"
|
||||
assert build_signed_media_url(external) == external
|
||||
@@ -0,0 +1,15 @@
|
||||
from app.core.password_denylist import is_denied_password, load_denylist
|
||||
|
||||
|
||||
def test_password_denylist_blocks_common_password():
|
||||
assert is_denied_password("password123")
|
||||
|
||||
|
||||
def test_password_denylist_loads_custom_entries(tmp_path, monkeypatch):
|
||||
denylist_file = tmp_path / "denylist.txt"
|
||||
denylist_file.write_text("# comment\nCustomBad1\n", encoding="utf-8")
|
||||
monkeypatch.setattr("app.core.password_denylist.settings.password_denylist_path", str(denylist_file))
|
||||
denylist = load_denylist()
|
||||
assert "custombad1" in denylist
|
||||
assert is_denied_password("CustomBad1")
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.main import _assert_production_guards
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_app_env(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "development")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", True)
|
||||
monkeypatch.setattr(settings, "enable_docs", True)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", False)
|
||||
monkeypatch.setattr(settings, "cookie_secure", False)
|
||||
monkeypatch.setattr(settings, "jwt_access_secret", "dev-access-secret-32bytes-minimum!!")
|
||||
monkeypatch.setattr(settings, "jwt_refresh_pepper", "dev-refresh-pepper-32bytes-minimum!!")
|
||||
monkeypatch.setattr(
|
||||
settings,
|
||||
"database_url",
|
||||
"postgresql+psycopg://compton_app:secret@postgres:5432/compton?sslmode=require",
|
||||
)
|
||||
|
||||
|
||||
def test_production_guards_skip_in_development():
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_test_routes(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
with pytest.raises(RuntimeError, match="ENABLE_TEST_ROUTES"):
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_insecure_cookie(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", False)
|
||||
monkeypatch.setattr(settings, "enable_docs", False)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
with pytest.raises(RuntimeError, match="COOKIE_SECURE"):
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_default_db_credentials(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", False)
|
||||
monkeypatch.setattr(settings, "enable_docs", False)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
monkeypatch.setattr(settings, "cookie_secure", True)
|
||||
monkeypatch.setattr(settings, "database_url", "postgresql://user:pass@db/app?sslmode=require")
|
||||
with pytest.raises(RuntimeError, match="Default database credentials"):
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_placeholder_jwt(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", False)
|
||||
monkeypatch.setattr(settings, "enable_docs", False)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
monkeypatch.setattr(settings, "cookie_secure", True)
|
||||
monkeypatch.setattr(settings, "jwt_access_secret", "change-me-access")
|
||||
with pytest.raises(RuntimeError, match="JWT_ACCESS_SECRET"):
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_placeholder_refresh_pepper(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", False)
|
||||
monkeypatch.setattr(settings, "enable_docs", False)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
monkeypatch.setattr(settings, "cookie_secure", True)
|
||||
monkeypatch.setattr(settings, "jwt_access_secret", "prod-access-secret-32bytes-minimum!!")
|
||||
monkeypatch.setattr(settings, "jwt_refresh_pepper", "change-me-pepper")
|
||||
with pytest.raises(RuntimeError, match="JWT_REFRESH_PEPPER"):
|
||||
_assert_production_guards()
|
||||
|
||||
|
||||
def test_production_guards_reject_missing_sslmode(monkeypatch):
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
monkeypatch.setattr(settings, "enable_test_routes", False)
|
||||
monkeypatch.setattr(settings, "enable_docs", False)
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
monkeypatch.setattr(settings, "cookie_secure", True)
|
||||
monkeypatch.setattr(settings, "database_url", "postgresql://app:secret@db/app")
|
||||
with pytest.raises(RuntimeError, match="sslmode=require"):
|
||||
_assert_production_guards()
|
||||
@@ -0,0 +1,32 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.redis import check_rate_limit, client_ip, _buckets
|
||||
|
||||
|
||||
def test_check_rate_limit_uses_memory_fallback(monkeypatch):
|
||||
monkeypatch.setattr(settings, "enable_rate_limit", True)
|
||||
_buckets.clear()
|
||||
with patch("app.core.redis.get_redis_client", return_value=None):
|
||||
check_rate_limit("memory-key", limit=1, window_seconds=60)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
check_rate_limit("memory-key", limit=1, window_seconds=60)
|
||||
assert exc.value.status_code == 429
|
||||
|
||||
|
||||
def test_client_ip_honors_trusted_proxy(monkeypatch):
|
||||
monkeypatch.setattr(settings, "trusted_proxy_ips", "127.0.0.1")
|
||||
request = MagicMock(spec=Request)
|
||||
request.headers = {"X-Forwarded-For": "203.0.113.10, 127.0.0.1"}
|
||||
request.client.host = "127.0.0.1"
|
||||
assert client_ip(request) == "203.0.113.10"
|
||||
|
||||
|
||||
def test_client_ip_unknown_without_client():
|
||||
request = MagicMock(spec=Request)
|
||||
request.headers = {}
|
||||
request.client = None
|
||||
assert client_ip(request) == "unknown"
|
||||
@@ -0,0 +1,18 @@
|
||||
from pathlib import Path
|
||||
|
||||
from app.core.server_logging import build_uvicorn_log_config, ensure_server_log_file
|
||||
|
||||
|
||||
def test_ensure_server_log_file(tmp_path, monkeypatch):
|
||||
log_file = tmp_path / "logs" / "server.log"
|
||||
monkeypatch.setattr("app.core.server_logging.settings.server_log_path", str(log_file))
|
||||
path = ensure_server_log_file()
|
||||
assert path.is_file()
|
||||
assert "server log initialized" in path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_build_uvicorn_log_config(tmp_path, monkeypatch):
|
||||
log_file = tmp_path / "logs" / "server.log"
|
||||
monkeypatch.setattr("app.core.server_logging.settings.server_log_path", str(log_file))
|
||||
cfg = build_uvicorn_log_config()
|
||||
assert cfg["handlers"]["file"]["filename"] == str(log_file.resolve())
|
||||
@@ -0,0 +1,14 @@
|
||||
from app.core.storage import download_object, ensure_bucket, memory_store, upload_object
|
||||
|
||||
|
||||
def test_ensure_bucket_noop_in_memory():
|
||||
ensure_bucket()
|
||||
|
||||
|
||||
def test_memory_storage_roundtrip():
|
||||
memory_store.clear()
|
||||
upload_object("avatars/test.png", b"abc", "image/png")
|
||||
payload = download_object("avatars/test.png")
|
||||
assert payload == (b"abc", "image/png")
|
||||
assert download_object("missing") is None
|
||||
|
||||
Reference in New Issue
Block a user