165 lines
8.9 KiB
Python
165 lines
8.9 KiB
Python
"""生产补齐脚本的离线 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()
|