50 lines
2.0 KiB
Python
50 lines
2.0 KiB
Python
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
UP_PATH = PROJECT_ROOT / "database" / "012_monthly_report_publication.sql"
|
|
DOWN_PATH = PROJECT_ROOT / "database" / "012_monthly_report_publication.down.sql"
|
|
|
|
|
|
class MonthlyPublicationMigrationTests(unittest.TestCase):
|
|
def test_forward_migration_is_additive_metadata_only(self) -> None:
|
|
sql = UP_PATH.read_text(encoding="utf-8")
|
|
|
|
for required in (
|
|
"current_database() <> 'booking_test'",
|
|
"CREATE SCHEMA reporting",
|
|
"CREATE TABLE reporting.monthly_runs",
|
|
"CREATE TABLE reporting.monthly_run_daily_versions",
|
|
"CREATE TABLE reporting.monthly_channel_manifest",
|
|
"monthly_runs_one_active_period_idx",
|
|
"source_snapshot_sha256",
|
|
"storage_provider IN ('oss', 's3', 'local_fixture', 'local')",
|
|
"published monthly run requires a continuous reconciled channel manifest",
|
|
"Greatest ARRIVAL included in this publication",
|
|
):
|
|
with self.subTest(required=required):
|
|
self.assertIn(required, sql)
|
|
|
|
self.assertNotIn("DROP SCHEMA finance", sql)
|
|
self.assertNotIn("DROP SCHEMA ingestion", sql)
|
|
self.assertNotIn("CREATE TABLE finance.report_versions", sql)
|
|
self.assertNotIn("full_name", sql.lower())
|
|
self.assertNotIn("confirmation_no", sql.lower())
|
|
|
|
def test_rollback_refuses_after_publication_data_exists(self) -> None:
|
|
sql = DOWN_PATH.read_text(encoding="utf-8")
|
|
|
|
self.assertIn("current_database() <> 'booking_test'", sql)
|
|
self.assertIn("SELECT 1 FROM reporting.monthly_runs", sql)
|
|
self.assertIn("storage_provider = 'local'", sql)
|
|
self.assertIn(
|
|
"refusing rollback: monthly publication or controlled local artifact data exists",
|
|
sql,
|
|
)
|
|
self.assertIn("DROP SCHEMA reporting CASCADE", sql)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|