Files
ss-tools/backend/tests/core/test_encryption.py
busya 3133e50645 perf: fix translate deadlock, speed, trace_id, UI bugs — fullstack patch
## Backend (7 production files + 6 test files)

### P0-2: LLM output truncation cascade fix
- _token_budget.py: OUTPUT_PER_ROW_PER_LANG 120→200, OUTPUT_SAFETY_FACTOR 0.70→0.55
- Prevents finish_reason=length → split → retry cascade (3 calls → 1 call per batch)
- P2-8: added qwen-flash/qwen-plus/qwen-max/qwen-coder to PROVIDER_DEFAULTS

### P1-4/P1-5: EncryptionManager singleton
- encryption.py: get_encryption_manager() process-wide singleton
- llm_provider.py: use singleton instead of new EncryptionManager() per batch
- Eliminates ~90 redundant Fernet key validations per translation run

### P1-6: Cache-hit log aggregation
- _batch_proc.py: one log per batch (batch_rows + cache_hits) instead of per-row
- 1076 log lines → ~30 per run

### P1-7: Timezone-aware datetime fix
- scheduler.py: _ensure_aware() helper for naive DB datetime → UTC-aware
- Fixes TypeError in scheduled translation concurrency check

### P2-9: Connection test timeout
- connection_service.py: asyncio.wait_for(15s) on all dialect tests
- Prevents 2-minute UI hangs from DNS/TCP stalls

### Trace ID propagation
- middleware/trace.py: inject x-trace-id response header via ASGI send wrapper

### Test fixes & integration tests
- test_scheduler.py: AsyncMock for execute_run, mock get_async_job_runner
- test_sql_insert_service.py: AsyncMock for execute_sql
- test_token_budget.py: batch_size 50→45 for new OUTPUT_PER_ROW_PER_LANG=200
- test_encryption.py: +2 singleton tests
- test_scheduler_ensure_aware.py: +4 (naive→aware, passthrough, None, subtraction)
- test_batch_classify_persist.py: +2 cache-hit aggregation tests
- test_connection_service_edge.py: +2 timeout tests
- test_trace_middleware.py: +4 x-trace-id header tests
- test_token_budget.py: +4 qwen-flash/O200 tests

## Frontend (7 production files + 5 test files)

### Trace ID propagation
- api.ts: _captureTraceId() reads x-trace-id → setTraceId() in fetchApi/requestApi/postApi/deleteApi

### Duplicate datasource columns fetch
- ConfigTabForm.svelte: guard availableColumns.length === 0 before fetch

### Admin pages Svelte 5 runes fix
- admin/users/+page.svelte: plain let → () for all template-bound vars
- admin/roles/+page.svelte: same fix
- Both pages were stuck on «Загрузка...» due to mixed reactivity models

### Validation popover positioning
- +page.svelte: pass trigger HTMLElement instead of event
- DashboardHubModel.svelte.ts: toggleValidationPopover(HTMLElement), closeValidationPopover()
- Added X close button + click-outside overlay + i18n

### Test fixes & integration tests
- api.test.ts: mock setTraceId/getTraceId, +3 _captureTraceId tests
- provider_config.integration.test.ts: handleDelete→promptDeleteProvider
- DatasetPreview.test.ts: dashboards/ → ROUTES.dashboards
- test_config_tab_form.svelte.js: +2 columns fetch guard tests (NEW)
- admin-users.test.ts: +3 loading→table tests (NEW)
- admin-roles.test.ts: +2 loading→table tests (NEW)

## Semantic curation
- Removed @COMPLEXITY N from 6 route files + metrics.py (duplicate of [C:N])
- Added [C:N] to 2 orphan child contracts in metrics.py
- Added [C:N] + @BRIEF to 4 frontend anchors
- Fixed #region → # #region consistency in validation_tasks.py

## Verification
- Backend: 608 pytest passed (0 failures)
- Frontend: 2472 vitest passed (128 files, 0 failures)
- Frontend build: ✓ built in 18s
- Browser: dashboards, admin/users, admin/roles, validation popover — all green
2026-06-18 23:54:57 +03:00

218 lines
10 KiB
Python

# #region Test.Encryption [C:3] [TYPE Module] [SEMANTICS test,encryption,fernet,key,crypto]
# @BRIEF Verify EncryptionManager encrypt/decrypt and ensure_encryption_key resolution contracts.
# @RELATION BINDS_TO -> [EncryptionCore]
# @RELATION BINDS_TO -> [EncryptionKeyModule]
# @TEST_EDGE: missing_key -> RuntimeError when ENCRYPTION_KEY absent
# @TEST_EDGE: invalid_key -> RuntimeError when ENCRYPTION_KEY is not valid Fernet format
# @TEST_EDGE: external_fail -> decrypt of garbage data raises Exception
# @TEST_INVARIANT: symmetric_encryption -> VERIFIED_BY: encrypt_decrypt_cycle, empty_string_encryption
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
import pytest
# Hardcoded valid Fernet key — generated once, frozen as fixture.
# 32 url-safe base64 bytes. Never recompute at test time (anti-tautology).
_VALID_FERNET_KEY = "MldUHg5kwSAcPnnYxmhWDS6ASb6e_bWQRV5gtwHrjQ0="
# ── _require_fernet_key ─────────────────────────────────────────────────────
class TestRequireFernetKey:
"""_require_fernet_key — loads and validates Fernet key from environment."""
# #region test_require_key_valid [C:2] [TYPE Function]
# @BRIEF Valid ENCRYPTION_KEY returns key bytes without error.
def test_require_key_valid(self, monkeypatch):
from src.core.encryption import _require_fernet_key
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
result = _require_fernet_key()
assert result == _VALID_FERNET_KEY.encode()
# #endregion test_require_key_valid
# #region test_require_key_missing [C:2] [TYPE Function]
# @BRIEF Missing ENCRYPTION_KEY raises RuntimeError with startup guidance.
def test_require_key_missing(self, monkeypatch):
from src.core.encryption import _require_fernet_key
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY must be set"):
_require_fernet_key()
# #endregion test_require_key_missing
# #region test_require_key_invalid [C:2] [TYPE Function]
# @BRIEF Malformed ENCRYPTION_KEY raises RuntimeError about valid Fernet key.
def test_require_key_invalid(self, monkeypatch):
from src.core.encryption import _require_fernet_key
monkeypatch.setenv("ENCRYPTION_KEY", "not-a-valid-fernet-key")
with pytest.raises(RuntimeError, match="valid Fernet key"):
_require_fernet_key()
# #endregion test_require_key_invalid
# #region test_require_key_whitespace_only [C:2] [TYPE Function]
# @BRIEF Whitespace-only ENCRYPTION_KEY is treated as missing.
def test_require_key_whitespace_only(self, monkeypatch):
from src.core.encryption import _require_fernet_key
monkeypatch.setenv("ENCRYPTION_KEY", " ")
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY must be set"):
_require_fernet_key()
# #endregion test_require_key_whitespace_only
# ── EncryptionManager ───────────────────────────────────────────────────────
class TestEncryptionManager:
"""EncryptionManager — encrypt/decrypt operations with validated Fernet key."""
# #region test_encrypt_decrypt_cycle [C:2] [TYPE Function]
# @BRIEF Encrypt then decrypt returns original plaintext (symmetric invariant).
def test_encrypt_decrypt_cycle(self, monkeypatch):
from src.core.encryption import EncryptionManager
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
mgr = EncryptionManager()
# Hardcoded fixture — plaintext declared before assertion, not computed.
plaintext = "my_secret_key"
encrypted = mgr.encrypt(plaintext)
assert encrypted != plaintext
assert mgr.decrypt(encrypted) == plaintext
# #endregion test_encrypt_decrypt_cycle
# #region test_empty_string_encryption [C:2] [TYPE Function]
# @BRIEF Empty string survives encrypt/decrypt cycle.
def test_empty_string_encryption(self, monkeypatch):
from src.core.encryption import EncryptionManager
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
mgr = EncryptionManager()
encrypted = mgr.encrypt("")
assert mgr.decrypt(encrypted) == ""
# #endregion test_empty_string_encryption
# #region test_decrypt_garbage_raises [C:2] [TYPE Function]
# @BRIEF Decrypting non-Fernet data raises an exception (external_fail edge).
def test_decrypt_garbage_raises(self, monkeypatch):
from src.core.encryption import EncryptionManager
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
mgr = EncryptionManager()
with pytest.raises(Exception):
mgr.decrypt("not-encrypted-data")
# #endregion test_decrypt_garbage_raises
# #region test_init_missing_key_raises [C:2] [TYPE Function]
# @BRIEF EncryptionManager() without ENCRYPTION_KEY raises RuntimeError at init.
def test_init_missing_key_raises(self, monkeypatch):
from src.core.encryption import EncryptionManager
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY must be set"):
EncryptionManager()
# #endregion test_init_missing_key_raises
# ── ensure_encryption_key ───────────────────────────────────────────────────
class TestEnsureEncryptionKey:
"""ensure_encryption_key — resolve key from env or .env file, crash-early if absent."""
# #region test_ensure_key_from_env [C:2] [TYPE Function]
# @BRIEF Returns key from process environment when ENCRYPTION_KEY is set.
def test_ensure_key_from_env(self, monkeypatch):
from src.core.encryption_key import ensure_encryption_key
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
result = ensure_encryption_key()
assert result == _VALID_FERNET_KEY
# #endregion test_ensure_key_from_env
# #region test_ensure_key_from_dotenv_file [C:2] [TYPE Function]
# @BRIEF Loads key from .env file when not in process environment; sets os.environ.
def test_ensure_key_from_dotenv_file(self, monkeypatch, tmp_path):
from src.core.encryption_key import ensure_encryption_key
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
env_file = tmp_path / ".env"
env_file.write_text(f"ENCRYPTION_KEY={_VALID_FERNET_KEY}\nOTHER=value\n")
result = ensure_encryption_key(env_file)
assert result == _VALID_FERNET_KEY
# Side-effect: key must be propagated to process environment.
import os
assert os.environ.get("ENCRYPTION_KEY") == _VALID_FERNET_KEY
# #endregion test_ensure_key_from_dotenv_file
# #region test_ensure_key_no_file_raises [C:2] [TYPE Function]
# @BRIEF Crash-early when no .env file exists and env var is unset.
def test_ensure_key_no_file_raises(self, monkeypatch):
from src.core.encryption_key import ensure_encryption_key
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY is not set"):
ensure_encryption_key(Path("/nonexistent/.env"))
# #endregion test_ensure_key_no_file_raises
# #region test_ensure_key_file_without_key_raises [C:2] [TYPE Function]
# @BRIEF Crash-early when .env file exists but contains no ENCRYPTION_KEY line.
def test_ensure_key_file_without_key_raises(self, monkeypatch, tmp_path):
from src.core.encryption_key import ensure_encryption_key
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
env_file = tmp_path / ".env"
env_file.write_text("OTHER=value\nUNRELATED=stuff\n")
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY is not set"):
ensure_encryption_key(env_file)
# #endregion test_ensure_key_file_without_key_raises
# ── get_encryption_manager singleton ────────────────────────────────────────
class TestGetEncryptionManager:
"""get_encryption_manager() — process-wide singleton."""
# #region test_singleton_returns_same_instance [C:2] [TYPE Function] [SEMANTICS test,singleton]
# @BRIEF Two calls to get_encryption_manager return the identical object.
def test_singleton_returns_same_instance(self, monkeypatch):
from src.core.encryption import get_encryption_manager
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
# Reset module-level singleton
import src.core.encryption as enc_mod
enc_mod._encryption_manager = None
mgr1 = get_encryption_manager()
mgr2 = get_encryption_manager()
assert mgr1 is mgr2
# #endregion test_singleton_returns_same_instance
# #region test_singleton_after_exception [C:2] [TYPE Function] [SEMANTICS test,singleton,recovery]
# @BRIEF If first creation fails (no ENCRYPTION_KEY), singleton resets
# properly — subsequent call with valid key succeeds and is cached.
def test_singleton_after_exception(self, monkeypatch):
from src.core.encryption import get_encryption_manager
# Ensure no key — first call raises
monkeypatch.delenv("ENCRYPTION_KEY", raising=False)
import src.core.encryption as enc_mod
enc_mod._encryption_manager = None
with pytest.raises(RuntimeError, match="ENCRYPTION_KEY must be set"):
get_encryption_manager()
# Singleton remains None after exception
assert enc_mod._encryption_manager is None
# Now set valid key — second call succeeds
monkeypatch.setenv("ENCRYPTION_KEY", _VALID_FERNET_KEY)
mgr = get_encryption_manager()
assert mgr is not None
# And subsequent call returns same instance
assert get_encryption_manager() is mgr
# #endregion test_singleton_after_exception
# #endregion Test.Encryption