yuqei
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from scripts import run_migration_static_checks
|
|
|
|
|
|
class MigrationStaticCheckTests(unittest.TestCase):
|
|
def test_extract_created_table_columns_handles_constraints_and_commas(self) -> None:
|
|
sql = """
|
|
create table if not exists sample_table (
|
|
id bigserial primary key,
|
|
amount numeric(20,10) not null,
|
|
payload jsonb not null default '{}'::jsonb,
|
|
unique (id, amount)
|
|
);
|
|
"""
|
|
|
|
result = run_migration_static_checks.extract_created_table_columns(sql)
|
|
|
|
self.assertEqual(result["sample_table"], {"id", "amount", "payload"})
|
|
|
|
def test_validate_migration_sequence_rejects_gaps(self) -> None:
|
|
files = [Path("001_create_a.sql"), Path("003_create_b.sql")]
|
|
|
|
errors = run_migration_static_checks.validate_migration_sequence(files)
|
|
|
|
self.assertEqual(errors, ["non_contiguous_migration_sequence:[1, 3]"])
|
|
|
|
def test_static_checks_pass_current_migrations(self) -> None:
|
|
result = run_migration_static_checks.run_migration_static_checks()
|
|
|
|
self.assertTrue(result["passed"], result["errors"])
|
|
self.assertGreater(result["migration_count"], 0)
|
|
self.assertEqual(result["seed_values_checked"]["venues.code"], ["OKX"])
|
|
|
|
def test_static_checks_report_missing_column(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "001_create_sample.sql"
|
|
path.write_text("create table if not exists sample (id bigserial primary key);", encoding="utf-8")
|
|
|
|
result = run_migration_static_checks.run_migration_static_checks(
|
|
migrations_dir=Path(tmpdir),
|
|
required_schema={"sample": ("id", "missing_column")},
|
|
)
|
|
|
|
self.assertFalse(result["passed"])
|
|
self.assertIn("missing_column:sample.missing_column", result["errors"])
|
|
|
|
def test_static_checks_report_missing_required_seed_value(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "001_create_venues.sql"
|
|
path.write_text("create table if not exists venues (id bigserial primary key, code text not null);", encoding="utf-8")
|
|
|
|
result = run_migration_static_checks.run_migration_static_checks(
|
|
migrations_dir=Path(tmpdir),
|
|
required_schema={"venues": ("id", "code")},
|
|
required_seed_values={"venues.code": ("OKX",)},
|
|
)
|
|
|
|
self.assertFalse(result["passed"])
|
|
self.assertIn("missing_seed_value:venues.code=OKX", result["errors"])
|
|
|
|
def test_main_can_emit_json(self) -> None:
|
|
output = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_migration_static_checks.main(env={"MIGRATION_CHECKS_OUTPUT_JSON": "1"})
|
|
|
|
summary = json.loads(output.getvalue())
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertTrue(summary["passed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|