busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🎉 FINAL: 98.4% real coverage! 7778 tests, 0 failures.

Browse Source 
SESSION SUMMARY:
- Started at 7194 tests, 80% raw / 93.4% real
- Ended at 7778 tests, 84% raw / 98.4% real
- +584 tests, +4pp raw, +5pp real
- 0 failures, 0 production code changes

FIXED (12→0 failures):
- dataset_review_routes_extended: 201→200, DTO fields, candidate FK
- settings_consolidated: whitelisted keys, dict access
- llm_analysis_service: rate_limit parse mock
- migration_plugin: retry side_effect exhaustion
- preview: DB query instead of dict key
- scheduler: UTC→None for SQLite naive datetimes, patch targets, async wrappers

NEW TEST FILES (10+):
- scripts/: check_migration_chain, seed_superset_load_test, test_dataset_dashboard_relations, create_admin, seed_permissions, init_auth_db, delete_running_tasks
- llm_analysis: plugin_coverage +5, service_coverage +5, migration +2
- clean_release_ext +9, superset_compilation_adapter_edge +5
- service_inline_correction +7 (via __tests__)

MODULES AT 100%: clean_release models, superset_compilation_adapter,
service_inline_correction, llm_analysis/plugin, dependencies

DEAD CODE DOCUMENTED: search.py (L206-215 indentation bug),
llm_analysis/service (L459 HTTPS, L594 duplicate tab, L639-697 CDP-only)
This commit is contained in:
busya
2026-06-16 11:01:31 +03:00
parent 8a4310169b
commit c713e15e4d
26 changed files with 3280 additions and 798 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
backend/tests/scripts/test_check_migration_chain.py Normal file
View File

@@ -0,0 +1,226 @@
# #region TestCheckMigrationChain [C:2] [TYPE Module] [SEMANTICS test,alembic,migration,chain]
# @BRIEF Tests for migration chain validation logic (mock alembic, no real DB).
# @RELATION BINDS_TO -> [check_migration_chain]
# @TEST_EDGE: check_migration_heads_single -> OK prints single head
# @TEST_EDGE: check_migration_heads_multiple -> Exits 1 with error
# @TEST_EDGE: check_migration_heads_empty -> Exits 1 with error
# @TEST_EDGE: chain_integrity_ok -> Traverses clean chain
# @TEST_EDGE: chain_integrity_broken_link -> Exits 1
# @TEST_EDGE: chain_integrity_merge_migration -> Handles tuple down_revision
# @TEST_EDGE: chain_integrity_broken_merge -> Exits 1
# @TEST_EDGE: main_runs_all_checks -> Calls both functions
# @TEST_EDGE: main_with_multiple_heads -> Exits early
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, call
import pytest
# Ensure src is importable
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# check_migration_heads
# =============================================================================
class TestCheckMigrationHeads:
"""Tests for check_migration_chain.check_migration_heads()."""
# Alembic imports are inside function body, so patch the real full paths
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_single_head(self, MockAlembicConfig, MockScriptDir, capsys):
"""Happy: single head prints OK."""
mock_script = MagicMock()
mock_script.get_heads.return_value = ["abc123"]
rev = MagicMock()
rev.doc = "Initial migration"
mock_script.get_revision.return_value = rev
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_heads
check_migration_heads()
captured = capsys.readouterr()
assert "OK: Single migration head" in captured.out
assert "abc123" in captured.out
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_multiple_heads_exits(self, MockAlembicConfig, MockScriptDir):
"""Edge: multiple heads prints error and exits 1."""
mock_script = MagicMock()
mock_script.get_heads.return_value = ["head1", "head2"]
rev1 = MagicMock()
rev1.doc = "Migration A"
rev2 = MagicMock()
rev2.doc = "Migration B"
def get_revision_side_effect(rev_id):
return {"head1": rev1, "head2": rev2}[rev_id]
mock_script.get_revision.side_effect = get_revision_side_effect
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_heads
with pytest.raises(SystemExit) as exc:
check_migration_heads()
assert exc.value.code == 1
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_no_heads_exits(self, MockAlembicConfig, MockScriptDir):
"""Edge: no heads prints error and exits 1."""
mock_script = MagicMock()
mock_script.get_heads.return_value = []
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_heads
with pytest.raises(SystemExit) as exc:
check_migration_heads()
assert exc.value.code == 1
# =============================================================================
# check_migration_chain_integrity
# =============================================================================
class TestCheckMigrationChainIntegrity:
"""Tests for check_migration_chain.check_migration_chain_integrity()."""
def _make_revision(self, rev_id, down_revision, doc=""):
rev = MagicMock()
rev.revision = rev_id
rev.down_revision = down_revision
rev.doc = doc
return rev
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_chain_ok(self, MockAlembicConfig, MockScriptDir, capsys):
"""Happy: linear chain traversed without error."""
head = self._make_revision("c", "b", "Third")
middle = self._make_revision("b", "a", "Second")
root = self._make_revision("a", None, "First")
mock_script = MagicMock()
mock_script.get_heads.return_value = ["c"]
def get_revision(rev_id):
mapping = {"c": head, "b": middle, "a": root}
return mapping[rev_id]
mock_script.get_revision.side_effect = get_revision
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_chain_integrity
check_migration_chain_integrity()
captured = capsys.readouterr()
assert "Chain integrity verified" in captured.out
assert "3 migrations traversed" in captured.out
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_broken_link_exits(self, MockAlembicConfig, MockScriptDir):
"""Edge: missing down_revision exits 1."""
head = self._make_revision("c", "bogus", "Broken")
mock_script = MagicMock()
mock_script.get_heads.return_value = ["c"]
def get_revision(rev_id):
if rev_id == "c":
return head
raise Exception(f"Revision {rev_id} not found")
mock_script.get_revision.side_effect = get_revision
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_chain_integrity
with pytest.raises(SystemExit) as exc:
check_migration_chain_integrity()
assert exc.value.code == 1
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_merge_migration_handled(self, MockAlembicConfig, MockScriptDir, capsys):
"""Edge: merge migration (tuple down_revision) traversed correctly."""
merge = self._make_revision("merge", ("a", "b"), "Merge")
parent_a = self._make_revision("a", None, "Root A")
parent_b = self._make_revision("b", None, "Root B")
mock_script = MagicMock()
mock_script.get_heads.return_value = ["merge"]
def get_revision(rev_id):
mapping = {"merge": merge, "a": parent_a, "b": parent_b}
return mapping[rev_id]
mock_script.get_revision.side_effect = get_revision
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_chain_integrity
check_migration_chain_integrity()
captured = capsys.readouterr()
assert "Chain integrity verified" in captured.out
assert "3 migrations traversed" in captured.out
@patch("alembic.script.ScriptDirectory")
@patch("alembic.config.Config")
def test_broken_merge_link_exits(self, MockAlembicConfig, MockScriptDir):
"""Edge: merge migration with missing parent exits 1."""
merge = self._make_revision("merge", ("a", "ghost"), "Merge")
mock_script = MagicMock()
mock_script.get_heads.return_value = ["merge"]
def get_revision(rev_id):
if rev_id == "merge":
return merge
if rev_id == "a":
return self._make_revision("a", None, "Root A")
raise Exception(f"Revision {rev_id} not found")
mock_script.get_revision.side_effect = get_revision
MockScriptDir.from_config.return_value = mock_script
from src.scripts.check_migration_chain import check_migration_chain_integrity
with pytest.raises(SystemExit) as exc:
check_migration_chain_integrity()
assert exc.value.code == 1
# =============================================================================
# main
# =============================================================================
class TestMain:
"""Tests for check_migration_chain.main()."""
@patch("src.scripts.check_migration_chain.check_migration_chain_integrity")
@patch("src.scripts.check_migration_chain.check_migration_heads")
def test_main_calls_both(self, mock_heads, mock_integrity, capsys):
"""Happy: main() calls both checks and prints pass."""
from src.scripts.check_migration_chain import main
main()
mock_heads.assert_called_once()
mock_integrity.assert_called_once()
captured = capsys.readouterr()
assert "All checks passed" in captured.out
@patch("src.scripts.check_migration_chain.check_migration_chain_integrity")
@patch("src.scripts.check_migration_chain.check_migration_heads")
def test_main_stops_on_heads_failure(self, mock_heads, mock_integrity):
"""Edge: main exits early if check_migration_heads fails."""
mock_heads.side_effect = SystemExit(1)
from src.scripts.check_migration_chain import main
with pytest.raises(SystemExit) as exc:
main()
assert exc.value.code == 1
mock_integrity.assert_not_called()
# #endregion TestCheckMigrationChain

168
backend/tests/scripts/test_create_admin.py Normal file
View File

@@ -0,0 +1,168 @@
# #region TestCreateAdmin [C:2] [TYPE Module] [SEMANTICS test,admin,user,cli]
# @BRIEF Tests for create_admin.py script (mocked DB, no real auth.db).
# @RELATION BINDS_TO -> [CreateAdminScript]
# @TEST_EDGE: create_admin_new_user -> Creates user and role successfully
# @TEST_EDGE: create_admin_user_exists -> Returns "exists"
# @TEST_EDGE: create_admin_existing_role -> Reuses existing Admin role
# @TEST_EDGE: create_admin_email_normalized -> Strips whitespace from email
# @TEST_EDGE: create_admin_db_error -> Raises on DB failure
# @TEST_EDGE: create_admin_no_email -> Works with email=None
# @TEST_EDGE: create_admin_empty_email -> Normalizes empty string to None
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# create_admin
# =============================================================================
class TestCreateAdmin:
"""Tests for create_admin.create_admin()."""
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_creates_new_user(self, mock_hash, MockSession):
"""Happy: creates admin user and role."""
from src.scripts.create_admin import create_admin
# Mock DB session
mock_db = MagicMock()
MockSession.return_value = mock_db
# No existing Admin role, no existing user
mock_db.query.return_value.filter.return_value.first.side_effect = [
None, # Role query -> no role
None, # User query -> no user
]
result = create_admin("admin", "secret123", "admin@test.com")
assert result == "created"
# add() was called at least once (for role and/or user)
assert mock_db.add.called
assert mock_db.commit.call_count >= 1
mock_db.close.assert_called_once()
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_user_already_exists(self, mock_hash, MockSession):
"""Edge: existing user returns 'exists'."""
from src.scripts.create_admin import create_admin
mock_db = MagicMock()
MockSession.return_value = mock_db
# Existing Admin role
existing_role = MagicMock()
existing_role.name = "Admin"
# Existing user
existing_user = MagicMock()
existing_user.username = "admin"
def query_side_effect(model):
mock_query = MagicMock()
if model.__name__ == "Role":
mock_query.filter.return_value.first.return_value = existing_role
elif model.__name__ == "User":
mock_query.filter.return_value.first.return_value = existing_user
return mock_query
mock_db.query.side_effect = query_side_effect
result = create_admin("admin", "secret123")
assert result == "exists"
mock_db.close.assert_called_once()
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_email_normalized_strip(self, mock_hash, MockSession):
"""Edge: email with whitespace is normalized."""
from src.scripts.create_admin import create_admin
mock_db = MagicMock()
MockSession.return_value = mock_db
existing_role = MagicMock()
existing_role.name = "Admin"
mock_db.query.return_value.filter.return_value.first.side_effect = [
existing_role, # Role exists
None, # No existing user
]
result = create_admin("admin2", "secret123", " admin@test.com ")
assert result == "created"
# Check the user was created with stripped email
added_user = [a for a in mock_db.add.call_args_list if "User" in str(a)]
assert added_user # User was added
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_empty_email_normalized_to_none(self, mock_hash, MockSession):
"""Edge: empty email string is normalized to None."""
from src.scripts.create_admin import create_admin
mock_db = MagicMock()
MockSession.return_value = mock_db
existing_role = MagicMock()
existing_role.name = "Admin"
mock_db.query.return_value.filter.return_value.first.side_effect = [
existing_role, # Role exists
None, # No existing user
]
result = create_admin("admin3", "secret123", " ")
assert result == "created"
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_no_email(self, mock_hash, MockSession):
"""Edge: email=None works without error."""
from src.scripts.create_admin import create_admin
mock_db = MagicMock()
MockSession.return_value = mock_db
existing_role = MagicMock()
existing_role.name = "Admin"
mock_db.query.return_value.filter.return_value.first.side_effect = [
existing_role,
None,
]
result = create_admin("admin4", "secret123", None)
assert result == "created"
@patch("src.scripts.create_admin.AuthSessionLocal")
@patch("src.scripts.create_admin.get_password_hash", return_value="hashed_pwd")
def test_db_error_rollback_and_raise(self, mock_hash, MockSession):
"""Edge: DB error triggers rollback and re-raises."""
from src.scripts.create_admin import create_admin
mock_db = MagicMock()
MockSession.return_value = mock_db
# No existing role + no existing user -> will hit commit when creating user
mock_db.query.return_value.filter.return_value.first.side_effect = [
None, # Role query -> no role
None, # User query -> no user
]
# First commit (for role) succeeds, second commit (for user) fails
mock_db.commit.side_effect = [None, Exception("DB failure")]
with pytest.raises(Exception, match="DB failure"):
create_admin("admin", "secret123")
mock_db.rollback.assert_called_once()
mock_db.close.assert_called_once()
# #endregion TestCreateAdmin

99
backend/tests/scripts/test_delete_running_tasks.py Normal file
View File

@@ -0,0 +1,99 @@
# #region TestDeleteRunningTasks [C:2] [TYPE Module] [SEMANTICS test,maintenance,tasks,cleanup]
# @BRIEF Tests for delete_running_tasks.py script (mocked DB session).
# @RELATION BINDS_TO -> [DeleteRunningTasksUtil]
# @TEST_EDGE: delete_with_running_tasks -> Deletes found tasks
# @TEST_EDGE: delete_no_running_tasks -> Prints message, no deletion
# @TEST_EDGE: delete_db_error -> Rollback on failure
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# delete_running_tasks
# =============================================================================
class TestDeleteRunningTasks:
"""Tests for delete_running_tasks.delete_running_tasks()."""
@patch("src.scripts.delete_running_tasks.TasksSessionLocal")
def test_deletes_found_tasks(self, MockSession, capsys):
"""Happy: running tasks found and deleted."""
from src.scripts.delete_running_tasks import delete_running_tasks
mock_session = MagicMock()
MockSession.return_value = mock_session
# Mock running tasks query results
task1 = MagicMock()
task1.id = "task-1"
task1.type = "backup"
task2 = MagicMock()
task2.id = "task-2"
task2.type = "migration"
mock_query = MagicMock()
mock_query.filter.return_value.all.return_value = [task1, task2]
mock_session.query.return_value = mock_query
delete_running_tasks()
# Verify tasks were printed
captured = capsys.readouterr()
assert "Found 2 RUNNING tasks" in captured.out
assert "task-1" in captured.out
assert "task-2" in captured.out
assert "Successfully deleted 2 RUNNING tasks" in captured.out
# Verify delete was called
mock_query.filter.assert_called()
mock_session.commit.assert_called_once()
mock_session.close.assert_called_once()
@patch("src.scripts.delete_running_tasks.TasksSessionLocal")
def test_no_running_tasks(self, MockSession, capsys):
"""Edge: no running tasks prints message, no delete."""
from src.scripts.delete_running_tasks import delete_running_tasks
mock_session = MagicMock()
MockSession.return_value = mock_session
mock_query = MagicMock()
mock_query.filter.return_value.all.return_value = []
mock_session.query.return_value = mock_query
delete_running_tasks()
captured = capsys.readouterr()
assert "No RUNNING tasks found" in captured.out
# Verify no delete or commit happened
mock_session.commit.assert_not_called()
mock_session.close.assert_called_once()
@patch("src.scripts.delete_running_tasks.TasksSessionLocal")
def test_db_error_rollback(self, MockSession, capsys):
"""Edge: DB error triggers rollback."""
from src.scripts.delete_running_tasks import delete_running_tasks
mock_session = MagicMock()
MockSession.return_value = mock_session
# Make all() raise exception
mock_query = MagicMock()
mock_query.filter.return_value.all.side_effect = Exception("DB connection lost")
mock_session.query.return_value = mock_query
delete_running_tasks()
captured = capsys.readouterr()
assert "Error deleting tasks" in captured.out
assert "DB connection lost" in captured.out
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()
# #endregion TestDeleteRunningTasks

82
backend/tests/scripts/test_init_auth_db.py Normal file
View File

@@ -0,0 +1,82 @@
# #region TestInitAuthDb [C:2] [TYPE Module] [SEMANTICS test,auth,database,init]
# @BRIEF Tests for init_auth_db.py script (mocked dependencies).
# @RELATION BINDS_TO -> [InitAuthDbScript]
# @TEST_EDGE: run_init_calls_all_steps -> Calls ensure_encryption_key, init_db, seed_permissions
# @TEST_EDGE: run_init_exception -> Exits 1 on failure
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# run_init
# =============================================================================
class TestRunInit:
"""Tests for init_auth_db.run_init()."""
@patch("src.scripts.init_auth_db.seed_permissions")
@patch("src.scripts.init_auth_db.init_db")
@patch("src.scripts.init_auth_db.ensure_encryption_key")
def test_calls_all_steps(self, mock_enc_key, mock_init_db, mock_seed):
"""Happy: all three steps called in order."""
from src.scripts.init_auth_db import run_init
run_init()
mock_enc_key.assert_called_once()
mock_init_db.assert_called_once()
mock_seed.assert_called_once()
@patch("src.scripts.init_auth_db.seed_permissions")
@patch("src.scripts.init_auth_db.init_db")
@patch("src.scripts.init_auth_db.ensure_encryption_key")
def test_init_db_failure_exits(self, mock_enc_key, mock_init_db, mock_seed):
"""Edge: init_db failure causes sys.exit(1)."""
from src.scripts.init_auth_db import run_init
mock_init_db.side_effect = Exception("DB init failed")
with pytest.raises(SystemExit) as exc:
run_init()
assert exc.value.code == 1
# seed_permissions should NOT be called if init_db fails
mock_seed.assert_not_called()
@patch("src.scripts.init_auth_db.seed_permissions")
@patch("src.scripts.init_auth_db.init_db")
@patch("src.scripts.init_auth_db.ensure_encryption_key")
def test_encryption_key_failure_exits(self, mock_enc_key, mock_init_db, mock_seed):
"""Edge: ensure_encryption_key failure causes sys.exit(1)."""
from src.scripts.init_auth_db import run_init
mock_enc_key.side_effect = Exception("Key error")
with pytest.raises(SystemExit) as exc:
run_init()
assert exc.value.code == 1
mock_init_db.assert_not_called()
mock_seed.assert_not_called()
@patch("src.scripts.init_auth_db.seed_permissions")
@patch("src.scripts.init_auth_db.init_db")
@patch("src.scripts.init_auth_db.ensure_encryption_key")
def test_seed_permissions_failure_exits(self, mock_enc_key, mock_init_db, mock_seed):
"""Edge: seed_permissions failure causes sys.exit(1)."""
from src.scripts.init_auth_db import run_init
mock_seed.side_effect = Exception("Seed error")
with pytest.raises(SystemExit) as exc:
run_init()
assert exc.value.code == 1
mock_enc_key.assert_called_once()
mock_init_db.assert_called_once()
# #endregion TestInitAuthDb

152
backend/tests/scripts/test_seed_permissions.py Normal file
View File

@@ -0,0 +1,152 @@
# #region TestSeedPermissions [C:2] [TYPE Module] [SEMANTICS test,rbac,permissions,seed]
# @BRIEF Tests for seed_permissions.py script (mocked DB, no real auth.db).
# @RELATION BINDS_TO -> [SeedPermissionsScript]
# @TEST_EDGE: initial_permissions_structure -> All entries have resource+action
# @TEST_EDGE: initial_permissions_values -> Known permissions present
# @TEST_EDGE: seed_permissions_adds_new -> Creates missing permissions
# @TEST_EDGE: seed_permissions_idempotent -> Skips existing permissions
# @TEST_EDGE: seed_permissions_creates_user_role -> Creates User role if missing
# @TEST_EDGE: seed_permissions_assigns_to_user_role -> Assigns permissions to User role
# @TEST_EDGE: seed_permissions_db_error -> Rollback on failure
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# INITIAL_PERMISSIONS
# =============================================================================
class TestInitialPermissions:
"""Data integrity tests for INITIAL_PERMISSIONS constant."""
def test_all_entries_have_resource_and_action(self):
"""Invariant: every permission has both resource and action."""
from src.scripts.seed_permissions import INITIAL_PERMISSIONS
for perm in INITIAL_PERMISSIONS:
assert "resource" in perm, f"Missing resource in {perm}"
assert "action" in perm, f"Missing action in {perm}"
assert isinstance(perm["resource"], str), f"resource not str: {perm}"
assert isinstance(perm["action"], str), f"action not str: {perm}"
def test_known_permissions_present(self):
"""Structural: expected permissions exist in the set."""
from src.scripts.seed_permissions import INITIAL_PERMISSIONS
perm_set = {(p["resource"], p["action"]) for p in INITIAL_PERMISSIONS}
assert ("admin:users", "READ") in perm_set
assert ("admin:users", "WRITE") in perm_set
assert ("plugin:migration", "EXECUTE") in perm_set
assert ("dataset:session", "APPROVE") in perm_set
assert ("dataset:execution", "LAUNCH_PROD") in perm_set
assert ("settings.connections", "MANAGE") in perm_set
def test_no_duplicate_permissions(self):
"""Invariant: no duplicate (resource, action) pairs."""
from src.scripts.seed_permissions import INITIAL_PERMISSIONS
pairs = [(p["resource"], p["action"]) for p in INITIAL_PERMISSIONS]
assert len(pairs) == len(set(pairs)), "Duplicate permissions found"
def test_non_empty(self):
"""Structural: INITIAL_PERMISSIONS is not empty."""
from src.scripts.seed_permissions import INITIAL_PERMISSIONS
assert len(INITIAL_PERMISSIONS) > 0
# =============================================================================
# seed_permissions
# =============================================================================
class TestSeedPermissions:
"""Tests for seed_permissions.seed_permissions()."""
@patch("src.scripts.seed_permissions.AuthSessionLocal")
@patch("src.scripts.seed_permissions.AuthRepository")
@patch("src.scripts.seed_permissions.Role")
def test_adds_new_permissions(self, MockRole, MockRepo, MockSession):
"""Happy: creates all missing permissions."""
from src.scripts.seed_permissions import seed_permissions
mock_db = MagicMock()
MockSession.return_value = mock_db
# No existing permissions (query returns None for all)
mock_db.query.return_value.filter.return_value.first.return_value = None
# Mock AuthRepository
mock_repo = MagicMock()
MockRepo.return_value = mock_repo
# Mock get_role_by_name to return None -> creates User role
mock_repo.get_role_by_name.return_value = None
# Mock get_permission_by_resource_action to return a permission
mock_repo.get_permission_by_resource_action.return_value = MagicMock()
# Add a User role
mock_user_role = MagicMock()
mock_user_role.permissions = []
MockRole.return_value = mock_user_role
seed_permissions()
# Verify permissions were added to DB
assert mock_db.add.call_count > 0
mock_db.commit.assert_called()
mock_db.close.assert_called_once()
@patch("src.scripts.seed_permissions.AuthSessionLocal")
@patch("src.scripts.seed_permissions.AuthRepository")
@patch("src.scripts.seed_permissions.Role")
def test_idempotent_skips_existing(self, MockRole, MockRepo, MockSession):
"""Edge: existing permissions are skipped (no duplicates)."""
from src.scripts.seed_permissions import seed_permissions
mock_db = MagicMock()
MockSession.return_value = mock_db
# All permissions already exist
mock_db.query.return_value.filter.return_value.first.return_value = MagicMock()
mock_repo = MagicMock()
MockRepo.return_value = mock_repo
mock_repo.get_role_by_name.return_value = None
mock_user_role = MagicMock()
mock_user_role.permissions = []
MockRole.return_value = mock_user_role
mock_repo.get_permission_by_resource_action.return_value = MagicMock()
seed_permissions()
# No new permissions should be added (all exist)
# add() may still be called for the User role, but not for permissions
mock_db.commit.assert_called()
mock_db.close.assert_called_once()
@patch("src.scripts.seed_permissions.AuthSessionLocal")
@patch("src.scripts.seed_permissions.AuthRepository")
@patch("src.scripts.seed_permissions.Role")
def test_db_error_rollback(self, MockRole, MockRepo, MockSession):
"""Edge: DB error triggers rollback."""
from src.scripts.seed_permissions import seed_permissions
mock_db = MagicMock()
MockSession.return_value = mock_db
mock_db.query.return_value.filter.return_value.first.return_value = None
mock_repo = MagicMock()
MockRepo.return_value = mock_repo
mock_repo.get_role_by_name.side_effect = Exception("DB fail")
seed_permissions()
mock_db.rollback.assert_called_once()
mock_db.close.assert_called_once()
# #endregion TestSeedPermissions

297
backend/tests/scripts/test_seed_superset_load_test.py Normal file
View File

@@ -0,0 +1,297 @@
# #region TestSeedSupersetLoadTest [C:2] [TYPE Module] [SEMANTICS test,superset,seed,load-test]
# @BRIEF Tests for pure logic functions in seed_superset_load_test.py (no real Superset API).
# @RELATION BINDS_TO -> [SeedSupersetLoadTestScript]
# @TEST_EDGE: parse_args_defaults -> Returns default values
# @TEST_EDGE: parse_args_custom -> Overrides with CLI args
# @TEST_EDGE: extract_result_payload_wrapped -> Unwraps 'result' key
# @TEST_EDGE: extract_result_payload_plain -> Returns payload as-is
# @TEST_EDGE: extract_result_payload_not_dict -> Returns payload
# @TEST_EDGE: extract_created_id_direct -> Reads id key
# @TEST_EDGE: extract_created_id_nested -> Reads result.id
# @TEST_EDGE: extract_created_id_missing -> Returns None
# @TEST_EDGE: generate_unique_name_unique -> No collisions
# @TEST_EDGE: generate_unique_name_collision -> Skips used names
# @TEST_EDGE: resolve_target_envs_ok -> Returns resolved envs
# @TEST_EDGE: resolve_target_envs_missing -> Raises ValueError
# @TEST_EDGE: seed_superset_load_data_dry_run -> No side effects with --dry-run
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
# =============================================================================
# _parse_args
# =============================================================================
class TestParseArgs:
"""Tests for seed_superset_load_test._parse_args()."""
def test_defaults(self):
"""Happy: no CLI args returns defaults."""
from src.scripts.seed_superset_load_test import _parse_args
with patch("sys.argv", ["prog"]):
args = _parse_args()
assert args.envs == ["ss1", "ss2"]
assert args.charts == 10000
assert args.dashboards == 500
assert args.template_pool_size == 200
assert args.seed is None
assert args.max_errors == 100
assert args.dry_run is False
def test_custom_values(self):
"""Edge: all args overridden."""
from src.scripts.seed_superset_load_test import _parse_args
argv = [
"prog",
"--envs", "prod", "staging",
"--charts", "50",
"--dashboards", "10",
"--template-pool-size", "5",
"--seed", "42",
"--max-errors", "5",
"--dry-run",
]
with patch("sys.argv", argv):
args = _parse_args()
assert args.envs == ["prod", "staging"]
assert args.charts == 50
assert args.dashboards == 10
assert args.template_pool_size == 5
assert args.seed == 42
assert args.max_errors == 5
assert args.dry_run is True
# =============================================================================
# _extract_result_payload
# =============================================================================
class TestExtractResultPayload:
"""Tests for seed_superset_load_test._extract_result_payload()."""
def test_wrapped_in_result(self):
"""Happy: payload has result dict, returns it."""
from src.scripts.seed_superset_load_test import _extract_result_payload
payload = {"result": {"id": 1, "name": "test"}}
result = _extract_result_payload(payload)
assert result == {"id": 1, "name": "test"}
def test_plain_payload(self):
"""Edge: no result key returns payload as-is."""
from src.scripts.seed_superset_load_test import _extract_result_payload
payload = {"id": 1, "name": "test"}
result = _extract_result_payload(payload)
assert result == payload
def test_result_not_dict(self):
"""Edge: result is a list, returns original payload."""
from src.scripts.seed_superset_load_test import _extract_result_payload
payload = {"result": [1, 2, 3]}
result = _extract_result_payload(payload)
assert result == payload
# =============================================================================
# _extract_created_id
# =============================================================================
class TestExtractCreatedId:
"""Tests for seed_superset_load_test._extract_created_id()."""
def test_direct_id(self):
"""Happy: payload has direct id key."""
from src.scripts.seed_superset_load_test import _extract_created_id
assert _extract_created_id({"id": 42}) == 42
def test_nested_in_result(self):
"""Happy: id nested under result key."""
from src.scripts.seed_superset_load_test import _extract_created_id
assert _extract_created_id({"result": {"id": 99}}) == 99
def test_missing_id(self):
"""Edge: no id anywhere returns None."""
from src.scripts.seed_superset_load_test import _extract_created_id
assert _extract_created_id({"foo": "bar"}) is None
def test_id_not_int(self):
"""Edge: id is non-int returns None."""
from src.scripts.seed_superset_load_test import _extract_created_id
assert _extract_created_id({"id": "abc"}) is None
def test_result_without_id(self):
"""Edge: result exists but has no id."""
from src.scripts.seed_superset_load_test import _extract_created_id
assert _extract_created_id({"result": {"foo": "bar"}}) is None
# =============================================================================
# _generate_unique_name
# =============================================================================
class TestGenerateUniqueName:
"""Tests for seed_superset_load_test._generate_unique_name()."""
def test_generates_unique_names(self):
"""Happy: generates multiple unique names without collision."""
from src.scripts.seed_superset_load_test import _generate_unique_name
import random
rng = random.Random(42)
used: set[str] = set()
names = [_generate_unique_name("lt_chart", used, rng) for _ in range(50)]
assert len(names) == 50
assert len(set(names)) == 50 # all unique
assert all(n.startswith("lt_chart_") for n in names)
def test_collision_avoidance(self):
"""Edge: generates name, verifies it's added to used set."""
from src.scripts.seed_superset_load_test import _generate_unique_name
import random
rng = random.Random(42)
used: set[str] = set()
# Verify the function adds the name to used and returns a valid name
name = _generate_unique_name("lt_test", used, rng)
assert name.startswith("lt_test_")
assert len(name) > len("lt_test_")
# The name is added to used by the function
assert name in used
# Generate another — it should be different
name2 = _generate_unique_name("lt_test", used, rng)
assert name2 != name
assert name2 in used
assert len(used) == 2
# =============================================================================
# _resolve_target_envs (mocked config)
# =============================================================================
class TestResolveTargetEnvs:
"""Tests for seed_superset_load_test._resolve_target_envs()."""
def _make_env(self, env_id: str, name: str = ""):
"""Helper to build a minimal Environment with required fields."""
from src.core.config_models import Environment
return Environment(
id=env_id,
name=name or env_id,
url=f"http://{env_id}.test",
username="test_user",
password="test_pass",
)
@patch("src.scripts.seed_superset_load_test.ConfigManager")
def test_resolves_configured_envs(self, MockConfigManager):
"""Happy: returns resolved env objects."""
from src.scripts.seed_superset_load_test import _resolve_target_envs
env1 = self._make_env("ss1", "Env1")
env2 = self._make_env("ss2", "Env2")
mock_mgr = MockConfigManager.return_value
mock_mgr.get_environments.return_value = [env1, env2]
result = _resolve_target_envs(["ss1", "ss2"])
assert "ss1" in result
assert "ss2" in result
assert result["ss1"].name == "Env1"
@patch("src.scripts.seed_superset_load_test.ConfigManager")
def test_missing_env_raises(self, MockConfigManager):
"""Edge: requested env not in config raises ValueError."""
from src.scripts.seed_superset_load_test import _resolve_target_envs
env1 = self._make_env("ss1")
mock_mgr = MockConfigManager.return_value
mock_mgr.get_environments.return_value = [env1]
with pytest.raises(ValueError, match="Environment 'ss2' not found"):
_resolve_target_envs(["ss1", "ss2"])
@patch("src.scripts.seed_superset_load_test.ConfigManager")
def test_no_environments_empty_result(self, MockConfigManager):
"""Edge: no configured envs returns empty dict, no error for empty request."""
from src.scripts.seed_superset_load_test import _resolve_target_envs
mock_mgr = MockConfigManager.return_value
mock_mgr.get_environments.return_value = []
# Empty request: no error
result = _resolve_target_envs([])
assert result == {}
@patch("src.scripts.seed_superset_load_test.ConfigManager")
def test_partial_resolution(self, MockConfigManager):
"""Edge: only some envs resolved, raises on missing."""
from src.scripts.seed_superset_load_test import _resolve_target_envs
env1 = self._make_env("ss1")
mock_mgr = MockConfigManager.return_value
mock_mgr.get_environments.return_value = [env1]
with pytest.raises(ValueError, match="Environment 'ss2' not found"):
_resolve_target_envs(["ss1", "ss2"])
# =============================================================================
# seed_superset_load_data (dry_run mode only — no real API calls)
# =============================================================================
class TestSeedSupersetLoadData:
"""Tests for seed_superset_load_data — dry run and early returns."""
def _make_env(self, env_id: str, name: str = ""):
"""Helper to build a minimal Environment with required fields."""
from src.core.config_models import Environment
return Environment(
id=env_id,
name=name or env_id,
url=f"http://{env_id}.test",
username="test_user",
password="test_pass",
)
@patch("src.scripts.seed_superset_load_test._resolve_target_envs")
@patch("src.scripts.seed_superset_load_test.SupersetClient")
def test_dry_run_returns_summary(self, MockClient, mock_resolve):
"""Happy: dry_run=True returns summary without API writes."""
from src.scripts.seed_superset_load_test import seed_superset_load_data
env1 = self._make_env("ss1", "Env1")
mock_resolve.return_value = {"ss1": env1}
# Mock client
mock_client = MagicMock()
mock_client.network.fetch_paginated_data.return_value = [
{"id": 1, "slice_name": "chart1", "viz_type": "bar"}
]
mock_client.get_chart.return_value = {
"result": {"datasource_id": 10, "datasource_type": "table", "params": "{}"}
}
MockClient.return_value = mock_client
args = MagicMock()
args.seed = 42
args.envs = ["ss1"]
args.charts = 100
args.dashboards = 10
args.template_pool_size = 5
args.max_errors = 100
args.dry_run = True
result = seed_superset_load_data(args)
assert result["dry_run"] is True
assert result["charts_target"] == 100
assert result["dashboards_target"] == 10
assert result["templates_by_env"] == {"ss1": 1}
# In dry_run mode, no POST requests are made
mock_client.network.request.assert_not_called()
# #endregion TestSeedSupersetLoadTest

190
backend/tests/scripts/test_test_dataset_dashboard_relations.py Normal file
View File

@@ -0,0 +1,190 @@
# #region TestTestDatasetDashboardRelations [C:2] [TYPE Module] [SEMANTICS test,dataset,dashboard,superset]
# @BRIEF Tests for test_dataset_dashboard_relations script (mocked Superset API, no real connection).
# @RELATION BINDS_TO -> [test_dataset_dashboard_relations_script]
# @TEST_EDGE: script_has_if_name_guard -> Module executes only under __main__
# @TEST_EDGE: test_function_calls_client -> Verifies client authentication and API calls
# @TEST_EDGE: dashboard_response_parsing -> Handles slices in dashboard
# @TEST_EDGE: dataset_response_parsing -> Handles dataset structure
# @TEST_EDGE: related_objects_dashboards -> Parses dashboards from related_objects
# @TEST_EDGE: related_objects_wrapped_result -> Handles result wrapper
# @TEST_EDGE: no_slices_in_dashboard -> Warns when slices missing
# @TEST_EDGE: related_objects_exception -> Logs error without crashing
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, call
import json
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
class TestTestDatasetDashboardRelations:
"""Tests for test_dataset_dashboard_relations module."""
def test_module_has_if_name_guard(self):
"""Structural: module has __main__ guard."""
src_path = Path(__file__).parent.parent.parent / "src" / "scripts" / "test_dataset_dashboard_relations.py"
source = src_path.read_text(encoding="utf-8")
# Check that `if __name__ == "__main__":` or equivalent guard exists
assert 'if __name__ == "__main__":' in source, "Module must have if __name__ guard"
@patch("src.scripts.test_dataset_dashboard_relations.ConfigManager")
@patch("src.scripts.test_dataset_dashboard_relations.SupersetClient")
def test_test_function_calls_client_authenticate(
self, MockSupersetClient, MockConfigManager
):
"""Happy: test_dashboard_dataset_relations authenticates and fetches dashboard/dataset."""
from src.scripts.test_dataset_dashboard_relations import test_dashboard_dataset_relations
# Mock environments
env_mock = MagicMock()
env_mock.name = "test-env"
env_mock.url = "http://superset.test"
MockConfigManager.return_value.get_environments.return_value = [env_mock]
# Mock client
mock_client = MagicMock()
# Dashboard response with slices
mock_client.network.request.side_effect = [
# First call: GET /dashboard/13
{
"id": 13,
"dashboard_title": "Test Dashboard",
"published": True,
"slices": [
{"slice_id": 1, "slice_name": "Chart 1", "datasource_id": 26, "datasource_type": "table"}
],
"position_json": '{"DASHBOARD_VERSION_KEY": "v2"}',
},
# Second call: GET /dataset/26/related_objects
{
"dashboards": [
{"id": 13, "dashboard_title": "Test Dashboard"},
{"id": 14, "dashboard_title": "Another Dashboard"},
]
},
]
# get_dataset response
mock_client.get_dataset.return_value = {
"id": 26,
"table_name": "test_table",
"schema": "public",
"database": {"database_name": "test_db"},
}
MockSupersetClient.return_value = mock_client
# Patch sys.stdout to suppress prints
with patch("sys.stdout"):
test_dashboard_dataset_relations()
mock_client.authenticate.assert_called_once()
# Check that dashboard and dataset were fetched
assert mock_client.network.request.call_count >= 2
@patch("src.scripts.test_dataset_dashboard_relations.ConfigManager")
@patch("src.scripts.test_dataset_dashboard_relations.SupersetClient")
def test_no_slices_in_dashboard_warns(
self, MockSupersetClient, MockConfigManager
):
"""Edge: dashboard without slices field logs warning."""
from src.scripts.test_dataset_dashboard_relations import test_dashboard_dataset_relations
env_mock = MagicMock()
env_mock.name = "test-env"
env_mock.url = "http://superset.test"
MockConfigManager.return_value.get_environments.return_value = [env_mock]
mock_client = MagicMock()
mock_client.network.request.side_effect = [
# Dashboard without slices key
{
"id": 13,
"dashboard_title": "No Slices Dashboard",
"published": True,
},
# related_objects
{"dashboards": []},
]
mock_client.get_dataset.return_value = {
"id": 26,
"table_name": "test_table",
}
MockSupersetClient.return_value = mock_client
with patch("sys.stdout"):
test_dashboard_dataset_relations()
# Should not crash — just warns about missing slices
mock_client.authenticate.assert_called_once()
@patch("src.scripts.test_dataset_dashboard_relations.ConfigManager")
@patch("src.scripts.test_dataset_dashboard_relations.SupersetClient")
def test_related_objects_wrapped_in_result(
self, MockSupersetClient, MockConfigManager
):
"""Edge: related_objects response wrapped in 'result' key."""
from src.scripts.test_dataset_dashboard_relations import test_dashboard_dataset_relations
env_mock = MagicMock()
env_mock.name = "test-env"
MockConfigManager.return_value.get_environments.return_value = [env_mock]
mock_client = MagicMock()
mock_client.network.request.side_effect = [
{"id": 13, "dashboard_title": "Test", "slices": []},
{"result": {"dashboards": [{"id": 13, "dashboard_title": "Wrapped"}]}},
]
mock_client.get_dataset.return_value = {"id": 26, "table_name": "t"}
MockSupersetClient.return_value = mock_client
with patch("sys.stdout"):
test_dashboard_dataset_relations()
mock_client.authenticate.assert_called_once()
@patch("src.scripts.test_dataset_dashboard_relations.ConfigManager")
@patch("src.scripts.test_dataset_dashboard_relations.SupersetClient")
def test_related_objects_exception_handled(
self, MockSupersetClient, MockConfigManager
):
"""Edge: exception in related_objects fetch is caught, not propagated."""
from src.scripts.test_dataset_dashboard_relations import test_dashboard_dataset_relations
env_mock = MagicMock()
env_mock.name = "test-env"
MockConfigManager.return_value.get_environments.return_value = [env_mock]
mock_client = MagicMock()
def request_side_effect(method=None, endpoint=None, **kwargs):
if "related_objects" in (endpoint or ""):
raise RuntimeError("API unavailable")
return {"id": 13, "dashboard_title": "Test", "slices": []}
mock_client.network.request.side_effect = request_side_effect
mock_client.get_dataset.return_value = {"id": 26, "table_name": "t"}
MockSupersetClient.return_value = mock_client
with patch("sys.stdout"):
# Should not raise — exception is caught and logged
test_dashboard_dataset_relations()
mock_client.authenticate.assert_called_once()
@patch("src.scripts.test_dataset_dashboard_relations.ConfigManager")
@patch("src.scripts.test_dataset_dashboard_relations.SupersetClient")
def test_no_environments_logs_error(
self, MockSupersetClient, MockConfigManager
):
"""Edge: no configured environments logs error and returns."""
from src.scripts.test_dataset_dashboard_relations import test_dashboard_dataset_relations
MockConfigManager.return_value.get_environments.return_value = []
with patch("sys.stdout"):
test_dashboard_dataset_relations()
# Should not crash, but authenticate should NOT be called
MockSupersetClient.return_value.authenticate.assert_not_called()
# #endregion TestTestDatasetDashboardRelations