| """Intel XPU routing/transport regressions. Mocks are not physical Intel GPU validation.""" |
| import argparse |
| import contextlib |
| import io |
| import json |
| import unittest |
| from unittest import mock |
|
|
| import torch |
| import dbmoe |
| from test_dbmoe import tiny_lease |
|
|
|
|
| class IntelDeviceTests(unittest.TestCase): |
| def fake_xpu(self, available=True, count=2, current=0): |
| backend = mock.Mock() |
| backend.is_available.return_value = available |
| backend.device_count.return_value = count |
| backend.current_device.return_value = current |
| backend.get_device_name.return_value = 'Mock Intel GPU' |
| backend.get_device_properties.return_value = argparse.Namespace(total_memory=16 * 2**30) |
| patcher = mock.patch.object(torch, 'xpu', backend, create=True) |
| patcher.start() |
| self.addCleanup(patcher.stop) |
| return backend |
|
|
| def test_cpu_never_probes_xpu(self): |
| backend = self.fake_xpu() |
| backend.is_available.side_effect = AssertionError('XPU touched') |
| self.assertEqual(dbmoe.resolve_worker_device('cpu')['backend'], 'cpu') |
| backend.is_available.assert_not_called() |
|
|
| def test_explicit_xpu_and_intel_alias_use_only_xpu(self): |
| backend = self.fake_xpu(current=1) |
| with mock.patch.object(torch.cuda, 'is_available', side_effect=AssertionError('CUDA touched')), \ |
| mock.patch.object(dbmoe, 'mps_available', side_effect=AssertionError('MPS touched')): |
| for name in ('xpu', 'intel', ' XPU:1 ', 'intel:1'): |
| self.assertEqual(dbmoe.resolve_worker_device(name), |
| dict(torch_device='xpu:1', device='xpu:1', backend='xpu')) |
| self.assertEqual(dbmoe.resolve_worker_device('xpu:0')['torch_device'], 'xpu:0') |
| self.assertTrue(backend.is_available.called) |
|
|
| def test_auto_order_cuda_xpu_mps_cpu(self): |
| backend = self.fake_xpu() |
| with mock.patch.object(torch.cuda, 'is_available', return_value=True), \ |
| mock.patch.object(torch.cuda, 'device_count', return_value=1), \ |
| mock.patch.object(torch.cuda, 'current_device', return_value=0), \ |
| mock.patch.object(torch.version, 'hip', None), \ |
| mock.patch.object(dbmoe, 'mps_available', return_value=True) as mps: |
| self.assertEqual(dbmoe.resolve_worker_device('auto')['backend'], 'cuda') |
| backend.is_available.assert_not_called() |
| mps.assert_not_called() |
| with mock.patch.object(torch.cuda, 'is_available', return_value=False), \ |
| mock.patch.object(torch.backends.mps, 'is_built', return_value=True), \ |
| mock.patch.object(torch.backends.mps, 'is_available', return_value=True), \ |
| mock.patch.object(dbmoe, 'mps_available', return_value=True) as mps: |
| self.assertEqual(dbmoe.resolve_worker_device('auto')['backend'], 'xpu') |
| mps.assert_not_called() |
| backend.is_available.return_value = False |
| self.assertEqual(dbmoe.resolve_worker_device('auto')['backend'], 'mps') |
| mps.return_value = False |
| self.assertEqual(dbmoe.resolve_worker_device('auto')['backend'], 'cpu') |
|
|
| def test_missing_xpu_backend_supports_cpu_and_auto(self): |
| with mock.patch.object(torch, 'xpu', None, create=True), \ |
| mock.patch.object(torch.cuda, 'is_available', return_value=False), \ |
| mock.patch.object(dbmoe, 'mps_available', return_value=False): |
| self.assertFalse(dbmoe.xpu_available()) |
| self.assertEqual(dbmoe.resolve_worker_device('cpu')['backend'], 'cpu') |
| self.assertEqual(dbmoe.resolve_worker_device('auto')['backend'], 'cpu') |
| with self.assertRaisesRegex(RuntimeError, 'XPU-enabled PyTorch'): |
| dbmoe.resolve_worker_device('xpu') |
|
|
| def test_unavailable_or_broken_explicit_xpu_never_falls_back(self): |
| backend = self.fake_xpu(available=False) |
| with mock.patch.object(torch.cuda, 'is_available', return_value=True) as cuda: |
| for name in ('xpu', 'xpu:0', 'intel', 'intel:1'): |
| with self.subTest(name=name), self.assertRaisesRegex(RuntimeError, 'cannot access an Intel GPU'): |
| dbmoe.resolve_worker_device(name) |
| backend.is_available.side_effect = RuntimeError('driver initialization failed') |
| with self.assertRaisesRegex(RuntimeError, 'cannot access an Intel GPU'): |
| dbmoe.resolve_worker_device('xpu') |
| cuda.assert_not_called() |
|
|
| def test_invalid_indexes_rejected_without_probing(self): |
| backend = self.fake_xpu() |
| for name in ('xpu:', 'xpu:-1', 'xpu:1:2', 'xpu:abc', 'xpu:²', 'intel:1.0', 'xpu:+1'): |
| with self.subTest(name=name), self.assertRaises(ValueError): |
| dbmoe.resolve_worker_device(name) |
| backend.is_available.assert_not_called() |
|
|
| def test_out_of_range_and_empty_device_list_rejected(self): |
| backend = self.fake_xpu() |
| with self.assertRaisesRegex(ValueError, 'out of range'): |
| dbmoe.resolve_worker_device('xpu:2') |
| backend.device_count.return_value = 0 |
| with self.assertRaisesRegex(ValueError, 'out of range'): |
| dbmoe.resolve_worker_device('xpu') |
|
|
| def test_selected_xpu_is_used_for_preflight(self): |
| backend = self.fake_xpu() |
| with mock.patch.object(torch.cuda, 'set_device', side_effect=AssertionError('CUDA touched')), \ |
| mock.patch.object(torch.cuda, 'get_device_name', side_effect=AssertionError('CUDA touched')), \ |
| mock.patch.object(dbmoe, 'device_smoke_test', return_value=[1., .9]) as probe: |
| info = dbmoe.prepare_worker_device('xpu:1') |
| backend.set_device.assert_called_once_with('xpu:1') |
| backend.get_device_name.assert_called_once_with('xpu:1') |
| probe.assert_called_once_with('xpu:1') |
| self.assertEqual(info['gpu_name'], 'Mock Intel GPU') |
|
|
| def test_probe_failure_precedes_network_enrolment_and_token_read(self): |
| self.fake_xpu() |
| a = argparse.Namespace(threads=1, device='xpu', token='@missing-token-file') |
| with mock.patch.object(dbmoe, 'device_smoke_test', side_effect=RuntimeError('unsupported operation')), \ |
| mock.patch.object(dbmoe, 'http') as http, \ |
| mock.patch.object(dbmoe, 'load_or_join') as join, \ |
| mock.patch('builtins.open', side_effect=AssertionError('token file touched')): |
| with self.assertRaisesRegex(SystemExit, 'failed the offline training check'): |
| dbmoe.worker(a) |
| http.assert_not_called() |
| join.assert_not_called() |
|
|
| def test_hardware_report_uses_selected_xpu_and_allowlist(self): |
| backend = self.fake_xpu() |
| with mock.patch.object(torch.cuda, 'get_device_properties', side_effect=AssertionError('CUDA touched')), \ |
| mock.patch.object(torch.version, 'xpu', 'test-xpu', create=True), \ |
| mock.patch.object(dbmoe.socket, 'gethostname', side_effect=AssertionError('hostname touched')): |
| report = dbmoe.worker_hardware(dict(backend='xpu', torch_device='xpu:1', gpu_name='Mock Intel GPU'), 2) |
| backend.get_device_properties.assert_called_once_with('xpu:1') |
| self.assertEqual(report['backend'], 'xpu') |
| self.assertEqual(report['runtime_version'], 'test-xpu') |
| self.assertEqual(report['gpu_memory_gib'], 16.) |
| self.assertEqual(report['gpu_name'], 'Mock Intel GPU') |
| self.assertNotIn('host', report) |
|
|
| def test_unknown_xpu_runtime_version_is_empty_not_invented(self): |
| self.fake_xpu() |
| with mock.patch.object(torch.version, 'xpu', None, create=True): |
| report = dbmoe.worker_hardware(dict(backend='xpu', torch_device='xpu:0', gpu_name='Mock Intel GPU'), 2) |
| self.assertEqual(report['runtime_version'], '') |
|
|
| def test_xpu_payload_retains_backend_and_hardware_opt_in(self): |
| torch.set_num_threads(1) |
| lease = tiny_lease() |
| update = dbmoe.train_lease(lease, 'cpu') |
| self.fake_xpu() |
| for opt_in in (False, True): |
| sent = [] |
| def http(method, url, token, body=None): |
| if method == 'GET': |
| return dbmoe.to_bytes(lease) |
| sent.append(dbmoe.from_bytes(body)) |
| return b'{"ok":true}' |
| args = argparse.Namespace(threads=1, device='xpu:1', url='https://example.invalid', token='test-only', |
| name='test', hours=0, steps=2, batch=1, once=1, no_share_hardware=not opt_in) |
| info = dict(torch_device='xpu:1', device='xpu:1', backend='xpu', gpu_name='Mock Intel GPU') |
| with mock.patch.object(dbmoe, 'prepare_worker_device', return_value=info), \ |
| mock.patch.object(dbmoe, 'train_lease', return_value=dict(update)) as train, \ |
| mock.patch.object(dbmoe, 'http', side_effect=http), \ |
| contextlib.redirect_stdout(io.StringIO()): |
| dbmoe.worker(args) |
| self.assertEqual(train.call_args.args[1], 'xpu:1') |
| self.assertEqual(sent[0]['device'], 'xpu:1') |
| self.assertEqual('hardware' in sent[0], opt_in) |
| if opt_in: |
| self.assertEqual(sent[0]['hardware']['backend'], 'xpu') |
| self.assertNotIn('host', sent[0]) |
| for section in ('d_shared', 'd_block'): |
| for value in sent[0][section].values(): |
| self.assertEqual(value.device.type, 'cpu') |
| self.assertEqual(value.dtype, torch.float32) |
|
|
| def test_cli_accepts_xpu_without_network(self): |
| with mock.patch.object(dbmoe.sys, 'argv', ['dbmoe.py', 'worker', '--device', 'xpu:1']), \ |
| mock.patch.object(dbmoe, 'worker') as worker: |
| dbmoe.main() |
| self.assertEqual(worker.call_args.args[0].device, 'xpu:1') |
|
|
| def test_offline_cli_reports_xpu_version_and_device(self): |
| self.fake_xpu() |
| output = io.StringIO() |
| with mock.patch.object(dbmoe, 'device_smoke_test', return_value=[1., .9]), \ |
| mock.patch.object(torch.version, 'xpu', 'test-xpu', create=True), \ |
| mock.patch.object(dbmoe, 'http', side_effect=AssertionError('network touched')), \ |
| contextlib.redirect_stdout(output): |
| dbmoe.check_device_cmd(argparse.Namespace(device='xpu:1', threads=1)) |
| report = json.loads(output.getvalue()) |
| self.assertTrue(report['ok']) |
| self.assertEqual(report['torch_device'], 'xpu:1') |
| self.assertEqual(report['xpu_version'], 'test-xpu') |
|
|
| @unittest.skipUnless(dbmoe.xpu_available(), 'No Intel XPU GPU available; physical Intel test not run') |
| def test_real_xpu_training_and_cpu_serialization(self): |
| torch.set_num_threads(1) |
| info = dbmoe.prepare_worker_device('xpu') |
| self.assertEqual(info['backend'], 'xpu') |
| self.assertEqual(len(info['check_losses']), 2) |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
