"""CPU-only routing regression: python test_cli_routing.py /path/to/patched/OneJev.""" import argparse import importlib.util import contextlib import json from pathlib import Path import sys import tempfile import types import unittest from unittest.mock import patch ROOT = Path(sys.argv.pop(1)) class RoutingTest(unittest.TestCase): def route(self, config, readout=None, multimodal=False): # Import the actual readout module and execute the CLI; stub GPU/server boundaries. called = [] def engine(kind): return lambda *a, **kw: called.append(kind) or object() mods = {} for name, attrs in { 'uvicorn': {'run': lambda *a, **kw: None}, 'qev': {}, 'qev.calibrate': {'Calibration': object}, 'qev.engine': {'DecisionEngine': engine('text')}, 'qev.mm_engine': {'MMDecisionEngine': engine('multimodal'), 'local_model_dir': lambda x: x}, 'torch': {'nn': types.SimpleNamespace(Module=object)}, 'qev.prompt': {'MAX_SLOTS': 256}, 'qev.server': {'create_app': lambda *a, **kw: object()}, }.items(): mods[name] = types.ModuleType(name) mods[name].__dict__.update(attrs) with tempfile.TemporaryDirectory() as d, patch.dict(sys.modules, mods), patch('importlib.util.find_spec', return_value=object()): Path(d, 'config.json').write_text(json.dumps(config)) if readout is not None: Path(d, 'readout.json').write_text(json.dumps(readout)) spec = importlib.util.spec_from_file_location('qev.readout', ROOT / 'qev/readout.py') module = importlib.util.module_from_spec(spec) sys.modules['qev.readout'] = module spec.loader.exec_module(module) cli = {'__name__': 'qev.cli', '__package__': 'qev'} exec(compile((ROOT / 'qev/cli.py').read_text(), 'cli.py', 'exec'), cli) args = argparse.Namespace(model=d, calibration=None, multimodal=multimodal, gguf=None, llama_url=None, device='cpu', dtype='bfloat16', fork_mode='sequential', head_dtype='float32', media_root=None, no_gpu_preprocess=False, no_cuda_graphs=False, quantize=None, max_branch_tokens=16384, max_request_tokens=65536, name=None, host='127.0.0.1', port=8000) cli['cmd_serve'](args) return called def test_routes(self): layout = {'vision_config': {}, 'modeling_overrides': {'attention': {'full_attention': 'block_bidirectional'}}} for config, readout, explicit, expected in [ (layout, None, False, 'text'), ({}, None, False, 'text'), ({'vision_config': {}}, None, False, 'multimodal'), ({'vision_config': {}}, None, True, 'multimodal'), ({'vision_config': {}}, {'attention': 'noncausal'}, False, 'text'), ({'vision_config': {}}, {'readout': 'head'}, False, 'multimodal'), ]: with self.subTest(config=config, readout=readout, explicit=explicit): self.assertEqual(self.route(config, readout, explicit), [expected]) with self.assertRaisesRegex(SystemExit, 'omit --multimodal'): self.route(layout, multimodal=True) with self.assertRaisesRegex(ValueError, 'unsupported full_attention'): self.route({'modeling_overrides': {'attention': {'full_attention': 'unknown'}}}) if __name__ == '__main__': unittest.main()