253 lines
10 KiB
Python
253 lines
10 KiB
Python
# SPDX-License-Identifier: GPL-2.0-only
|
|
#
|
|
# Copyright (c) 2023 Linux CAN project
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License for more details.
|
|
|
|
import pytest
|
|
import subprocess
|
|
import os
|
|
import time
|
|
import signal
|
|
import select
|
|
import shutil
|
|
import re
|
|
|
|
# --- Helper Functions ---
|
|
|
|
class CanSnifferMonitor:
|
|
"""
|
|
Context manager to run cansniffer.
|
|
Captures stdout/stderr non-blocking way.
|
|
"""
|
|
def __init__(self, bin_path, interface, args=None):
|
|
# CRITICAL: Use stdbuf -o0 to force unbuffered stdout.
|
|
# cansniffer uses printf, which is fully buffered when writing to a pipe.
|
|
# Without stdbuf, we would detect empty output until 4KB of data accumulates.
|
|
if not shutil.which("stdbuf"):
|
|
pytest.fail("stdbuf utility (coreutils) is required for testing cansniffer TUI")
|
|
|
|
self.cmd = ["stdbuf", "-o0", os.path.join(bin_path, "cansniffer"), interface]
|
|
if args:
|
|
self.cmd.extend(args)
|
|
self.process = None
|
|
self.output_buffer = ""
|
|
|
|
def __enter__(self):
|
|
# We need bufsize=0 or unbuffered to get output in real-time
|
|
self.process = subprocess.Popen(
|
|
self.cmd,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=0,
|
|
preexec_fn=os.setsid
|
|
)
|
|
time.sleep(0.5) # Wait for initialization (clearing screen etc)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self.process:
|
|
try:
|
|
os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
def send_input(self, text):
|
|
"""Sends commands to cansniffer stdin."""
|
|
if self.process and self.process.stdin:
|
|
self.process.stdin.write(text)
|
|
self.process.stdin.flush()
|
|
time.sleep(0.5) # Allow processing time (increased for stability)
|
|
|
|
def read_output(self, timeout=1.0):
|
|
"""
|
|
Reads available output from stdout without blocking forever.
|
|
"""
|
|
collected = ""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
# Check if there is data to read
|
|
reads = [self.process.stdout.fileno()]
|
|
ret = select.select(reads, [], [], 0.1)
|
|
|
|
if ret[0]:
|
|
# Read chunks
|
|
chunk = self.process.stdout.read(1024)
|
|
if not chunk:
|
|
break
|
|
collected += chunk
|
|
else:
|
|
# If we have collected something and no new data is coming fast, break early
|
|
# to speed up tests, unless we are waiting for specific timeout
|
|
if collected and (time.time() - start > 0.3):
|
|
break
|
|
|
|
self.output_buffer += collected
|
|
return collected
|
|
|
|
def clear_buffer(self):
|
|
"""Discards currently available output."""
|
|
self.read_output(timeout=0.2)
|
|
|
|
# --- Tests for cansniffer ---
|
|
|
|
def test_help_option(bin_path):
|
|
"""Test -? option (cansniffer uses -? for help)."""
|
|
# Note: The help text says '-?' prints help.
|
|
result = subprocess.run([os.path.join(bin_path, "cansniffer"), "-?"], capture_output=True, text=True)
|
|
# cansniffer usually returns error code because ? is invalid opt for standard parsers
|
|
assert "Usage: cansniffer" in result.stdout or "Usage: cansniffer" in result.stderr
|
|
|
|
def test_basic_sniffing(bin_path, can_interface):
|
|
"""Test if cansniffer shows traffic."""
|
|
with CanSnifferMonitor(bin_path, can_interface) as sniffer:
|
|
# Generate traffic: ID 123
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "123#112233"], check=True)
|
|
|
|
# Read output
|
|
out = sniffer.read_output(timeout=2.0)
|
|
|
|
# Check if 123 appears.
|
|
assert "123" in out
|
|
|
|
def test_filtering_add(bin_path, can_interface):
|
|
"""Test adding specific ID filter (+ID)."""
|
|
# Start with -q (quiet, all IDs deactivated)
|
|
with CanSnifferMonitor(bin_path, can_interface, args=["-q"]) as sniffer:
|
|
# 1. Send ID 123 -> Should NOT appear (quiet mode)
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "123#AA"], check=True)
|
|
out = sniffer.read_output(timeout=1.0)
|
|
assert "123" not in out
|
|
|
|
# 2. Add filter +123 via stdin
|
|
sniffer.send_input("+123\n")
|
|
|
|
# 3. Send ID 123 again -> Should appear now
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "123#BB"], check=True)
|
|
out = sniffer.read_output(timeout=2.0)
|
|
assert "123" in out
|
|
|
|
def test_filtering_remove(bin_path, can_interface):
|
|
"""Test removing specific ID filter (-ID)."""
|
|
# Start normal mode (sniffs all)
|
|
with CanSnifferMonitor(bin_path, can_interface) as sniffer:
|
|
# 1. Send ID 456 -> Should appear
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "456#11"], check=True)
|
|
out = sniffer.read_output(timeout=1.0)
|
|
assert "456" in out
|
|
|
|
# 2. Remove 456 via stdin
|
|
sniffer.send_input("-456\n")
|
|
|
|
# Clear buffer to ignore previous output about 456
|
|
sniffer.clear_buffer()
|
|
|
|
# Send a different ID to force a screen refresh / activity on a visible ID
|
|
# This ensures cansniffer produces NEW output
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "789#AA"], check=True)
|
|
time.sleep(0.2)
|
|
|
|
# 3. Send ID 456 again with NEW unique data (CC)
|
|
# Using CC is safer than 22 as numbers might appear in timestamps
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "456#CC"], check=True)
|
|
|
|
# Read subsequent output
|
|
out = sniffer.read_output(timeout=1.5)
|
|
|
|
# Verify 789 is there (sanity check that we captured output)
|
|
assert "789" in out
|
|
|
|
# Verify 456 (and specifically its new data CC) is NOT in the new block.
|
|
assert "CC" not in out
|
|
|
|
def test_binary_mode_b(bin_path, can_interface):
|
|
"""Test binary mode output (-b)."""
|
|
with CanSnifferMonitor(bin_path, can_interface, args=["-b"]) as sniffer:
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "123#DEADBEEF"], check=True)
|
|
out = sniffer.read_output(timeout=1.0)
|
|
# In binary mode (-b), data bytes are often visualized as bits/dots.
|
|
# But the ID "123" should still be visible.
|
|
assert "123" in out
|
|
|
|
def test_color_mode_c(bin_path, can_interface):
|
|
"""Test color mode (-c)."""
|
|
# Color output depends on terminal capabilities usually, but cansniffer -c forces it.
|
|
with CanSnifferMonitor(bin_path, can_interface, args=["-c"]) as sniffer:
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "123#11"], check=True)
|
|
out = sniffer.read_output(timeout=1.0)
|
|
# Color mode uses ANSI escape sequences (e.g. \033[...m)
|
|
# Note: In Python string literals, \x1b is ESC.
|
|
assert "\x1b[" in out or "\033[" in out
|
|
|
|
def test_clear_screen_space(bin_path, can_interface):
|
|
"""Test clearing screen (<SPACE>)."""
|
|
with CanSnifferMonitor(bin_path, can_interface) as sniffer:
|
|
# Give it a moment to startup
|
|
time.sleep(0.5)
|
|
|
|
# Clear screen command
|
|
sniffer.send_input("\n") # Just enter sometimes works, or Space+Enter
|
|
sniffer.send_input(" \n")
|
|
|
|
# If it clears, it emits ANSI clear sequence "\033[2J" or "\033[H" (Home)
|
|
out = sniffer.read_output(timeout=1.0)
|
|
|
|
# Check for Common ANSI escape sequences for clearing/home
|
|
# \033[2J = Clear Screen, \033[H = Cursor Home
|
|
assert "\x1b[2J" in out or "\x1b[H" in out or "\x1b[" in out
|
|
|
|
def test_timeout_t(bin_path, can_interface):
|
|
"""Test timeout for ID display (-t)."""
|
|
# -t <time> in 10ms steps. e.g., -t 5 = 50ms (very short).
|
|
with CanSnifferMonitor(bin_path, can_interface, args=["-t", "5"]) as sniffer:
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "789#AA"], check=True)
|
|
out = sniffer.read_output(timeout=0.5)
|
|
assert "789" in out
|
|
|
|
# Wait for timeout (50ms + margin)
|
|
time.sleep(0.5)
|
|
|
|
# Send something else to trigger redraw
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "999#BB"], check=True)
|
|
out = sniffer.read_output(timeout=0.5)
|
|
|
|
# 789 should have timed out.
|
|
if "999" in out:
|
|
# If we see the update for 999, we expect 789 to NOT be present in this new block
|
|
pass
|
|
|
|
@pytest.mark.xfail(reason="cansniffer hides static data, so the ID disappears despite refresh")
|
|
def test_constant_data_refresh(bin_path, can_interface):
|
|
"""
|
|
Test if sending the SAME data repeatedly keeps the ID alive.
|
|
User suspects that identical data might not reset the timeout.
|
|
"""
|
|
# Set timeout to 500ms (-t 50)
|
|
with CanSnifferMonitor(bin_path, can_interface, args=["-t", "50"]) as sniffer:
|
|
# 1. Send initial packet
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "555#AA"], check=True)
|
|
out = sniffer.read_output(timeout=0.5)
|
|
assert "555" in out
|
|
|
|
# 2. Send IDENTICAL data repeatedly for 1.5 seconds (3x timeout)
|
|
# We send every 0.1s (well within 0.5s timeout)
|
|
start_time = time.time()
|
|
while time.time() - start_time < 1.5:
|
|
subprocess.run([os.path.join(bin_path, "cansend"), can_interface, "555#AA"], check=True)
|
|
time.sleep(0.1)
|
|
# Read output to prevent buffer overflow, though we are unbuffered
|
|
sniffer.read_output(timeout=0.01)
|
|
|
|
# 3. Check if it is STILL visible
|
|
out = sniffer.read_output(timeout=0.5)
|
|
|
|
# If the user is right (bug), '555' might be missing because content didn't change.
|
|
# If it works as expected (refresh on any frame), '555' should be there.
|
|
if "555" not in out:
|
|
pytest.fail("ID 555 disappeared despite constant refresh with identical data! Possible bug confirmed.")
|
|
|
|
assert "555" in out
|