321 lines
10 KiB
Python
321 lines
10 KiB
Python
"""
|
||
Test Suite: BusMonitorCore ARTOS API Compliance
|
||
|
||
This test validates that BusMonitorCore implements the ARTOS BaseModule
|
||
interface correctly and can be used identically by both GUI and ARTOS Collector.
|
||
"""
|
||
import time
|
||
from pathlib import Path
|
||
import sys
|
||
import os
|
||
|
||
# Ensure we can import the module
|
||
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
||
if project_root not in sys.path:
|
||
sys.path.insert(0, project_root)
|
||
|
||
from pybusmonitor1553.core import BusMonitorCore, BaseModule
|
||
|
||
|
||
def test_basemodule_compliance():
|
||
"""Test that BusMonitorCore correctly implements BaseModule interface."""
|
||
print("=" * 60)
|
||
print("TEST 1: BaseModule Interface Compliance")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
|
||
# Check that it's an instance of BaseModule
|
||
assert isinstance(monitor, BaseModule), "BusMonitorCore must inherit from BaseModule"
|
||
print("✓ BusMonitorCore inherits from BaseModule")
|
||
|
||
# Check required methods exist
|
||
required_methods = ['initialize', 'start_session', 'stop_session', 'get_status']
|
||
for method in required_methods:
|
||
assert hasattr(monitor, method), f"Missing required method: {method}"
|
||
assert callable(getattr(monitor, method)), f"{method} must be callable"
|
||
print(f"✓ All required methods present: {', '.join(required_methods)}")
|
||
|
||
print("✓ TEST 1 PASSED\n")
|
||
|
||
|
||
def test_initialization_workflow():
|
||
"""Test the initialization workflow (same as ARTOS would use)."""
|
||
print("=" * 60)
|
||
print("TEST 2: Initialization Workflow")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
|
||
# Configure like ARTOS would
|
||
config = {
|
||
'ip': '127.0.0.1',
|
||
'send_port': 5001,
|
||
'recv_port': 5002
|
||
}
|
||
|
||
print(f"Initializing with config: {config}")
|
||
success = monitor.initialize(config)
|
||
|
||
if not success:
|
||
status = monitor.get_status()
|
||
print(f"⚠ Initialization failed (expected in test environment): {status.get('errors')}")
|
||
print("This is OK - testing API structure, not actual hardware")
|
||
else:
|
||
print("✓ Initialization successful")
|
||
|
||
# Check status is accessible
|
||
status = monitor.get_status()
|
||
assert isinstance(status, dict), "get_status() must return a dict"
|
||
assert 'is_running' in status, "Status must include 'is_running'"
|
||
assert 'connected' in status, "Status must include 'connected'"
|
||
assert 'errors' in status, "Status must include 'errors'"
|
||
print(f"✓ Status structure correct: {list(status.keys())}")
|
||
|
||
print("✓ TEST 2 PASSED\n")
|
||
|
||
|
||
def test_session_lifecycle():
|
||
"""Test start/stop session workflow."""
|
||
print("=" * 60)
|
||
print("TEST 3: Session Lifecycle")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
config = {'ip': '127.0.0.1', 'send_port': 5001, 'recv_port': 5002}
|
||
|
||
# Initialize
|
||
monitor.initialize(config)
|
||
initial_status = monitor.get_status()
|
||
print(f"Initial status: is_running={initial_status['is_running']}")
|
||
|
||
# Start session
|
||
monitor.start_session()
|
||
running_status = monitor.get_status()
|
||
print(f"After start: is_running={running_status['is_running']}")
|
||
|
||
# Give it a moment
|
||
time.sleep(0.5)
|
||
|
||
# Stop session
|
||
monitor.stop_session()
|
||
stopped_status = monitor.get_status()
|
||
print(f"After stop: is_running={stopped_status['is_running']}")
|
||
|
||
assert stopped_status['is_running'] == False, "Should not be running after stop"
|
||
print("✓ Session lifecycle works correctly")
|
||
|
||
print("✓ TEST 3 PASSED\n")
|
||
|
||
|
||
def test_message_access():
|
||
"""Test message retrieval API."""
|
||
print("=" * 60)
|
||
print("TEST 4: Message Access API")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
config = {'ip': '127.0.0.1', 'send_port': 5001, 'recv_port': 5002}
|
||
monitor.initialize(config)
|
||
|
||
# Test get_message method exists
|
||
assert hasattr(monitor, 'get_message'), "Must have get_message method"
|
||
assert hasattr(monitor, 'get_all_messages'), "Must have get_all_messages method"
|
||
print("✓ Message access methods present")
|
||
|
||
# Try to get messages (may be empty in test environment)
|
||
all_messages = monitor.get_all_messages()
|
||
print(f"Messages available: {len(all_messages) if all_messages else 0}")
|
||
|
||
if all_messages:
|
||
# Try to get a specific message
|
||
first_label = list(all_messages.keys())[0]
|
||
msg = monitor.get_message(first_label)
|
||
print(f"✓ Successfully retrieved message: {first_label}")
|
||
else:
|
||
print("ℹ No messages in test environment (OK for unit test)")
|
||
|
||
print("✓ TEST 4 PASSED\n")
|
||
|
||
|
||
def test_callback_system():
|
||
"""Test callback registration (ARTOS test hooks)."""
|
||
print("=" * 60)
|
||
print("TEST 5: Callback System")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
|
||
# Check callback registration method exists
|
||
assert hasattr(monitor, 'register_callback'), "Must have register_callback method"
|
||
print("✓ Callback registration method present")
|
||
|
||
# Register a test callback
|
||
callback_invoked = {'count': 0}
|
||
|
||
def test_callback(message):
|
||
callback_invoked['count'] += 1
|
||
print(f" Callback invoked with message: {message}")
|
||
|
||
monitor.register_callback("test_label", test_callback)
|
||
print("✓ Callback registered successfully")
|
||
|
||
print("✓ TEST 5 PASSED\n")
|
||
|
||
|
||
def test_recording_api():
|
||
"""Test recording/replay API (ARTOS feature)."""
|
||
print("=" * 60)
|
||
print("TEST 6: Recording & Replay API")
|
||
print("=" * 60)
|
||
|
||
monitor = BusMonitorCore()
|
||
|
||
# Check recording methods exist
|
||
required_recording_methods = [
|
||
'start_recording',
|
||
'stop_recording',
|
||
'load_recording'
|
||
]
|
||
|
||
for method in required_recording_methods:
|
||
assert hasattr(monitor, method), f"Missing recording method: {method}"
|
||
assert callable(getattr(monitor, method)), f"{method} must be callable"
|
||
|
||
print(f"✓ All recording methods present: {', '.join(required_recording_methods)}")
|
||
|
||
# Test recording workflow
|
||
test_file = Path("test_recordings/test_session.json")
|
||
test_file.parent.mkdir(exist_ok=True)
|
||
|
||
monitor.start_recording(filepath=test_file)
|
||
status = monitor.get_status()
|
||
assert status['recording'] == True, "Should be recording"
|
||
print("✓ Recording started")
|
||
|
||
saved_path = monitor.stop_recording(save=True)
|
||
status = monitor.get_status()
|
||
assert status['recording'] == False, "Should not be recording"
|
||
print("✓ Recording stopped")
|
||
|
||
if saved_path and saved_path.exists():
|
||
print(f"✓ Recording saved to: {saved_path}")
|
||
# Clean up
|
||
saved_path.unlink()
|
||
saved_path.parent.rmdir()
|
||
|
||
print("✓ TEST 6 PASSED\n")
|
||
|
||
|
||
def test_same_api_for_gui_and_artos():
|
||
"""
|
||
Meta-test: Verify GUI and ARTOS would use identical code.
|
||
|
||
This is the critical validation - if this passes, GUI is truly
|
||
testing the same API that ARTOS will use.
|
||
"""
|
||
print("=" * 60)
|
||
print("TEST 7: API Identity Validation")
|
||
print("=" * 60)
|
||
|
||
# This is what GUI does:
|
||
from pybusmonitor1553.core import BusMonitorCore as GUIMonitor
|
||
gui_monitor = GUIMonitor()
|
||
|
||
# This is what ARTOS would do:
|
||
from pybusmonitor1553.core import BusMonitorCore as ARTOSMonitor
|
||
artos_monitor = ARTOSMonitor()
|
||
|
||
# They should be the same class
|
||
assert type(gui_monitor) == type(artos_monitor), "GUI and ARTOS use different classes!"
|
||
print("✓ GUI and ARTOS import identical BusMonitorCore class")
|
||
|
||
# Both should have the same methods
|
||
gui_methods = {m for m in dir(gui_monitor) if not m.startswith('_') and callable(getattr(gui_monitor, m))}
|
||
artos_methods = {m for m in dir(artos_monitor) if not m.startswith('_') and callable(getattr(artos_monitor, m))}
|
||
|
||
assert gui_methods == artos_methods, "GUI and ARTOS have different public methods!"
|
||
print(f"✓ Both expose identical public API: {len(gui_methods)} methods")
|
||
|
||
# Test that workflow is identical
|
||
config = {'ip': '127.0.0.1', 'send_port': 5001, 'recv_port': 5002}
|
||
|
||
# GUI workflow
|
||
gui_monitor.initialize(config)
|
||
gui_status = gui_monitor.get_status()
|
||
|
||
# ARTOS workflow
|
||
artos_monitor.initialize(config)
|
||
artos_status = artos_monitor.get_status()
|
||
|
||
# Status structures should be identical
|
||
assert gui_status.keys() == artos_status.keys(), "Status structures differ!"
|
||
print("✓ Status structures identical")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("CRITICAL VALIDATION PASSED!")
|
||
print("=" * 60)
|
||
print("GUI is testing the EXACT API that ARTOS will use.")
|
||
print("If GUI works → ARTOS will work!")
|
||
print("=" * 60)
|
||
|
||
print("✓ TEST 7 PASSED\n")
|
||
|
||
|
||
def run_all_tests():
|
||
"""Run complete test suite."""
|
||
print("\n" + "=" * 60)
|
||
print("ARTOS API COMPLIANCE TEST SUITE")
|
||
print("=" * 60)
|
||
print("Validating that BusMonitorCore implements ARTOS interface\n")
|
||
|
||
tests = [
|
||
test_basemodule_compliance,
|
||
test_initialization_workflow,
|
||
test_session_lifecycle,
|
||
test_message_access,
|
||
test_callback_system,
|
||
test_recording_api,
|
||
test_same_api_for_gui_and_artos,
|
||
]
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
for test_func in tests:
|
||
try:
|
||
test_func()
|
||
passed += 1
|
||
except AssertionError as e:
|
||
print(f"✗ TEST FAILED: {test_func.__name__}")
|
||
print(f" Error: {e}\n")
|
||
failed += 1
|
||
except Exception as e:
|
||
print(f"✗ TEST ERROR: {test_func.__name__}")
|
||
print(f" Exception: {e}\n")
|
||
failed += 1
|
||
|
||
# Summary
|
||
print("\n" + "=" * 60)
|
||
print("TEST SUITE SUMMARY")
|
||
print("=" * 60)
|
||
print(f"Total tests: {len(tests)}")
|
||
print(f"Passed: {passed} ✓")
|
||
print(f"Failed: {failed} ✗")
|
||
print("=" * 60)
|
||
|
||
if failed == 0:
|
||
print("\n🎉 ALL TESTS PASSED!")
|
||
print("\nBusMonitorCore is ARTOS-compliant and ready for integration.")
|
||
print("The GUI is now a true test client for the ARTOS module.")
|
||
return 0
|
||
else:
|
||
print("\n⚠ SOME TESTS FAILED")
|
||
print("Please review failures before ARTOS integration.")
|
||
return 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
exit_code = run_all_tests()
|
||
sys.exit(exit_code)
|