"""Offline PostgreSQL DDL checks for the articles/cards migration; no database access.""" import io import os import unittest from pathlib import Path from unittest.mock import patch os.environ['DATABASE_URL'] = 'sqlite:///:memory:' from alembic import command from alembic.config import Config from alembic.script import ScriptDirectory class ArticlesMigrationTests(unittest.TestCase): def test_models_register_fresh_and_offline_upgrade_downgrade(self): import subprocess import sys result = subprocess.run([sys.executable, '-c', "import app.models; from app.database import Base; " "assert {'articles','question_article_links','flashcard_question_links','flashcard_article_links'} <= set(Base.metadata.tables)"], capture_output=True, text=True, env={**os.environ, 'DATABASE_URL': 'sqlite:///:memory:'}) self.assertEqual(result.returncode, 0, result.stderr) output = io.StringIO() config = Config(output_buffer=output) config.set_main_option('script_location', str(Path(__file__).resolve().parents[1] / 'alembic')) scripts = ScriptDirectory.from_config(config) self.assertEqual(len(scripts.get_heads()), 1) self.assertEqual(scripts.get_revision('e8d4f1a27c93').down_revision, 'd94a26b8f302') with patch.dict(os.environ, {'DATABASE_URL': 'postgresql://unused@127.0.0.1/offline_only'}): command.upgrade(config, 'd94a26b8f302:e8d4f1a27c93', sql=True) sql = output.getvalue() for table in ('articles', 'question_article_links', 'flashcard_question_links', 'flashcard_article_links'): self.assertIn(f'CREATE TABLE IF NOT EXISTS {table}', sql) self.assertIn('UNIQUE (question_id, article_id, section_id)', sql) self.assertIn('ON DELETE CASCADE', sql) output.seek(0) output.truncate() command.downgrade(config, 'e8d4f1a27c93:d94a26b8f302', sql=True) for table in ('flashcard_article_links', 'flashcard_question_links', 'question_article_links', 'articles'): self.assertIn(f'DROP TABLE IF EXISTS {table}', output.getvalue()) if __name__ == '__main__': unittest.main()