cluster4npu/test_multi_series_integration_final.py

203 lines
8.3 KiB
Python

"""
Final Integration Test for Multi-Series Multidongle
Comprehensive test suite for the completed multi-series integration
"""
import unittest
import sys
import os
# Add project root to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'core', 'functions'))
from Multidongle import MultiDongle, DongleSeriesSpec
class TestMultiSeriesIntegration(unittest.TestCase):
def setUp(self):
"""Set up test fixtures"""
self.multi_series_config = {
"KL520": {
"port_ids": [28, 32],
"model_path": "/path/to/kl520_model.nef",
"firmware_paths": {
"scpu": "/path/to/kl520_scpu.bin",
"ncpu": "/path/to/kl520_ncpu.bin"
}
},
"KL720": {
"port_ids": [40, 44],
"model_path": "/path/to/kl720_model.nef",
"firmware_paths": {
"scpu": "/path/to/kl720_scpu.bin",
"ncpu": "/path/to/kl720_ncpu.bin"
}
}
}
def test_multi_series_initialization_success(self):
"""Test that multi-series initialization works correctly"""
multidongle = MultiDongle(multi_series_config=self.multi_series_config)
# Should be in multi-series mode
self.assertTrue(multidongle.multi_series_mode)
# Should have series groups configured
self.assertIsNotNone(multidongle.series_groups)
self.assertIn("KL520", multidongle.series_groups)
self.assertIn("KL720", multidongle.series_groups)
# Should have correct configuration for each series
kl520_config = multidongle.series_groups["KL520"]
self.assertEqual(kl520_config["port_ids"], [28, 32])
self.assertEqual(kl520_config["model_path"], "/path/to/kl520_model.nef")
kl720_config = multidongle.series_groups["KL720"]
self.assertEqual(kl720_config["port_ids"], [40, 44])
self.assertEqual(kl720_config["model_path"], "/path/to/kl720_model.nef")
# Should have GOPS weights calculated
self.assertIsNotNone(multidongle.gops_weights)
self.assertIn("KL520", multidongle.gops_weights)
self.assertIn("KL720", multidongle.gops_weights)
# KL720 should have higher weight due to higher GOPS (28 vs 3 GOPS)
# But since both have 2 devices: KL520=3*2=6 total GOPS, KL720=28*2=56 total GOPS
# Total = 62 GOPS, so KL520 weight = 6/62 ≈ 0.097, KL720 weight = 56/62 ≈ 0.903
self.assertGreater(multidongle.gops_weights["KL720"],
multidongle.gops_weights["KL720"])
# Weights should sum to 1.0
total_weight = sum(multidongle.gops_weights.values())
self.assertAlmostEqual(total_weight, 1.0, places=5)
print("Multi-series initialization test passed")
def test_single_series_to_multi_series_conversion_success(self):
"""Test that single-series config gets converted to multi-series internally"""
# Legacy single-series initialization
multidongle = MultiDongle(
port_id=[28, 32],
scpu_fw_path="/path/to/scpu.bin",
ncpu_fw_path="/path/to/ncpu.bin",
model_path="/path/to/model.nef",
upload_fw=True
)
# Should NOT be in explicit multi-series mode (legacy mode)
self.assertFalse(multidongle.multi_series_mode)
# But should internally convert to multi-series format
self.assertIsNotNone(multidongle.series_groups)
self.assertEqual(len(multidongle.series_groups), 1)
# Should auto-detect series (will be KL520 based on available devices or fallback)
series_keys = list(multidongle.series_groups.keys())
self.assertEqual(len(series_keys), 1)
detected_series = series_keys[0]
self.assertIn(detected_series, DongleSeriesSpec.SERIES_SPECS.keys())
# Should have correct port configuration
series_config = multidongle.series_groups[detected_series]
self.assertEqual(series_config["port_ids"], [28, 32])
self.assertEqual(series_config["model_path"], "/path/to/model.nef")
# Should have 100% weight since it's single series
self.assertEqual(multidongle.gops_weights[detected_series], 1.0)
print(f"Single-to-multi-series conversion test passed (detected: {detected_series})")
def test_load_balancing_success(self):
"""Test that load balancing works based on GOPS weights"""
multidongle = MultiDongle(multi_series_config=self.multi_series_config)
# Should have load balancing method
optimal_series = multidongle._select_optimal_series()
self.assertIsNotNone(optimal_series)
self.assertIn(optimal_series, ["KL520", "KL720"])
# With zero load, should select the series with highest weight (KL720)
self.assertEqual(optimal_series, "KL720")
# Test load balancing under different conditions
# Simulate high load on KL720
multidongle.current_loads["KL720"] = 100
multidongle.current_loads["KL520"] = 0
# Now should prefer KL520 despite lower GOPS due to lower load
optimal_series_with_load = multidongle._select_optimal_series()
self.assertEqual(optimal_series_with_load, "KL520")
print("Load balancing test passed")
def test_backward_compatibility_maintained(self):
"""Test that existing single-series API still works perfectly"""
# This should work exactly as before
multidongle = MultiDongle(
port_id=[28, 32],
scpu_fw_path="/path/to/scpu.bin",
ncpu_fw_path="/path/to/ncpu.bin",
model_path="/path/to/model.nef"
)
# Legacy properties should still exist and work
self.assertIsNotNone(multidongle.port_id)
self.assertEqual(multidongle.port_id, [28, 32])
self.assertEqual(multidongle.model_path, "/path/to/model.nef")
self.assertEqual(multidongle.scpu_fw_path, "/path/to/scpu.bin")
self.assertEqual(multidongle.ncpu_fw_path, "/path/to/ncpu.bin")
# Legacy attributes should be available
self.assertIsNotNone(multidongle.device_group) # Will be None initially
self.assertIsNotNone(multidongle._input_queue)
self.assertIsNotNone(multidongle._output_queue)
print("Backward compatibility test passed")
def test_series_specs_are_correct(self):
"""Test that series specifications match expected values"""
specs = DongleSeriesSpec.SERIES_SPECS
# Check that all expected series are present
expected_series = ["KL520", "KL720", "KL630", "KL730", "KL540"]
for series in expected_series:
self.assertIn(series, specs)
# Check GOPS values are reasonable
self.assertEqual(specs["KL520"]["gops"], 3)
self.assertEqual(specs["KL720"]["gops"], 28)
self.assertEqual(specs["KL630"]["gops"], 400)
self.assertEqual(specs["KL730"]["gops"], 1600)
self.assertEqual(specs["KL540"]["gops"], 800)
print("Series specifications test passed")
def test_edge_cases(self):
"""Test various edge cases and error handling"""
# Test with empty port list (single-series)
multidongle_empty = MultiDongle(port_id=[])
self.assertEqual(len(multidongle_empty.series_groups), 0)
# Test with unknown series (should raise error)
with self.assertRaises(ValueError):
MultiDongle(multi_series_config={"UNKNOWN_SERIES": {"port_ids": [1, 2]}})
# Test with no port IDs in multi-series config
config_no_ports = {
"KL520": {
"port_ids": [],
"model_path": "/path/to/model.nef"
}
}
multidongle_no_ports = MultiDongle(multi_series_config=config_no_ports)
self.assertEqual(multidongle_no_ports.gops_weights["KL520"], 0.0) # 0 weight due to no devices
print("Edge cases test passed")
if __name__ == '__main__':
print("Running Multi-Series Integration Tests")
print("=" * 50)
unittest.main(verbosity=2)