test_startup_creates_every_table.py 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566
  1. """Startup creates the table of every model (#3502).
  2. init_db() imports the model modules so create_all knows their tables. A module
  3. missing from that list (and from models/__init__) gets no table on a real
  4. start, while the test suite, which imports every model itself, never notices:
  5. hms_notified_faults (#3486) was never created that way. So this starts a clean
  6. Python process, runs init_db() against an empty SQLite database and checks the
  7. tables against the ones the model files declare.
  8. """
  9. import os
  10. import re
  11. import sqlite3
  12. import subprocess
  13. import sys
  14. from pathlib import Path
  15. import pytest
  16. REPO = Path(__file__).resolve().parents[3]
  17. MODELS = REPO / "backend" / "app" / "models"
  18. def _declared_tables() -> set[str]:
  19. tables = set()
  20. for path in MODELS.glob("*.py"):
  21. source = path.read_text()
  22. tables.update(re.findall(r'__tablename__\s*=\s*"(\w+)"', source))
  23. tables.update(re.findall(r'Table\(\s*"(\w+)"', source))
  24. return tables
  25. @pytest.mark.integration
  26. @pytest.mark.sqlite_only # inspects the SQLite file the started process creates
  27. def test_a_fresh_start_creates_every_model_table(tmp_path):
  28. db_path = tmp_path / "bambuddy.db"
  29. env = {
  30. **os.environ,
  31. "DATA_DIR": str(tmp_path),
  32. "LOG_DIR": str(tmp_path / "logs"),
  33. "DATABASE_URL": f"sqlite+aiosqlite:///{db_path}",
  34. }
  35. result = subprocess.run(
  36. [
  37. sys.executable,
  38. "-c",
  39. "import asyncio\n"
  40. "from backend.app.core.database import init_db, engine\n"
  41. "async def main():\n"
  42. " await init_db()\n"
  43. " await engine.dispose()\n"
  44. "asyncio.run(main())\n",
  45. ],
  46. cwd=REPO,
  47. env=env,
  48. capture_output=True,
  49. text=True,
  50. timeout=300,
  51. )
  52. assert result.returncode == 0, result.stderr[-3000:]
  53. with sqlite3.connect(db_path) as db:
  54. created = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
  55. declared = _declared_tables()
  56. assert "hms_notified_faults" in declared
  57. assert declared - created == set()