tests/test_sql_fixture_checker.py

import importlib.util
from pathlib import Path
import unittest

MODULE_PATH = Path(__file__).resolve().parents[1] / 'scripts' / 'sql_fixture_checker.py'
spec = importlib.util.spec_from_file_location('checker', MODULE_PATH)
checker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(checker)

class SQLFixtureTests(unittest.TestCase):
    def setUp(self): self.conn=checker.fixture_connection()
    def tearDown(self): self.conn.close()
    def test_fixture_order_count(self): self.assertEqual(self.conn.execute('SELECT COUNT(*) FROM orders').fetchone()[0],3)
    def test_grain_join_rows(self): self.assertEqual(checker.join_fanout(self.conn)['joined_rows'],6)
    def test_join_fanout_source_sum(self): self.assertEqual(checker.join_fanout(self.conn)['source_total'],275.0)
    def test_join_fanout_wrong_sum(self): self.assertEqual(checker.join_fanout(self.conn)['incorrect_joined_total'],575.0)
    def test_correct_items_total(self): self.assertEqual(checker.join_fanout(self.conn)['correct_item_total'],275.0)
    def test_latest_status_tie_break(self): self.assertEqual(checker.latest_status(self.conn),[(1,'shipped'),(2,'paid')])
    def test_null_not_in_is_wrong(self): self.assertEqual(checker.null_filter(self.conn)['not_in'],[])
    def test_not_exists_handles_null(self): self.assertEqual(checker.null_filter(self.conn)['not_exists'],['B','C'])
    def test_half_open_ends_before_feb(self): self.assertEqual(checker.half_open_window(self.conn),[1,2])
    def test_cdc_deterministic_tie_break(self): self.assertEqual(checker.cdc_winners(self.conn),[('K1','winner'),('K2','only')])
    def test_simple_select_advisory_clean(self): self.assertEqual(checker.advisory_sql_flags('SELECT id FROM t'),[])
    def test_select_star_advisory(self): self.assertIn('select_all_projection_review', checker.advisory_sql_flags('SELECT * FROM users'))
    def test_delete_advisory(self): self.assertIn('potentially_mutating_or_privilege_related', checker.advisory_sql_flags('DELETE FROM users;'))
    def test_unqualified_delete_flag(self): self.assertIn('possible_unqualified_delete', checker.advisory_sql_flags('DELETE FROM users;'))
    def test_scoped_delete_does_not_trigger_unqualified(self): self.assertNotIn('possible_unqualified_delete', checker.advisory_sql_flags('DELETE FROM users WHERE id=1;'))
    def test_grant_is_mutating(self): self.assertIn('potentially_mutating_or_privilege_related', checker.advisory_sql_flags('GRANT SELECT ON tab TO role;'))
    def test_summary_declares_sqlite_only(self): self.assertEqual(checker.report()['fixture_type'],'sqlite_in_memory_only')

if __name__=='__main__': unittest.main()