Back to Repositories

Testing Logmessaged System Communication and Storage in OpenPilot

This test suite validates the functionality of the logmessaged system in OpenPilot, focusing on log message handling, storage, and IPC communication. It ensures proper logging behavior for both normal and edge cases while verifying message integrity and size limitations.

Test Coverage Overview

The test suite provides comprehensive coverage of the logmessaged system’s core functionality.

Key areas tested include:

Basic log message transmission and reception
Large message handling and size limits
Log file creation and storage
IPC socket communication
Message buffering and drainage

Implementation Analysis

The testing approach utilizes Python’s pytest framework with a robust setup/teardown pattern for consistent test isolation. Tests leverage the cereal messaging system for IPC communication and implement custom socket management for reliable message verification.

Technical patterns include:

Socket subscription and management
Process lifecycle control
File system interaction
Message size validation

Technical Details

Testing infrastructure includes:

cereal.messaging for IPC communication
pytest for test organization
swaglog for logging implementation
Custom process management through managed_processes
File system utilities for log verification

Best Practices Demonstrated

The test suite exemplifies several testing best practices including proper test isolation, resource cleanup, and comprehensive edge case coverage.

Notable practices include:

Explicit setup and teardown procedures
Resource management and cleanup
Verification of both success and edge cases
Systematic message handling validation
Performance consideration in large message tests

commaai/openpilot

system/tests/test_logmessaged.py

            
import glob
import os
import time

import cereal.messaging as messaging
from openpilot.system.manager.process_config import managed_processes
from openpilot.system.hardware.hw import Paths
from openpilot.common.swaglog import cloudlog, ipchandler


class TestLogmessaged:
  def setup_method(self):
    # clear the IPC buffer in case some other tests used cloudlog and filled it
    ipchandler.close()
    ipchandler.connect()

    managed_processes['logmessaged'].start()
    self.sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)
    self.error_sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)

    # ensure sockets are connected
    time.sleep(0.5)
    messaging.drain_sock(self.sock)
    messaging.drain_sock(self.error_sock)

  def teardown_method(self):
    del self.sock
    del self.error_sock
    managed_processes['logmessaged'].stop(block=True)

  def _get_log_files(self):
    return list(glob.glob(os.path.join(Paths.swaglog_root(), "swaglog.*")))

  def test_simple_log(self):
    msgs = [f"abc {i}" for i in range(10)]
    for m in msgs:
      cloudlog.error(m)
    time.sleep(0.5)
    m = messaging.drain_sock(self.sock)
    assert len(m) == len(msgs)
    assert len(self._get_log_files()) >= 1

  def test_big_log(self):
    n = 10
    msg = "a"*3*1024*1024
    for _ in range(n):
      cloudlog.info(msg)
    time.sleep(0.5)

    msgs = messaging.drain_sock(self.sock)
    assert len(msgs) == 0

    logsize = sum([os.path.getsize(f) for f in self._get_log_files()])
    assert (n*len(msg)) < logsize < (n*(len(msg)+1024))