"""生产补齐脚本的离线 SQL 契约：只解析文件，不导入应用、不连接数据库。 检查实际交付的 ALTER/INSERT 所描述的变更，不把解析通过当作 MySQL 执行通过。 删掉关联字段、改成非空、重建已有表或插入角色授权时，以下契约应失败。 存储过程前置检查的运行行为仍须在生产备份的隔离恢复实例验收。 """ import unittest from pathlib import Path import sqlglot from sqlglot import exp from sqlglot.tokens import Tokenizer, TokenType SQL_ROOT = Path(__file__).resolve().parents[2] / 'sql' PATCH = SQL_ROOT / 'patches/20261008_prod_existing_tunnel_repair_mysql.sql' def split_top_level(source: str, delimiter: TokenType) -> list[str]: """仅切分顶层语句/动作；字符串、注释及括号内的分隔符不作为边界。""" parts, start, depth = [], 0, 0 for token in Tokenizer(dialect='mysql').tokenize(source): if token.token_type == TokenType.L_PAREN: depth += 1 elif token.token_type == TokenType.R_PAREN: depth -= 1 elif token.token_type == delimiter and depth == 0: parts.append(source[start:token.start].strip()) start = token.end + 1 if source[start:].strip(): parts.append(source[start:].strip()) return parts def token_contract(source: str) -> list[tuple[TokenType, str]]: return [(token.token_type, token.text) for token in Tokenizer(dialect='mysql').tokenize(source)] class TestProdTunnelRepairSql(unittest.TestCase): def setUp(self) -> None: self.assertTrue(PATCH.is_file(), '缺少适配生产已存在隧道表的专用补丁') self.source = PATCH.read_text(encoding='utf-8') # 当前 SQLGlot 不完整支持 MODIFY ... NULL / DROP CHECK 的多动作 ALTER。 # 不把 Command 回退当作解析成功；整条用 token 契约比对，新列另做 AST 解析。 self.ddl = split_top_level(self.block('DDL'), TokenType.SEMICOLON) self.dml = sqlglot.parse(self.block('MENU'), read='mysql') def block(self, name: str) -> str: begin, end = f'-- BEGIN {name}', f'-- END {name}' self.assertEqual(self.source.count(begin), 1) self.assertEqual(self.source.count(end), 1) return self.source.split(begin, 1)[1].split(end, 1)[0] def test_only_expected_tables_are_altered_without_recreating_tunnel_or_dates(self) -> None: self.assertEqual(len(self.ddl), 5) heads = [Tokenizer(dialect='mysql').tokenize(stmt)[:3] for stmt in self.ddl] self.assertTrue(all([token.text.upper() for token in head[:2]] == ['ALTER', 'TABLE'] for head in heads)) self.assertEqual([head[2].text for head in heads], [ 'biz_usage_record', 'biz_return_item', 'biz_pallet_inventory', 'biz_pallet_event', 'biz_pallet_event', ]) def test_twelve_new_history_fields_are_nullable_and_have_expected_types(self) -> None: columns = {} for stmt in self.ddl: head = Tokenizer(dialect='mysql').tokenize(stmt)[:3] for action in split_top_level(stmt[head[2].end + 1:], TokenType.COMMA): tokens = Tokenizer(dialect='mysql').tokenize(action) if [token.text.upper() for token in tokens[:2]] != ['ADD', 'COLUMN']: continue end = next((token.start for token in tokens if token.token_type == TokenType.VAR and token.text.upper() == 'AFTER'), len(action)) # AFTER 的列位置不属于 CREATE TABLE 的列定义，先移除再解析类型/NULL。 definition = action[tokens[2].start:end].strip() parsed = sqlglot.parse_one(f'CREATE TABLE contract_column ({definition})', read='mysql') column = parsed.this.expressions[0] self.assertIsInstance(column, exp.ColumnDef) columns[(head[2].text, column.name)] = column.args['kind'].sql(dialect='mysql') self.assertFalse(any(isinstance(item.kind, exp.NotNullColumnConstraint) and not item.kind.args.get('allow_null') for item in column.args.get('constraints', []))) self.assertEqual(columns, { ('biz_usage_record', 'tunnel_id'): 'BIGINT', ('biz_usage_record', 'trace_no'): 'VARCHAR(64)', ('biz_usage_record', 'batch_no_snapshot'): 'VARCHAR(64)', ('biz_return_item', 'tunnel_id_snapshot'): 'BIGINT', ('biz_return_item', 'trace_no_snapshot'): 'VARCHAR(64)', ('biz_return_item', 'batch_no_snapshot'): 'VARCHAR(64)', ('biz_pallet_inventory', 'tunnel_id'): 'BIGINT', ('biz_pallet_inventory', 'trace_no'): 'VARCHAR(64)', ('biz_pallet_event', 'before_tunnel_id'): 'BIGINT', ('biz_pallet_event', 'after_tunnel_id'): 'BIGINT', ('biz_pallet_event', 'trace_no'): 'VARCHAR(64)', ('biz_pallet_event', 'batch_no_snapshot'): 'VARCHAR(64)', }) def test_alters_keep_the_original_reviewed_tunnel_semantics(self) -> None: original = split_top_level((SQL_ROOT / 'patches/20261005_tunnel_material_flow_mysql.sql').read_text(), TokenType.SEMICOLON) original_alters = [stmt for stmt in original if token_contract(stmt) and token_contract(stmt)[0][1].upper() == 'ALTER'] self.assertEqual([token_contract(stmt) for stmt in self.ddl], [token_contract(stmt) for stmt in original_alters]) def test_menu_rows_add_only_expected_permissions_without_role_grants(self) -> None: self.assertEqual(len(self.dml), 1) insert = self.dml[0] self.assertIsInstance(insert, exp.Insert) self.assertEqual(insert.this.this.name, 'sys_menu') columns = [column.name for column in insert.this.expressions] self.assertIsInstance(insert.expression, exp.Values) actual = [] for row in insert.expression.expressions: self.assertEqual(len(row.expressions), len(columns)) values = dict(zip(columns, row.expressions, strict=True)) actual.append((int(values['menu_id'].this), int(values['parent_id'].this), values['perms'].this, values['menu_type'].this)) self.assertEqual(actual, [ (520, 121, 'master:tunnel:list', 'C'), (1116, 520, 'master:tunnel:query', 'F'), (1117, 520, 'master:tunnel:add', 'F'), (1118, 520, 'master:tunnel:edit', 'F'), ]) def test_baseline_reads_seven_tables_without_new_columns_or_writes(self) -> None: source = (SQL_ROOT / 'checks/20261008_prod_material_flow_baseline.sql').read_text() statements = sqlglot.parse(source, read='mysql') expected = { 'biz_tunnel': ('tunnel_id', 13), 'biz_pallet_inventory': ('pallet_id', 11), 'biz_usage_record': ('usage_id', 13), 'biz_return_item': ('return_item_id', 8), 'biz_pallet_event': ('event_id', 21), 'biz_filling_record': ('filling_id', 24), 'biz_shipment_order': ('shipment_id', 15), } self.assertEqual(len(statements), len(expected)) for stmt in statements: self.assertIsInstance(stmt, exp.Select) table = next(stmt.find_all(exp.Table)).name primary_key, original_column_count = expected.pop(table) self.assertEqual(stmt.args['order'].expressions[0].this.name, primary_key) self.assertEqual(len(list(stmt.expressions[1].find_all(exp.Column))), original_column_count) self.assertIsNone(stmt.args.get('where'), '原数据基线不能过滤正式/测试数据或历史记录') self.assertFalse(expected) def test_postcheck_is_select_only_and_reports_expected_number_of_fields(self) -> None: source = (SQL_ROOT / 'checks/20261008_prod_material_flow_verify.sql').read_text() statements = sqlglot.parse(source, read='mysql') self.assertTrue(all(isinstance(stmt, exp.Select) for stmt in statements)) expected = statements[1].args['with_'].expressions[0] rows = list(expected.find_all(exp.Select)) self.assertEqual(len(rows), 25) actual = {(row.expressions[0].this.this if isinstance(row.expressions[0], exp.Alias) else row.expressions[0].this, row.expressions[1].this.this if isinstance(row.expressions[1], exp.Alias) else row.expressions[1].this) for row in rows} self.assertEqual(len(actual), 25, '验收字段不得重复以掩盖漏项') self.assertTrue({('biz_pallet_inventory', 'tunnel_id'), ('biz_pallet_inventory', 'site_arrived_at'), ('biz_usage_record', 'line_id'), ('biz_shipment_order', 'receipt_state'), ('biz_usage_record', 'fifo_skip_reason'), ('biz_pallet_event', 'fifo_recommendation')} <= actual) if __name__ == '__main__': unittest.main()