import gzip import json from pathlib import Path import struct import tempfile import unittest from verify_front_values import canonical, compare, ContractError def col(name, kind): return {'name': name, 'data_type': kind} class CanonicalTests(unittest.TestCase): def test_decimal_format_equality(self): c = col('v', 'decimal') self.assertEqual(canonical('001.2300', c), canonical('1.23e0', c)) def test_cent_difference_is_not_tolerated(self): c = col('v', 'decimal') self.assertNotEqual(canonical('1.23', c), canonical('1.24', c)) def test_large_decimal_not_context_rounded(self): c = col('v', 'decimal') self.assertNotEqual(canonical('123456789012345678901234567890.01', c), canonical('123456789012345678901234567890.02', c)) def test_unsigned_bigint_exact(self): c = col('v', 'bigint') self.assertNotEqual(canonical('18446744073709551615', c), canonical('18446744073709551614', c)) def test_nonintegral_integer_rejected(self): with self.assertRaises(ContractError): canonical('1.1', col('v', 'int')) def test_null_empty_distinct(self): self.assertNotEqual(canonical(None, col('v', 'text')), canonical('', col('v', 'text'))) def test_unicode_exact(self): c = col('v', 'text') self.assertEqual(canonical('中文😀', c), ('text', '中文😀')) self.assertNotEqual(canonical('é', c), canonical('e\u0301', c)) def test_char_mysql_trailing_spaces(self): self.assertEqual(canonical('中文 ', col('v', 'char')), canonical('中文', col('v', 'char'))) self.assertNotEqual(canonical('中文\t', col('v', 'char')), canonical('中文', col('v', 'char'))) def test_varchar_trailing_spaces_preserved(self): self.assertNotEqual(canonical('x ', col('v', 'varchar')), canonical('x', col('v', 'varchar'))) def test_float32_equivalent_text(self): self.assertEqual(canonical('1.2', col('v', 'float')), canonical('1.2000000476837158', col('v', 'float'))) def test_adjacent_float32_bits_differ(self): self.assertNotEqual(canonical('1', col('v', 'float')), canonical('1.0000001192092896', col('v', 'float'))) def test_adjacent_float64_bits_differ(self): self.assertNotEqual(canonical('1', col('v', 'double')), canonical('1.0000000000000002', col('v', 'double'))) def test_signed_float_zero_preserved(self): self.assertNotEqual(canonical('-0', col('v', 'float')), canonical('0', col('v', 'float'))) def test_nonfinite_and_overflow_rejected(self): for value in ['NaN', 'Infinity', '1e999', '3.5e38']: with self.subTest(value=value), self.assertRaises(ContractError): canonical(value, col('v', 'float')) def test_temporal_not_silently_normalized(self): self.assertNotEqual(canonical('0000-00-00', col('v', 'date')), canonical('0001-01-01', col('v', 'date'))) class FileTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.source = self.root / 'source' self.target = self.root / 'target' self.source.mkdir() self.target.mkdir() self.columns = [col('id', 'int'), col('price', 'decimal'), col('note', 'text')] self.source_rows = [['1', '1.20', '中文😀'], ['2', '1.21', None]] self.target_rows = [list(row) for row in self.source_rows] self.floats = {} self.deferred = [] self.extra_manifest = {} def tearDown(self): self.tmp.cleanup() def run_fixture(self, declared_target=None): def write_rows(file, rows, hex_source=False): with gzip.open(file, 'wt', encoding='utf-8') as f: for row in rows: if hex_source: row = [v.encode('utf-8').hex().upper() if v is not None else None for v in row] f.write(json.dumps(row, ensure_ascii=False) + '\n') baseline = {'db': 'synthetic', 'tables': 1, 'rows': len(self.source_rows), 'metadata': {'t': self.columns}, 'counts': {'t': len(self.source_rows)}} write_rows(self.source / 't.hex.jsonl.gz', self.source_rows, True) write_rows(self.target / 't.jsonl.gz', self.target_rows) counts = {} if self.deferred else {'t': len(self.target_rows) if declared_target is None else declared_target} manifest = {'schema': 'SYNTHETIC', 'build': 'synthetic', 'counts': counts, 'deferred_tables': self.deferred} manifest.update(self.extra_manifest) (self.source / 'baseline.json').write_text(json.dumps(baseline)) (self.root / 'float.json').write_text(json.dumps(self.floats)) (self.root / 'export-manifest.json').write_text(json.dumps(manifest)) return compare(self.source / 'baseline.json', self.root / 'float.json', self.target, self.root / 'export-manifest.json') def test_complete_equality_row_order_independent(self): self.target_rows.reverse() result = self.run_fixture() self.assertTrue(result['equal']) self.assertEqual(result['status'], 'complete') def test_missing_row(self): self.target_rows.pop() result = self.run_fixture() self.assertFalse(result['equal']) self.assertEqual(result['tables'][0]['missing_or_changed_rows'], 1) def test_duplicate_cannot_replace_different_row_same_total(self): self.target_rows[1] = self.target_rows[0] result = self.run_fixture() self.assertFalse(result['equal']) self.assertEqual(result['tables'][0]['extra_or_changed_rows'], 1) self.assertEqual(result['tables'][0]['missing_or_changed_rows'], 1) def test_extra_duplicate_detected(self): self.target_rows.append(self.target_rows[0]) self.assertFalse(self.run_fixture()['equal']) def test_wrong_export_count_fails(self): self.assertFalse(self.run_fixture(declared_target=99)['equal']) def test_null_empty_mismatch_full_pipeline(self): self.target_rows[1][2] = '' self.assertFalse(self.run_fixture()['equal']) def test_decimal_cent_mismatch_full_pipeline(self): self.target_rows[0][1] = '1.21' self.assertFalse(self.run_fixture()['equal']) def test_precise_float_sidecar_overrides_lossy_display(self): self.columns = [col('id', 'int'), col('v', 'float')] self.source_rows = [['1', '1.00']] self.target_rows = [['1', '1.0000001192092896']] self.floats = {'t': {'1': {'v': '1.0000001192092896'}}} self.assertTrue(self.run_fixture()['equal']) self.target_rows = [['1', '1.00']] self.assertFalse(self.run_fixture()['equal']) def test_missing_float_sidecar_fails_closed(self): self.columns = [col('id', 'int'), col('v', 'float')] self.source_rows = [['1', '1.00']] self.target_rows = [['1', '1.00']] with self.assertRaises(ContractError): self.run_fixture() def test_explicit_deferred_never_full_pass(self): self.deferred = ['t'] result = self.run_fixture() self.assertFalse(result['equal']) self.assertTrue(result['verified_subset_equal']) self.assertEqual(result['status'], 'partial') self.assertEqual(result['verified_tables'], 0) def test_object_deferred_never_full_pass(self): self.deferred = [{'table': 't', 'rows': 2, 'zero_date_cells': 1}] result = self.run_fixture() self.assertEqual(result['status'], 'partial') self.assertFalse(result['equal']) self.assertEqual(result['deferred_tables'], ['t']) def test_object_deferred_wrong_count_rejected(self): self.deferred = [{'table': 't', 'rows': 999, 'zero_date_cells': 1}] with self.assertRaises(ContractError): self.run_fixture() def test_object_deferred_missing_count_rejected(self): self.deferred = [{'table': 't'}] with self.assertRaises(ContractError): self.run_fixture() def test_duplicate_mixed_deferred_rejected(self): self.deferred = ['t', {'table': 't', 'rows': 2}] with self.assertRaises(ContractError): self.run_fixture() def test_negative_deferred_zero_cells_rejected(self): self.deferred = [{'table': 't', 'rows': 2, 'zero_date_cells': -1}] with self.assertRaises(ContractError): self.run_fixture() def test_partial_flag_without_deferred_never_full_pass(self): self.extra_manifest['partial_rehearsal_only'] = True self.assertEqual(self.run_fixture()['status'], 'partial') def test_export_error_rejected(self): self.extra_manifest['errors'] = ['synthetic failure'] with self.assertRaises(ContractError): self.run_fixture() def test_declared_column_order_must_match(self): self.extra_manifest['columns'] = {'t': ['price', 'id', 'note']} with self.assertRaises(ContractError): self.run_fixture() if __name__ == '__main__': unittest.main(verbosity=2)