busya
a9a453109c
CRITICAL (CVE-class): - C-4: Remove DEV_MODE fallback with hardcoded postgres:postgres credentials - C-3: WebSocket endpoints now require JWT or API key token (?token=) auth - C-1: Superset environment passwords encrypted via Fernet at rest in DB - H-4: SESSION_SECRET_KEY separated from JWT AUTH_SECRET_KEY - H-1: Log injection via %s-formatting instead of f-strings - H-5: X-Trace-ID header validated as UUID4 to prevent trace poisoning INFRASTRUCTURE: - New src/core/encryption.py — EncryptionManager extracted from llm_provider - Test DB: per-module sqlite:///:memory: with PRAGMA foreign_keys=ON - testcontainers PostgreSQL via TEST_DB=postgres env var - conftest temp-file global engine (no 10GB shared-cache leak) TEST FIXES (44 pre-existing → 0): - test_smoke_plugins: module-level sys.modules mock isolated to per-test fixture - test_migration_engine: EXT:Python:uuid → uuid syntax fix - test_dashboards_api: mock env attributes + correct patch target - test_constants_audit_fixes: sync expected constant names with actual - test_defensive_guards: patch.object instead of module-level Repo mock - test_clean_release_cli: removed empty config.json directory - FK violations: TaskRecord parents in log_persistence, Environment in mapping_service, ReleaseCandidate in candidate_manifest_services ORTHOGONAL TESTS (18 new): - test_security_orthogonal.py: bcrypt 72-byte limit, unicode, API key format invariants, Fernet robustness, encryption key lifecycle, log injection protection, JWT edge cases MODEL FIXES: - MetricSnapshot.job_id: nullable with SET NULL (was broken FK for aggregate prune snapshots, hidden by SQLite) CLEANUP: - Removed stale :memory:test_main/test_auth/test_tasks file databases - Removed duplicate #endregion in encryption_key.py
86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import tests.conftest # noqa: F401 — ensure conftest runs first
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
# ── Save original module before it gets mocked ──
|
|
_ORIG_DATABASE_MODULE = sys.modules.get('src.core.database')
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolate_database():
|
|
"""Isolate this test module from the real database.
|
|
|
|
Saves the real src.core.database, replaces it with a MagicMock,
|
|
and restores it after the test completes. This prevents PluginLoader
|
|
imports from triggering real database initialization.
|
|
"""
|
|
# Remove real module if loaded
|
|
sys.modules.pop('src.core.database', None)
|
|
# Insert mock
|
|
mock_db = MagicMock()
|
|
sys.modules['src.core.database'] = mock_db
|
|
yield
|
|
# Restore real module
|
|
sys.modules.pop('src.core.database', None)
|
|
if _ORIG_DATABASE_MODULE is not None:
|
|
sys.modules['src.core.database'] = _ORIG_DATABASE_MODULE
|
|
|
|
|
|
class TestPluginSmoke:
|
|
"""Smoke tests for plugin loading and initialization."""
|
|
|
|
def test_plugins_load_successfully(self):
|
|
"""
|
|
Verify that all standard plugins can be discovered and instantiated
|
|
by the PluginLoader without throwing errors (e.g., missing imports,
|
|
syntax errors, missing class declarations).
|
|
"""
|
|
from src.core.plugin_loader import PluginLoader
|
|
|
|
plugin_dir = os.path.join(str(Path(__file__).parent.parent), "src", "plugins")
|
|
|
|
# This will discover and instantiate plugins
|
|
loader = PluginLoader(plugin_dir)
|
|
|
|
plugins = loader.get_all_plugin_configs()
|
|
plugin_ids = {p.id for p in plugins}
|
|
|
|
# We expect at least the migration and git plugins to be present
|
|
expected_plugins = {"superset-migration", "git-integration"}
|
|
|
|
missing_plugins = expected_plugins - plugin_ids
|
|
assert not missing_plugins, f"Missing expected plugins: {missing_plugins}"
|
|
|
|
@pytest.mark.anyio
|
|
async def test_task_manager_initializes_with_plugins(self):
|
|
"""
|
|
Verify that the TaskManager can initialize with the real PluginLoader.
|
|
"""
|
|
from src.core.plugin_loader import PluginLoader
|
|
from src.core.task_manager.manager import TaskManager
|
|
|
|
plugin_dir = os.path.join(str(Path(__file__).parent.parent), "src", "plugins")
|
|
loader = PluginLoader(plugin_dir)
|
|
|
|
# Initialize TaskManager with real loader
|
|
with patch("src.core.task_manager.manager.TaskPersistenceService") as MockPersistence, \
|
|
patch("src.core.task_manager.manager.TaskLogPersistenceService"):
|
|
|
|
MockPersistence.return_value.load_tasks.return_value = []
|
|
|
|
with patch("src.dependencies.config_manager"):
|
|
manager = TaskManager(loader)
|
|
|
|
# Stop the flusher thread to prevent hanging
|
|
manager._flusher_stop_event.set()
|
|
manager._flusher_thread.join(timeout=2)
|
|
|
|
assert manager is not None
|