import sqlite3,tempfile,unittest
from pathlib import Path
from rehearse import *
class MigrationTests(unittest.TestCase):
 def setUp(self):
  self.t=tempfile.TemporaryDirectory();self.path=Path(self.t.name)/'test.sqlite';create_fixture(self.path,103);self.db=connect(self.path)
 def tearDown(self):self.db.close();self.t.cleanup()
 def test_complete_rollout_preserves_data(self):
  before=content_digest(self.db);expand(self.db);self.assertTrue(contracts(self.db)['consistent']);backfill(self.db,20);cutover(self.db);contract(self.db);self.assertEqual(content_digest(self.db),before);self.assertTrue(contracts(self.db,False)['consistent']);self.assertNotIn('tag',[r['name'] for r in self.db.execute('PRAGMA table_info(assets)')])
 def test_resume_committed_batches_from_new_connection(self):
  expand(self.db)
  with self.assertRaises(InterruptedBackfill):backfill(self.db,20,2)
  self.assertEqual(self.db.execute('SELECT COUNT(*) FROM assets WHERE asset_key IS NOT NULL').fetchone()[0],40)
  self.db.close();self.db=connect(self.path);r=backfill(self.db,20);self.assertEqual(r['updated_rows'],63);self.assertTrue(contracts(self.db)['consistent'])
 def test_repeated_backfill_is_noop(self):
  expand(self.db);backfill(self.db,20);self.assertEqual(backfill(self.db,20)['updated_rows'],0)
 def test_legacy_write_during_overlap_updates_new_key(self):
  expand(self.db);self.db.execute("UPDATE assets SET tag=' AHU-NEW ' WHERE id=1");self.db.execute("INSERT INTO assets VALUES (999,' AHU-999 ','New fixture',NULL)");self.db.commit();backfill(self.db);self.assertTrue(contracts(self.db)['consistent']);self.assertEqual(self.db.execute('SELECT asset_key FROM assets WHERE id=1').fetchone()[0],'ahu-new')
 def test_old_app_blocks_contraction(self):
  expand(self.db);backfill(self.db)
  with self.assertRaises(MigrationError):contract(self.db)
  self.assertEqual(metadata(self.db,'stage'),'backfilled')
 def test_duplicate_constraint_failure(self):
  self.db.execute('UPDATE assets SET tag=(SELECT tag FROM assets WHERE id=1) WHERE id=2');self.db.commit();expand(self.db)
  with self.assertRaises(sqlite3.IntegrityError):backfill(self.db)
  self.assertEqual(metadata(self.db,'stage'),'expanded');self.assertEqual(len(profile(self.db)['duplicate_keys']),1)
 def test_null_source_fails_without_partial_batch(self):
  self.db.execute('UPDATE assets SET tag=NULL WHERE id=2');self.db.commit();expand(self.db)
  with self.assertRaises(MigrationError):backfill(self.db)
  self.assertEqual(self.db.execute('SELECT COUNT(*) FROM assets WHERE asset_key IS NOT NULL').fetchone()[0],0)
 def test_empty_source_rejected(self):
  self.db.execute("UPDATE assets SET tag=' ' WHERE id=2");self.db.commit();expand(self.db)
  with self.assertRaises(MigrationError):backfill(self.db)
 def test_same_named_wrong_index_rejected(self):
  expand(self.db);self.db.execute('CREATE INDEX assets_key_unique ON assets(description)');self.db.commit()
  with self.assertRaises(MigrationError):backfill(self.db)
  self.assertEqual(metadata(self.db,'stage'),'expanded')
 def test_failed_ddl_rolls_back(self):
  before=schema(self.db)
  def bad():self.db.execute('ALTER TABLE assets ADD COLUMN bogus TEXT');self.db.execute('SELECT * FROM nonexistent')
  with self.assertRaises(sqlite3.OperationalError):atomic(self.db,bad)
  self.assertEqual(schema(self.db),before)
 def test_lock_timeout_and_recovery(self):
  r=lock_probe(self.path);self.assertGreaterEqual(r['wait_seconds'],.05);self.assertLess(r['wait_seconds'],1);expand(self.db)
 def test_old_query_retired_after_contract(self):
  expand(self.db);backfill(self.db);cutover(self.db);contract(self.db)
  with self.assertRaises(MigrationError):contracts(self.db,True)
 def test_contract_has_actual_unique_not_null_constraints(self):
  expand(self.db);backfill(self.db);cutover(self.db);contract(self.db)
  for val in ['ahu-000001',None,'']:
   with self.subTest(value=val),self.assertRaises(sqlite3.IntegrityError):self.db.execute('INSERT INTO assets VALUES (999,?,?)',(val,'bad'))
   self.db.rollback()
 def test_runner_reports_invalid_fixture_and_cleans_temp(self):
  r=rehearse(Path(self.t.name)/'report',100,20,2,'duplicate');self.assertFalse(r['passed']);self.assertIn('UNIQUE constraint',r['error']);self.assertTrue(r['temporary_directory_removed_after_run'])
if __name__=='__main__':unittest.main()