Robert Helewka
81426327bf
Replace plaintext token storage with SHA-256 hashes so leaked database contents cannot be used to authenticate. Plaintext is generated, shown once at creation time, and never persisted. - Add `hash_token()` helper and `MCPTokenManager.create_token()` that returns `(instance, plaintext)`. - Replace `token` field with indexed `token_hash`; look up bearers by hashing the incoming value. - Update dashboard, management command, and admin to surface plaintext only at creation. Disable admin "add" since it cannot reveal plaintext. - Migration drops the old `token` column and adds `token_hash`; pre-existing tokens are invalidated and must be reissued.
126 lines
5.2 KiB
Python
126 lines
5.2 KiB
Python
"""
|
|
Django Test Framework Integration.
|
|
|
|
Custom test runner that spins up PostgreSQL (pgvector) and Neo4j containers.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any, Dict, List
|
|
|
|
from django.test.runner import DiscoverRunner
|
|
|
|
from .config import TestDatabaseConfig, TestNeo4jConfig
|
|
from .manager import DockerNeo4jManager, DockerPostgreSQLManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PostgreSQLTestRunner(DiscoverRunner):
|
|
"""
|
|
Custom Django test runner using Docker PostgreSQL + Neo4j containers.
|
|
|
|
Usage in settings.py or pytest.ini:
|
|
TEST_RUNNER = "test_db_manager.django_integration.PostgreSQLTestRunner"
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.pg_manager: DockerPostgreSQLManager | None = None
|
|
self.neo4j_manager: DockerNeo4jManager | None = None
|
|
self._original_databases: Dict[str, Dict[str, Any]] = {}
|
|
|
|
def setup_databases(self, **kwargs) -> List[tuple]:
|
|
"""Spin up containers, swap DATABASES, then delegate to Django."""
|
|
from django.conf import settings
|
|
|
|
self._original_databases = dict(settings.DATABASES)
|
|
|
|
# ── PostgreSQL ─────────────────────────────────────────────────
|
|
pg_config = TestDatabaseConfig(
|
|
name=os.environ.get("TEST_DB_NAME", "mnemosyne"),
|
|
memory_limit=os.environ.get("TEST_DB_MEMORY", "512m"),
|
|
port=int(os.environ.get("TEST_DB_PORT", "0")),
|
|
)
|
|
self.pg_manager = DockerPostgreSQLManager(pg_config)
|
|
self.pg_manager.create_container()
|
|
|
|
if not self.pg_manager.wait_for_ready(timeout=pg_config.connection_timeout):
|
|
self.pg_manager.cleanup()
|
|
raise RuntimeError("PostgreSQL test container failed to start")
|
|
|
|
from django.db import connections
|
|
|
|
db_cfg = self.pg_manager.get_django_database_config()
|
|
# Preserve Django's defaulted top-level keys (ATOMIC_REQUESTS,
|
|
# AUTOCOMMIT, OPTIONS, …) and the TEST sub-dict (CHARSET, MIRROR,
|
|
# MIGRATE, …) — these are populated lazily by Django and absent from
|
|
# the user's raw settings.DATABASES, so a naive overwrite breaks
|
|
# request handling that consults them.
|
|
existing = connections["default"].settings_dict
|
|
existing_test = existing.get("TEST", {})
|
|
merged = {**existing, **db_cfg}
|
|
merged["TEST"] = {**existing_test, **db_cfg.get("TEST", {})}
|
|
settings.DATABASES["default"] = merged
|
|
# The default connection was instantiated at Django bootstrap; its
|
|
# settings_dict is independent of settings.DATABASES. Sync it
|
|
# manually so test code talks to the container, not the dev DB.
|
|
connections["default"].settings_dict.update(merged)
|
|
logger.info("PostgreSQL test DB ready on port %s", self.pg_manager.assigned_port)
|
|
|
|
# ── Neo4j ──────────────────────────────────────────────────────
|
|
neo4j_enabled = os.environ.get("TEST_NEO4J_ENABLED", "1") == "1"
|
|
if neo4j_enabled:
|
|
neo4j_config = TestNeo4jConfig(
|
|
name=os.environ.get("TEST_NEO4J_NAME", "mnemosyne"),
|
|
memory_limit=os.environ.get("TEST_NEO4J_MEMORY", "1g"),
|
|
port_bolt=int(os.environ.get("TEST_NEO4J_BOLT_PORT", "0")),
|
|
port_http=int(os.environ.get("TEST_NEO4J_HTTP_PORT", "0")),
|
|
)
|
|
self.neo4j_manager = DockerNeo4jManager(neo4j_config)
|
|
self.neo4j_manager.create_container()
|
|
|
|
if not self.neo4j_manager.wait_for_ready(timeout=neo4j_config.connection_timeout):
|
|
self.neo4j_manager.cleanup()
|
|
self.pg_manager.cleanup()
|
|
raise RuntimeError("Neo4j test container failed to start")
|
|
|
|
# Configure neomodel to use the test container
|
|
bolt_url = self.neo4j_manager.get_bolt_url()
|
|
settings.NEOMODEL_NEO4J_BOLT_URL = bolt_url
|
|
os.environ["NEO4J_BOLT_URL"] = bolt_url
|
|
|
|
try:
|
|
from neomodel import config as neo_config
|
|
|
|
neo_config.DATABASE_URL = bolt_url
|
|
except ImportError:
|
|
pass
|
|
|
|
logger.info("Neo4j test DB ready on bolt port %s", self.neo4j_manager.assigned_bolt_port)
|
|
|
|
# Containers were just created — DB already exists, so flip keepdb to
|
|
# skip "CREATE DATABASE" (which would fail; test_user is not superuser).
|
|
# Django still runs migrations to populate the schema.
|
|
self.keepdb = True
|
|
return super().setup_databases(**kwargs)
|
|
|
|
def teardown_databases(self, old_config, **kwargs) -> None:
|
|
"""Tear down databases and clean up containers."""
|
|
super().teardown_databases(old_config, **kwargs)
|
|
|
|
from django.conf import settings
|
|
|
|
if self.neo4j_manager:
|
|
self.neo4j_manager.cleanup()
|
|
if self.pg_manager:
|
|
self.pg_manager.cleanup()
|
|
|
|
# Restore original database config
|
|
settings.DATABASES = dict(self._original_databases)
|
|
|
|
|
|
def get_postgresql_test_runner():
|
|
"""Factory function returning the test runner class."""
|
|
return PostgreSQLTestRunner
|