`test_executor_progress.py`

56 lines
"""Exercise the real executor main flow with remote/budget boundaries replaced."""
import argparse
import ast
import json
import os
import pathlib
import sys
import tempfile
import time
import unittest
from unittest.mock import patch
from worklist_progress import split_work, summarize


class ExecutorFlowTests(unittest.TestCase):
    def run_main(self, work, audit, display=False):
        events = []
        with tempfile.TemporaryDirectory() as directory:
            result = os.path.join(directory, 'result.json')
            with open(result, 'w') as handle:
                json.dump(audit, handle)
            def process(row, results, dry_run=False):
                events.append(('process', row['shopify_id'], row['fix_type'], dry_run))
                results.append({'product_id': row['shopify_id'].rsplit('/', 1)[-1],
                                'fix': row['fix_type'], 'status': 'dryrun'})
                return 0
            namespace = {'argparse': argparse, 'os': os, 'json': json, 'time': time,
                         'OUTDIR': directory, 'RESULT': result, 'DOMAIN': 'test-store', 'API': 'test',
                         'load_worklist': lambda: work, 'split_work': split_work, 'summarize': summarize,
                         'process': process, 'DailyVariantLimit': type('DailyVariantLimit', (BaseException,), {}),
                         'backfill_display_variant': lambda cap, dry_run: events.append(('display', cap, dry_run)),
                         'budget_take': lambda *args: self.fail('dry replay requested a live budget')}
            module = ast.parse(pathlib.Path('bulk-fivefield-exec.py').read_text())
            main = next(node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == 'main')
            exec(compile(ast.Module(body=[main], type_ignores=[]), 'bulk-fivefield-exec.py', 'exec'), namespace)
            with patch.object(sys, 'argv', ['executor', '--dry-run', '--max', '10']), patch.dict(
                    os.environ, {'DISPLAY_VARIANT_BACKFILL': '1' if display else '0'}):
                namespace['main']()
            with open(result) as handle:
                self.assertEqual(json.load(handle), audit, 'dry replay changed persisted audit')
        return events

    def test_unrelated_blank_sku_skip_does_not_hide_a_repair(self):
        events = self.run_main([{'shopify_id': 'gid://shopify/Product/2', 'dw_sku': None, 'fix_type': 'add-sample'}],
                               [{'product_id': '1', 'dw_sku': None, 'fix': 'add-sample', 'status': 'skipped'}])
        self.assertEqual(events, [('process', 'gid://shopify/Product/2', 'add-sample', True)])

    def test_empty_variant_queue_preserves_separate_display_pass_without_budget(self):
        events = self.run_main([{'shopify_id': '1', 'fix_type': 'add-sample'}],
                               [{'product_id': '1', 'fix': 'add-sample', 'status': 'fixed'}], display=True)
        self.assertEqual(events, [('display', 30, True)])


if __name__ == '__main__':
    unittest.main()