Testing Examples
Overview
This guide demonstrates testing techniques for:
Unit testing with mock instruments
Integration testing with mock/real hardware
Performance validation testing
Type safety verification
Basic Unit Tests
Mock Controller Tests
import unittest
from typing import Dict, Optional
import numpy as np
from serdes_validation_framework.instrument_control.mock_controller import (
get_instrument_controller,
MockInstrumentController
)
class TestMockController(unittest.TestCase):
"""Test mock controller functionality"""
def setUp(self) -> None:
"""Set up test fixtures"""
# Force mock mode
os.environ['SVF_MOCK_MODE'] = '1'
self.controller = get_instrument_controller()
def test_basic_communication(self) -> None:
"""Test basic instrument communication"""
resource = 'GPIB::1::INSTR'
try:
# Connect
self.controller.connect_instrument(resource)
# Query ID
response = self.controller.query_instrument(resource, '*IDN?')
self.assertIsInstance(response, str)
self.assertGreater(len(response), 0)
finally:
# Cleanup
self.controller.disconnect_instrument(resource)
def test_custom_responses(self) -> None:
"""Test custom response configuration"""
# Configure response
self.controller.add_mock_response(
'TEST:VALUE?',
lambda: f"{np.random.normal(0, 0.1):.6f}",
delay=0.1
)
resource = 'GPIB::1::INSTR'
self.controller.connect_instrument(resource)
try:
# Query multiple times
values = []
for _ in range(5):
response = self.controller.query_instrument(
resource,
'TEST:VALUE?'
)
values.append(float(response))
# Check variation
self.assertGreater(np.std(values), 0)
finally:
self.controller.disconnect_instrument(resource)
Data Type Tests
class TestDataValidation(unittest.TestCase):
"""Test data type validation"""
def setUp(self) -> None:
"""Set up test data"""
self.test_data = {
'time': np.arange(1000, dtype=np.float64) / 256e9,
'voltage': np.random.normal(0, 1, 1000).astype(np.float64)
}
def test_array_validation(self) -> None:
"""Test array type validation"""
from serdes_validation_framework.data_analysis import validate_array
# Test valid array
valid_array = np.array([1.0, 2.0, 3.0], dtype=np.float64)
validate_array(valid_array, "test_array")
# Test invalid type
with self.assertRaises(AssertionError):
invalid_array = np.array([1, 2, 3]) # Integer array
validate_array(invalid_array, "test_array")
# Test empty array
with self.assertRaises(AssertionError):
empty_array = np.array([], dtype=np.float64)
validate_array(empty_array, "test_array")
def test_parameter_validation(self) -> None:
"""Test numeric parameter validation"""
def validate_parameter(
value: float,
name: str,
min_val: Optional[float] = None,
max_val: Optional[float] = None
) -> None:
assert isinstance(value, float), \
f"{name} must be float"
if min_val is not None:
assert value >= min_val, \
f"{name} must be >= {min_val}"
if max_val is not None:
assert value <= max_val, \
f"{name} must be <= {max_val}"
# Test valid value
validate_parameter(1.0, "test_param", 0.0, 2.0)
# Test invalid type
with self.assertRaises(AssertionError):
validate_parameter(1, "test_param") # Integer
# Test range violation
with self.assertRaises(AssertionError):
validate_parameter(3.0, "test_param", 0.0, 2.0)
Integration Tests
Mock Scope Tests
class TestMockScope(unittest.TestCase):
"""Test scope functionality with mock controller"""
def setUp(self) -> None:
"""Set up test fixtures"""
os.environ['SVF_MOCK_MODE'] = '1'
self.scope = HighBandwidthScope('GPIB0::7::INSTR')
def tearDown(self) -> None:
"""Clean up test fixtures"""
self.scope.cleanup()
def test_waveform_capture(self) -> None:
"""Test waveform capture functionality"""
# Configure scope
config = ScopeConfig(
sampling_rate=256e9,
bandwidth=120e9,
timebase=5e-12,
voltage_range=0.8
)
self.scope.configure_for_224g(config)
# Capture waveform
waveform = self.scope.capture_waveform(
duration=1e-6,
sample_rate=config.sampling_rate
)
# Validate data
self.assertIsInstance(waveform.time, np.ndarray)
self.assertIsInstance(waveform.voltage, np.ndarray)
self.assertEqual(len(waveform.time), len(waveform.voltage))
self.assertTrue(np.issubdtype(waveform.voltage.dtype, np.floating))
def test_eye_measurement(self) -> None:
"""Test eye diagram measurement"""
result = self.scope.measure_eye_diagram()
# Validate results
self.assertIsInstance(result.eye_heights, list)
self.assertEqual(len(result.eye_heights), 3) # PAM4 has 3 eyes
self.assertTrue(all(isinstance(h, float) for h in result.eye_heights))
Signal Analysis Tests
class TestSignalAnalysis(unittest.TestCase):
"""Test signal analysis with mock data"""
def setUp(self) -> None:
"""Set up test data"""
# Generate test signal
num_points = 10000
time = np.arange(num_points, dtype=np.float64) / 256e9
# PAM4 levels
levels = np.array([-3.0, -1.0, 1.0, 3.0], dtype=np.float64)
symbols = np.random.choice(levels, size=num_points)
voltage = symbols + np.random.normal(0, 0.1, num_points)
self.test_data = {
'time': time,
'voltage': voltage.astype(np.float64)
}
def test_level_analysis(self) -> None:
"""Test PAM4 level analysis"""
analyzer = PAM4Analyzer(self.test_data)
levels = analyzer.analyze_level_separation()
# Check level count
self.assertEqual(len(levels.level_means), 4)
# Check level ordering
sorted_levels = np.sort(levels.level_means)
self.assertTrue(np.all(np.diff(sorted_levels) > 0))
# Check uniformity
self.assertLess(levels.uniformity, 0.2)
def test_evm_calculation(self) -> None:
"""Test EVM calculation"""
analyzer = PAM4Analyzer(self.test_data)
evm = analyzer.calculate_evm()
# Check EVM ranges
self.assertGreater(evm.rms_evm_percent, 0)
self.assertLess(evm.rms_evm_percent, 10)
self.assertGreater(evm.peak_evm_percent, evm.rms_evm_percent)
Performance Tests
Mock Performance Testing
class TestPerformance(unittest.TestCase):
"""Test performance with mock controller"""
def setUp(self) -> None:
"""Set up test environment"""
os.environ['SVF_MOCK_MODE'] = '1'
self.controller = get_instrument_controller()
def test_response_timing(self) -> None:
"""Test response timing"""
import time
# Configure delayed response
delay = 0.1 # 100 ms
self.controller.add_mock_response(
'TEST:DELAY?',
'response',
delay=delay
)
resource = 'GPIB::1::INSTR'
self.controller.connect_instrument(resource)
try:
# Measure response time
start = time.time()
self.controller.query_instrument(resource, 'TEST:DELAY?')
elapsed = time.time() - start
# Check timing
self.assertGreaterEqual(elapsed, delay)
self.assertLess(elapsed, delay * 1.5)
finally:
self.controller.disconnect_instrument(resource)
def test_throughput(self) -> None:
"""Test data throughput"""
# Configure waveform response
num_points = 1000000
self.controller.add_mock_response(
':WAVeform:DATA?',
lambda: ','.join(['0.0'] * num_points),
delay=0.5
)
resource = 'GPIB::1::INSTR'
self.controller.connect_instrument(resource)
try:
# Measure transfer time
start = time.time()
response = self.controller.query_instrument(
resource,
':WAVeform:DATA?'
)
elapsed = time.time() - start
# Calculate throughput
data_size = len(response.encode())
throughput = data_size / elapsed / 1e6 # MB/s
print(f"Throughput: {throughput:.1f} MB/s")
finally:
self.controller.disconnect_instrument(resource)
Error Handling Tests
Mock Error Tests
class TestErrorHandling(unittest.TestCase):
"""Test error handling with mock controller"""
def setUp(self) -> None:
"""Set up test environment"""
self.controller = get_instrument_controller()
# Configure high error rates
self.controller.set_error_rates(
connection_error_rate=0.5,
command_error_rate=0.5
)
def test_connection_errors(self) -> None:
"""Test connection error handling"""
attempts = 10
failures = 0
for _ in range(attempts):
try:
self.controller.connect_instrument('GPIB::1::INSTR')
self.controller.disconnect_instrument('GPIB::1::INSTR')
except Exception:
failures += 1
# Check failure rate
failure_rate = failures / attempts
self.assertGreater(failure_rate, 0.2)
self.assertLess(failure_rate, 0.8)
def test_command_errors(self) -> None:
"""Test command error handling"""
resource = 'GPIB::1::INSTR'
self.controller.connect_instrument(resource)
try:
attempts = 10
failures = 0
for _ in range(attempts):
try:
self.controller.query_instrument(resource, '*IDN?')
except Exception:
failures += 1
# Check failure rate
failure_rate = failures / attempts
self.assertGreater(failure_rate, 0.2)
self.assertLess(failure_rate, 0.8)
finally:
self.controller.disconnect_instrument(resource)
Running Tests
Test Execution
def run_test_suite() -> None:
"""Run complete test suite"""
# Create test suite
suite = unittest.TestSuite()
# Add test cases
suite.addTests(unittest.makeSuite(TestMockController))
suite.addTests(unittest.makeSuite(TestDataValidation))
suite.addTests(unittest.makeSuite(TestMockScope))
suite.addTests(unittest.makeSuite(TestSignalAnalysis))
suite.addTests(unittest.makeSuite(TestPerformance))
suite.addTests(unittest.makeSuite(TestErrorHandling))
# Run tests
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
if __name__ == '__main__':
run_test_suite()
Best Practices
Always validate numeric types:
def test_numeric_validation(self) -> None: def validate_float(value: float) -> None: assert isinstance(value, float), \ f"Value must be float, got {type(value)}" # Test cases validate_float(1.0) # OK with self.assertRaises(AssertionError): validate_float(1) # Integer not allowed
Use proper cleanup:
def setUp(self) -> None: self.resources = [] def tearDown(self) -> None: for resource in self.resources: try: self.controller.disconnect_instrument(resource) except Exception as e: print(f"Cleanup error: {e}")
Test error conditions:
def test_error_handling(self) -> None: with self.assertRaises(ValueError): # Test invalid input self.controller.connect_instrument("")