test_sqlite_fk.py 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  1. """core/sqlite_fk.py applies the selected ON DELETE rules that SQLite leaves out.
  2. The app's SQLite database runs with PRAGMA foreign_keys off, so a delete left
  3. every ON DELETE CASCADE / SET NULL undone: rows pointing at the deleted row kept
  4. its id, and SQLite hands the highest freed id to the next row created.
  5. """
  6. import pytest
  7. from sqlalchemy import delete, event, select, text
  8. from backend.app.core import sqlite_fk
  9. from backend.app.core.database import load_all_models
  10. from backend.app.models.archive import PrintArchive
  11. from backend.app.models.library import LibraryFile
  12. from backend.app.models.notification import NotificationProvider
  13. from backend.app.models.print_log import PrintLogEntry
  14. from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
  15. from backend.app.models.printer import Printer
  16. async def _archive_with_dependents(db):
  17. printer = Printer(name="P", serial_number="S1", ip_address="192.0.2.1", access_code="x")
  18. library_file = LibraryFile(filename="f.3mf", file_path="l/f.3mf", file_type="3mf", file_size=1)
  19. db.add_all([printer, library_file])
  20. await db.flush()
  21. archive = PrintArchive(
  22. printer_id=printer.id, filename="a.3mf", file_path="a/a.3mf", file_size=1, status="completed"
  23. )
  24. db.add(archive)
  25. await db.flush()
  26. item = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, position=1)
  27. db.add(item)
  28. await db.flush()
  29. variant = PrintQueueVariant(queue_item_id=item.id, library_file_id=library_file.id, target_model="X1C")
  30. run = PrintLogEntry(status="completed", print_name="a", archive_id=archive.id, queue_item_id=item.id)
  31. db.add_all([variant, run])
  32. await db.commit()
  33. return archive.id, item.id, variant.id, run.id
  34. async def _exists(db, model, row_id) -> bool:
  35. return (await db.execute(select(model.id).where(model.id == row_id))).scalar_one_or_none() is not None
  36. async def _run(db, run_id):
  37. statement = select(PrintLogEntry).where(PrintLogEntry.id == run_id).execution_options(populate_existing=True)
  38. return (await db.execute(statement)).scalar_one()
  39. async def _assert_rules_applied(db, item_id, variant_id, run_id):
  40. assert not await _exists(db, PrintQueueItem, item_id) # CASCADE
  41. assert not await _exists(db, PrintQueueVariant, variant_id) # CASCADE of the CASCADE
  42. run = await _run(db, run_id)
  43. assert run.archive_id is None # SET NULL
  44. assert run.queue_item_id is None # SET NULL of the cascaded row
  45. @pytest.mark.asyncio
  46. async def test_an_orm_delete_applies_the_rules(db_session):
  47. archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
  48. await db_session.delete(await db_session.get(PrintArchive, archive_id))
  49. await db_session.commit()
  50. await _assert_rules_applied(db_session, item_id, variant_id, run_id)
  51. @pytest.mark.asyncio
  52. async def test_a_bulk_delete_applies_the_rules(db_session):
  53. archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
  54. await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
  55. await db_session.commit()
  56. await _assert_rules_applied(db_session, item_id, variant_id, run_id)
  57. @pytest.mark.asyncio
  58. async def test_a_bulk_delete_sees_changes_not_flushed_yet(db_session):
  59. """A queue item moved off the archive in the same transaction must survive."""
  60. archive_id, item_id, _, _ = await _archive_with_dependents(db_session)
  61. other = PrintArchive(filename="b.3mf", file_path="a/b.3mf", file_size=1, status="completed")
  62. db_session.add(other)
  63. await db_session.commit()
  64. item = await db_session.get(PrintQueueItem, item_id)
  65. item.archive_id = other.id
  66. await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
  67. await db_session.commit()
  68. assert await _exists(db_session, PrintQueueItem, item_id)
  69. @pytest.mark.asyncio
  70. async def test_references_not_on_the_list_are_left_alone(db_session):
  71. """Not every model rule is right: deleting a printer must not widen a
  72. printer-scoped notification provider (SET NULL means all printers)."""
  73. printer = Printer(name="P", serial_number="S2", ip_address="192.0.2.2", access_code="x")
  74. db_session.add(printer)
  75. await db_session.flush()
  76. provider = NotificationProvider(name="n", provider_type="webhook", config="{}", printer_id=printer.id)
  77. db_session.add(provider)
  78. await db_session.commit()
  79. printer_id, provider_id = printer.id, provider.id
  80. await db_session.execute(delete(Printer).where(Printer.id == printer_id))
  81. await db_session.commit()
  82. scoped = (
  83. await db_session.execute(select(NotificationProvider.printer_id).where(NotificationProvider.id == provider_id))
  84. ).scalar_one()
  85. assert scoped == printer_id
  86. @pytest.mark.asyncio
  87. async def test_many_orm_deletes_in_one_flush_cost_one_statement_per_rule(db_session):
  88. """Applied once per flush, not once per row: a folder of thousands of files
  89. or a big purge must not run the rules per row while holding the write lock."""
  90. archives = [PrintArchive(filename=f"{i}.3mf", file_path=f"a/{i}.3mf", file_size=1, status="x") for i in range(50)]
  91. db_session.add_all(archives)
  92. await db_session.commit()
  93. rule_statements: list[str] = []
  94. sync_engine = db_session.bind.sync_engine
  95. def count(conn, cursor, statement, *args):
  96. head = statement.lstrip().upper()
  97. if head.startswith(("UPDATE", "DELETE")) and not head.startswith("DELETE FROM PRINT_ARCHIVES"):
  98. rule_statements.append(statement)
  99. event.listen(sync_engine, "before_cursor_execute", count)
  100. try:
  101. for archive in archives:
  102. await db_session.delete(archive)
  103. await db_session.commit()
  104. finally:
  105. event.remove(sync_engine, "before_cursor_execute", count)
  106. # The rules of print_archives and of the rows they cascade to, once each.
  107. assert len(rule_statements) < 30, len(rule_statements)
  108. @pytest.mark.asyncio
  109. async def test_a_rollback_drops_ids_collected_for_a_flush_that_failed(db_session):
  110. """Ids collected by a flush that then failed must not be applied by the next one."""
  111. await db_session.execute(select(1)) # an open transaction, as during a failed flush
  112. db_session.sync_session.info[sqlite_fk._PENDING] = {"print_archives": [1]}
  113. await db_session.rollback()
  114. assert sqlite_fk._PENDING not in db_session.sync_session.info
  115. @pytest.mark.asyncio
  116. async def test_nothing_to_do_on_postgresql(db_session, monkeypatch):
  117. """PostgreSQL applies the rules itself."""
  118. archive_id, item_id, _, run_id = await _archive_with_dependents(db_session)
  119. monkeypatch.setattr(sqlite_fk, "_applies", lambda connection: False)
  120. await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
  121. await db_session.commit()
  122. assert await _exists(db_session, PrintQueueItem, item_id)
  123. assert (await _run(db_session, run_id)).archive_id == archive_id
  124. @pytest.mark.asyncio
  125. async def test_the_startup_sweep_repairs_rows_whose_parent_is_already_gone(db_session):
  126. archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
  127. # Deleted the old way (raw SQL bypasses the hooks), leaving every dependent behind.
  128. await db_session.execute(text("DELETE FROM print_archives WHERE id = :id"), {"id": archive_id})
  129. await db_session.commit()
  130. await sqlite_fk.sweep_orphans(await db_session.connection())
  131. await db_session.commit()
  132. await _assert_rules_applied(db_session, item_id, variant_id, run_id)
  133. @pytest.mark.asyncio
  134. async def test_the_sweep_leaves_valid_references_and_unlisted_ones_alone(db_session):
  135. archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
  136. provider = NotificationProvider(name="n", provider_type="webhook", config="{}", printer_id=987654)
  137. db_session.add(provider)
  138. await db_session.commit()
  139. provider_id = provider.id
  140. await sqlite_fk.sweep_orphans(await db_session.connection())
  141. await db_session.commit()
  142. assert await _exists(db_session, PrintQueueItem, item_id)
  143. assert await _exists(db_session, PrintQueueVariant, variant_id)
  144. assert (await _run(db_session, run_id)).archive_id == archive_id
  145. stays = (
  146. await db_session.execute(select(NotificationProvider.printer_id).where(NotificationProvider.id == provider_id))
  147. ).scalar_one()
  148. assert stays == 987654
  149. def test_every_listed_reference_is_a_model_foreign_key_with_a_rule():
  150. """A typo or a renamed column must not silently drop a rule."""
  151. metadata = load_all_models()
  152. with_rule = {
  153. (table.name, fk.parent.name)
  154. for table in metadata.sorted_tables
  155. for fk in table.foreign_keys
  156. if fk.ondelete in ("CASCADE", "SET NULL")
  157. }
  158. assert sqlite_fk.APPLIED - with_rule == set()
  159. assert len(sqlite_fk._applied_foreign_keys()) == len(sqlite_fk.APPLIED)