import copy import gzip import json from pathlib import Path import tempfile import unittest from verify_approved_normalization import verify, InvalidEvidence, NULLABLE_FIELDS class ApprovedChangeTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.before = self.root / 'before' self.after = self.root / 'after' self.before.mkdir() self.after.mkdir() def c(name, kind='varchar', nullable='NO', extra=''): return {'name': name, 'data_type': kind, 'type': kind, 'nullable': nullable, 'extra': extra, 'default': None, 'comment': '', 'charset': '', 'collation': ''} self.meta = { 'ydy_menu': [c('id', 'int'), c('create_time', 'datetime'), c('note')], 'ydy_metabolism_pal': [c('id', 'int'), c('turnover_time', 'date')], 'ydy_station': [c('id', 'int'), c('create_time', 'datetime')], 'ydy_user': [c('id', 'int'), c('uuid'), c('useful_start', 'datetime', 'YES'), c('useful_end', 'datetime', 'YES'), c('status', 'int')], 'ydy_restaurants': [c('id', 'int'), c('uuid')], 'ydy_user_data': [c('id', 'int', extra='auto_increment'), c('user_uuid'), c('data_uuid'), c('staff_uuid_str', nullable='YES'), c('type', 'int')], 'role_data': [c('id', 'int', extra='auto_increment'), c('permission')], } self.rows = { 'ydy_menu': [['1', '0000-00-00 00:00:00', 'δΈ­ζ–‡πŸ˜€']], 'ydy_metabolism_pal': [['1', '0000-00-00']], 'ydy_station': [['1', '0000-00-00 00:00:00']], 'ydy_user': [['1', 'owner-a', '0000-00-00 00:00:00', '0000-00-00 00:00:00', '1']], 'ydy_restaurants': [['1', 'restaurant-a']], 'ydy_user_data': [], 'role_data': [['1', 'unchanged']], } self.after_meta = copy.deepcopy(self.meta) self.after_rows = copy.deepcopy(self.rows) cells = [] for table, names in [('ydy_menu', ['create_time']), ('ydy_metabolism_pal', ['turnover_time']), ('ydy_station', ['create_time']), ('ydy_user', ['useful_start', 'useful_end'])]: for name in names: index = [c['name'] for c in self.meta[table]].index(name) cells.append({'table': table, 'pk': {'id': '1'}, 'column': name, 'before': self.rows[table][0][index], 'after': None}) self.after_rows[table][0][index] = None if (table, name) in NULLABLE_FIELDS: self.after_meta[table][index]['nullable'] = 'YES' row = {'id': '1', 'user_uuid': 'owner-a', 'data_uuid': 'restaurant-a', 'staff_uuid_str': None, 'type': '3'} self.after_rows['ydy_user_data'] = [[row[c['name']] for c in self.meta['ydy_user_data']]] self.plan = {'source_database': 'stage', 'target_database': 'stage', 'zero_date_cells': cells, 'nullable_changes': [{'table': t, 'column': c, 'before': 'NO', 'after': 'YES'} for t, c in sorted(NULLABLE_FIELDS)], 'permission_insert': {'table': 'ydy_user_data', 'row': row}, 'auto_increment_change': {'table': 'ydy_user_data', 'before': '1', 'after': '2'}} self.before_seeds = {'ydy_user_data': 1, 'role_data': 5} self.after_seeds = {'ydy_user_data': 2, 'role_data': 5} self.after_ddl_suffix = '' def tearDown(self): self.temp.cleanup() def write_snapshot(self, folder, metadata, rows, seeds): for table, columns in metadata.items(): with gzip.open(folder / (table + '.hex.jsonl.gz'), 'wt', encoding='utf-8') as f: for row in rows[table]: f.write(json.dumps([v.encode('utf-8').hex() if v is not None else None for v in row]) + '\n') ddl = 'CREATE TABLE `' + table + '` (\n' for c in columns: ddl += ' `' + c['name'] + '` ' + c['type'] + (' NOT NULL' if c['nullable'] == 'NO' else ' DEFAULT NULL') + ',\n' ddl += ' PRIMARY KEY (`id`)\n) ENGINE=InnoDB ' if table in seeds: ddl += 'AUTO_INCREMENT=' + str(seeds[table]) + ' ' ddl += 'DEFAULT CHARSET=utf8mb4;' if folder == self.after and table == 'role_data': ddl += self.after_ddl_suffix (folder / (table + '.ddl.txt')).write_text(ddl) baseline = {'db': 'stage', 'tables': len(metadata), 'rows': sum(len(v) for v in rows.values()), 'counts': {t: len(v) for t, v in rows.items()}, 'metadata': metadata} (folder / 'baseline.json').write_text(json.dumps(baseline)) def run_case(self): self.write_snapshot(self.before, self.meta, self.rows, self.before_seeds) self.write_snapshot(self.after, self.after_meta, self.after_rows, self.after_seeds) (self.root / 'approval.json').write_text(json.dumps(self.plan)) return verify(self.before / 'baseline.json', self.after / 'baseline.json', self.root / 'approval.json') def test_only_approved_changes_pass(self): result = self.run_case() self.assertTrue(result['verified']) self.assertEqual(result['approved_zero_date_cells'], 5) self.assertEqual(result['rows_after'], result['rows_before'] + 1) def test_unapproved_text_change_rejected(self): self.after_rows['ydy_menu'][0][2] = 'changed' with self.assertRaises(InvalidEvidence): self.run_case() def test_unapproved_existing_permission_change_rejected(self): self.after_rows['role_data'][0][1] = 'all' with self.assertRaises(InvalidEvidence): self.run_case() def test_unapproved_status_change_rejected(self): self.after_rows['ydy_user'][0][-1] = '0' with self.assertRaises(InvalidEvidence): self.run_case() def test_missing_row_rejected(self): self.after_rows['role_data'] = [] with self.assertRaises(InvalidEvidence): self.run_case() def test_duplicate_primary_key_rejected(self): self.after_rows['role_data'].append(self.after_rows['role_data'][0]) with self.assertRaises(InvalidEvidence): self.run_case() def test_extra_permission_row_rejected(self): self.after_rows['ydy_user_data'].append(['2', 'owner-a', 'restaurant-other', None, '3']) with self.assertRaises(InvalidEvidence): self.run_case() def test_wrong_restaurant_reference_rejected(self): self.plan['permission_insert']['row']['data_uuid'] = 'restaurant-other' self.after_rows['ydy_user_data'][0][2] = 'restaurant-other' with self.assertRaises(InvalidEvidence): self.run_case() def test_wrong_owner_reference_rejected(self): self.plan['permission_insert']['row']['user_uuid'] = 'owner-other' self.after_rows['ydy_user_data'][0][1] = 'owner-other' with self.assertRaises(InvalidEvidence): self.run_case() def test_wildcard_permission_type_rejected(self): self.plan['permission_insert']['row']['type'] = '1' self.after_rows['ydy_user_data'][0][-1] = '1' with self.assertRaises(InvalidEvidence): self.run_case() def test_extra_nullable_change_rejected(self): self.after_meta['role_data'][1]['nullable'] = 'YES' with self.assertRaises(InvalidEvidence): self.run_case() def test_other_default_change_rejected(self): self.after_meta['role_data'][1]['default'] = 'all' with self.assertRaises(InvalidEvidence): self.run_case() def test_unapproved_counter_change_rejected(self): self.after_seeds['role_data'] = 6 with self.assertRaises(InvalidEvidence): self.run_case() def test_permission_counter_wrong_increment_rejected(self): self.plan['auto_increment_change']['after'] = '3' self.after_seeds['ydy_user_data'] = 3 with self.assertRaises(InvalidEvidence): self.run_case() def test_wrong_original_zero_cell_rejected(self): self.plan['zero_date_cells'][0]['before'] = '0000-00-00' with self.assertRaises(InvalidEvidence): self.run_case() def test_duplicate_approved_cell_rejected(self): self.plan['zero_date_cells'].append(self.plan['zero_date_cells'][0]) with self.assertRaises(InvalidEvidence): self.run_case() def test_unlisted_null_change_rejected(self): self.plan['zero_date_cells'].pop() with self.assertRaises(InvalidEvidence): self.run_case() def test_remaining_zero_even_if_omitted_from_plan_rejected(self): omitted = self.plan['zero_date_cells'].pop() self.after_rows['ydy_user'][0][3] = omitted['before'] with self.assertRaises(InvalidEvidence): self.run_case() def test_unapproved_ddl_change_rejected(self): self.after_ddl_suffix = '\nALTER TABLE role_data DROP INDEX x;' with self.assertRaises(InvalidEvidence): self.run_case() if __name__ == '__main__': unittest.main(verbosity=2)