185 lines
7.9 KiB
Python
185 lines
7.9 KiB
Python
"""Offline restoration regressions; extract functions, never import scripts or run main.
|
|
|
|
The historical filename remains so the session's blanket-disable tests are replaced
|
|
in place. Explicit migration clear modes retain their original dry-run behavior.
|
|
"""
|
|
import argparse
|
|
import ast
|
|
from collections import defaultdict
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / 'scripts'
|
|
CASES = [
|
|
('migrate_crm_product.py', 'run_migration', 'clear_target', 'clear_target_products'),
|
|
('migrate_crm_product_category.py', 'run_migration', 'clear_target', 'clear_target_categories'),
|
|
('migrate_dibang_data.py', 'migrate', 'truncate_target', 'truncate_target_tables'),
|
|
('migrate_bx_data.py', 'migrate_table', 'replace_workflow', 'clear_portal_workflow'),
|
|
]
|
|
|
|
|
|
def tree(filename):
|
|
return ast.parse((SCRIPTS / filename).read_text())
|
|
|
|
|
|
def function_node(filename, name):
|
|
return next(n for n in tree(filename).body if isinstance(n, ast.FunctionDef) and n.name == name)
|
|
|
|
|
|
def extract(filename, name, namespace):
|
|
module = ast.Module(body=[ast.ImportFrom(module='__future__', names=[ast.alias(name='annotations')], level=0), function_node(filename, name)], type_ignores=[])
|
|
exec(compile(ast.fix_missing_locations(module), filename, 'exec'), namespace)
|
|
return namespace[name]
|
|
|
|
|
|
class FakeEngine:
|
|
"""Records transaction/SQL intent without creating any database engine."""
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.transactions = 0
|
|
|
|
def begin(self):
|
|
self.transactions += 1
|
|
return self
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
self.statements.append((sql, params))
|
|
return SimpleNamespace(rowcount=2)
|
|
|
|
|
|
class Stats:
|
|
def __init__(self):
|
|
self.source_counts = {}
|
|
self.inserted = {}
|
|
self.steps = defaultdict(lambda: SimpleNamespace(source_count=0, inserted=0, skipped=0))
|
|
|
|
def step(self, table):
|
|
return self.steps[table]
|
|
|
|
|
|
def namespace_for(filename, helper):
|
|
keys = ('purchase_plans', 'inbound_outbound', 'weighing_slips')
|
|
namespace = dict(
|
|
text=lambda sql: sql, TARGET_TABLE='crm_test',
|
|
TARGET_TABLES={key: 'db_' + key for key in keys},
|
|
SOURCE_TABLES={key: key for key in keys},
|
|
TABLE_OPTIONS={}, MigrationStats=Stats,
|
|
load_user_mapping=lambda _: ({}, 'admin'),
|
|
load_bx_user_mapping=lambda _: ({}, 'admin'),
|
|
fetch_source_rows=lambda *a, **kw: [], fetch_rows=lambda *a: [],
|
|
topo_sort_categories=lambda rows: rows,
|
|
build_id_maps=lambda rows: {key: {} for key in rows},
|
|
print_report=lambda *a, **kw: None,
|
|
)
|
|
extract(filename, helper, namespace)
|
|
return namespace
|
|
|
|
|
|
@pytest.mark.parametrize('filename,name,flag,helper', CASES)
|
|
@pytest.mark.parametrize('dry_run', [False, True])
|
|
@pytest.mark.parametrize('clear', [False, True])
|
|
def test_original_clear_mode_and_dry_run(filename, name, flag, helper, dry_run, clear):
|
|
import inspect
|
|
namespace = namespace_for(filename, helper)
|
|
fn = extract(filename, name, namespace)
|
|
kwargs = {key: None for key in inspect.signature(fn).parameters}
|
|
target = FakeEngine()
|
|
kwargs.update(source_engine=FakeEngine(), target_engine=target, tenant_no='tenant', dry_run=dry_run)
|
|
kwargs[flag] = clear
|
|
if name == 'migrate_table':
|
|
kwargs.update(table='bx_workflow_def', stats=Stats())
|
|
fn(**kwargs)
|
|
assert target.transactions == (0 if dry_run else 1)
|
|
expected = 0 if dry_run or not clear else (2 if filename == 'migrate_bx_data.py' else 3 if filename == 'migrate_dibang_data.py' else 1)
|
|
assert len(target.statements) == expected
|
|
|
|
|
|
@pytest.mark.parametrize('filename,name,flag,helper', CASES)
|
|
def test_original_clear_helpers_execute_expected_sql_offline(filename, name, flag, helper):
|
|
namespace = namespace_for(filename, helper)
|
|
target = FakeEngine()
|
|
fn = namespace[helper]
|
|
if filename == 'migrate_bx_data.py':
|
|
fn(target)
|
|
assert target.statements == [('DELETE FROM bx_workflow_node', None), ('DELETE FROM bx_workflow_def', None)]
|
|
else:
|
|
result = fn(target, 'tenant')
|
|
assert all('DELETE FROM' in sql for sql, _ in target.statements)
|
|
key = 'tenant_no' if filename == 'migrate_dibang_data.py' else 't'
|
|
assert all(params == {key: 'tenant'} for _, params in target.statements)
|
|
if key == 't':
|
|
assert result == 2
|
|
assert "tenant_no='' OR tenant_no IS NULL" in target.statements[0][0]
|
|
|
|
|
|
def cli_parser(filename):
|
|
"""Execute only argparse setup, stopping immediately after parse_args."""
|
|
node = function_node(filename, 'main')
|
|
statements = []
|
|
for statement in node.body:
|
|
statements.append(statement)
|
|
if isinstance(statement, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'args' for t in statement.targets):
|
|
break
|
|
else:
|
|
raise AssertionError('parse_args not found')
|
|
node.name = 'parse_only'
|
|
node.body = statements + [ast.Return(value=ast.Name(id='args', ctx=ast.Load()))]
|
|
namespace = dict(argparse=argparse, os=os, sys=sys, Path=Path, DEFAULT_TENANT_NO='t', BACKEND_ROOT=SCRIPTS.parent)
|
|
exec(compile(ast.fix_missing_locations(ast.Module(body=[node], type_ignores=[])), filename, 'exec'), namespace)
|
|
return namespace['parse_only']
|
|
|
|
|
|
@pytest.mark.parametrize('filename,flag,attribute', [
|
|
('migrate_crm_product.py', '--clear-target', 'clear_target'),
|
|
('migrate_crm_product_category.py', '--clear-target', 'clear_target'),
|
|
('migrate_dibang_data.py', '--truncate-target', 'truncate_target'),
|
|
])
|
|
def test_cli_preserves_explicit_clear_option(filename, flag, attribute, monkeypatch):
|
|
monkeypatch.setattr(sys, 'argv', ['offline-test', flag])
|
|
assert getattr(cli_parser(filename)(), attribute) is True
|
|
main = function_node(filename, 'main')
|
|
assert not any(isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == 'error' for n in ast.walk(main))
|
|
|
|
|
|
def test_dibang_cli_keeps_dry_run_clear_conflict_guard():
|
|
main = function_node('migrate_dibang_data.py', 'main')
|
|
guards = [n for n in main.body if isinstance(n, ast.If) and 'args.truncate_target' in ast.unparse(n.test) and 'args.dry_run' in ast.unparse(n.test)]
|
|
assert len(guards) == 1
|
|
assert any(isinstance(n, ast.Return) and isinstance(n.value, ast.Constant) and n.value.value == 2 for n in ast.walk(guards[0]))
|
|
|
|
|
|
@pytest.mark.parametrize('legacy_flag', [False, True])
|
|
def test_bx_cli_preserves_original_replace_default_and_opt_out(legacy_flag, monkeypatch):
|
|
monkeypatch.setattr(sys, 'argv', ['offline-test'] + (['--no-replace-workflow'] if legacy_flag else []))
|
|
args = cli_parser('migrate_bx_data.py')()
|
|
main = function_node('migrate_bx_data.py', 'main')
|
|
assignment = next(n for n in main.body if isinstance(n, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'replace_workflow' for t in n.targets))
|
|
assert eval(compile(ast.Expression(assignment.value), '<offline-default>', 'eval'), {'args': args}) is (not legacy_flag)
|
|
|
|
|
|
@pytest.mark.parametrize('dry_run', [False, True])
|
|
@pytest.mark.parametrize('replace', [False, True])
|
|
def test_bx_run_migration_forwards_replace_and_dry_run(dry_run, replace):
|
|
calls = []
|
|
namespace = dict(MigrationStats=Stats, load_bx_user_mapping=lambda _: ({}, 'admin'),
|
|
resolve_steps=lambda *a: ['bx_workflow_def'],
|
|
migrate_table=lambda *a, **kw: calls.append(kw) or [],
|
|
print_report=lambda *a: None)
|
|
fn = extract('migrate_bx_data.py', 'run_migration', namespace)
|
|
fn(FakeEngine(), FakeEngine(), tenant_no='tenant', user_mapping_path=None,
|
|
mapping_dir=None, dry_run=dry_run, phase='base', only_tables=None,
|
|
replace_workflow=replace)
|
|
assert len(calls) == 1
|
|
assert calls[0]['replace_workflow'] is replace
|
|
assert calls[0]['dry_run'] is dry_run
|