| """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): |
| |
| 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() |
|
|
