shanxia2.0/backend/tests/test_migration_clear_modes_disabled.py

"""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