Files
crank/tests/unit/test_check_migration_boundaries.py

49 lines
1.9 KiB
Python

import subprocess
import tempfile
import unittest
from pathlib import Path
SCRIPT = Path(__file__).parents[2] / "scripts" / "check-migration-boundaries.py"
class MigrationBoundaryTests(unittest.TestCase):
def run_check(self, root: Path) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["python3", str(SCRIPT), "--root", str(root)],
text=True,
capture_output=True,
check=False,
)
def test_rejects_ddl_runner_and_sql_include_outside_authority(self) -> None:
samples = (
'sqlx::query("CREATE UNIQUE INDEX idx ON sample(id)");',
'sqlx::query("CREATE TEMP TABLE sample(id int)");',
'sqlx::query("CREATE TYPE state AS ENUM (\'ready\')");',
'sqlx::migrate!("./migrations");',
'include_str!("local.sql");',
'sqlx::query("INSERT INTO __crank_migrations VALUES (1)");',
)
for index, sample in enumerate(samples):
with self.subTest(index=index), tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "apps" / "sample" / "src" / "main.rs"
source.parent.mkdir(parents=True)
source.write_text(sample, encoding="utf-8")
result = self.run_check(root)
self.assertNotEqual(result.returncode, 0)
def test_allows_canonical_migration_assets(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "crates" / "crank-registry" / "src" / "migrations" / "v1.sql"
source.parent.mkdir(parents=True)
source.write_text("CREATE TABLE sample(id int);", encoding="utf-8")
result = self.run_check(root)
self.assertEqual(result.returncode, 0, result.stderr)
if __name__ == "__main__":
unittest.main()