diff --git a/test/case/hardware/all.yaml b/test/case/hardware/all.yaml index 1ca32dd3..4eaf24f8 100644 --- a/test/case/hardware/all.yaml +++ b/test/case/hardware/all.yaml @@ -7,3 +7,6 @@ - name: Watchdog reset on system lockup case: watchdog/test.py + +- name: GPS receiver basic test + case: gps_simple/test.py diff --git a/test/case/hardware/gps_simple/Readme.adoc b/test/case/hardware/gps_simple/Readme.adoc new file mode 120000 index 00000000..ae32c841 --- /dev/null +++ b/test/case/hardware/gps_simple/Readme.adoc @@ -0,0 +1 @@ +test.adoc \ No newline at end of file diff --git a/test/case/hardware/gps_simple/test.adoc b/test/case/hardware/gps_simple/test.adoc new file mode 100644 index 00000000..b8fc64db --- /dev/null +++ b/test/case/hardware/gps_simple/test.adoc @@ -0,0 +1,29 @@ +=== GPS receiver basic test + +ifdef::topdoc[:imagesdir: {topdoc}../../test/case/hardware/gps_simple] + +==== Description + +Verify that a simulated GPS receiver is detected and reports a valid +fix via the ietf-hardware operational datastore. + +The test injects NMEA sentences through a QEMU pipe chardev FIFO, +which appears as a virtio serial port inside the guest. + +==== Topology + +image::topology.svg[GPS receiver basic test topology, align=center, scaledwidth=75%] + +==== Sequence + +. Set up topology and attach to target DUT +. Configure GPS hardware component +. Verify GPS is activated +. Verify GPS has a fix +. Verify the position is near the coordinates you test with +. Save the configuration to startup configuration and reboot +. Verify GPS is activated +. Verify GPS has a fix +. Verify the position is near the coordinates you test with + + diff --git a/test/case/hardware/gps_simple/test.py b/test/case/hardware/gps_simple/test.py new file mode 100755 index 00000000..beb87267 --- /dev/null +++ b/test/case/hardware/gps_simple/test.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""GPS receiver basic test + +Verify that two simulated GPS receivers are detected and report valid +fixes via the ietf-hardware operational datastore. + +The test injects NMEA sentences through QEMU pipe chardev FIFOs, +which appear as virtio serial ports inside the guest. +""" +import infamy +import infamy.gps as gps +from infamy.util import until, wait_boot + +# Fun facts: The top of mount everest +test_lat = 27.9881 +test_lon = 86.9250 +test_alt = 8848.86 + + +def _near(a, b, tol): + return abs(a - b) <= tol + +def verify_position(target, name="gps0"): + state = gps.get_gps_state(target, name) + + try: + lat = float(state["latitude"]) + lon = float(state["longitude"]) + alt = float(state["altitude"]) + except (KeyError, TypeError, ValueError): + test.fail() + + if not _near(lat, test_lat, 0.01): + test.fail() + if not _near(lon, test_lon, 0.01): + test.fail() + if not _near(alt, test_alt, 100): + test.fail() + + try: + sat_used = int(state["satellites-used"]) + except (KeyError, TypeError, ValueError): + test.fail() + + if sat_used != 8: + test.fail() + +with infamy.Test() as test: + with test.step("Set up topology and attach to target DUT"): + env = infamy.Env() + target = env.attach("target", "mgmt") + + phys_name = env.ltop.mapping["target"][None] + + if not target.has_feature("infix-hardware", "gps"): + test.skip() + + # These are hacks, will only work on virtual devices + pipe0 = f"/tmp/{phys_name}-gps" + pipe1 = f"/tmp/{phys_name}-gps1" + + with gps.NMEAGenerator(pipe0, lat=test_lat, lon=test_lon, alt=test_alt), \ + gps.NMEAGenerator(pipe1, lat=test_lat, lon=test_lon, alt=test_alt): + + with test.step("Configure GPS hardware components"): + target.put_config_dicts({"ietf-hardware": { + "hardware": { + "component": [ + { + "name": "gps0", + "class": "infix-hardware:gps", + "infix-hardware:gps-receiver": {} + }, + { + "name": "gps1", + "class": "infix-hardware:gps", + "infix-hardware:gps-receiver": {} + } + ] + } + }}) + + with test.step("Verify both GPS receivers are activated"): + until(lambda: gps.is_activated(target, "gps0"), attempts=500) + until(lambda: gps.is_activated(target, "gps1"), attempts=500) + + with test.step("Verify both GPS receivers have a fix"): + until(lambda: gps.has_fix(target, "gps0"), attempts=60) + until(lambda: gps.has_fix(target, "gps1"), attempts=60) + + with test.step("Verify gps0 position is near the coordinates"): + verify_position(target, "gps0") + + with test.step("Verify gps1 position is near the coordinates"): + verify_position(target, "gps1") + + with test.step("Save the configuration to startup configuration and reboot"): + target.startup_override() + target.copy("running", "startup") + target.reboot() + if not wait_boot(target, env): + test.fail() + target = env.attach("target", "mgmt", test_reset=False) + + with test.step("Verify both GPS receivers are activated"): + until(lambda: gps.is_activated(target, "gps0"), attempts=500) + until(lambda: gps.is_activated(target, "gps1"), attempts=500) + + with test.step("Verify both GPS receivers have a fix"): + until(lambda: gps.has_fix(target, "gps0"), attempts=60) + until(lambda: gps.has_fix(target, "gps1"), attempts=60) + + with test.step("Verify gps0 position is near the coordinates"): + verify_position(target, "gps0") + + with test.step("Verify gps1 position is near the coordinates"): + verify_position(target, "gps1") + + test.succeed() diff --git a/test/case/hardware/gps_simple/topology.dot b/test/case/hardware/gps_simple/topology.dot new file mode 100644 index 00000000..e21cd6b3 --- /dev/null +++ b/test/case/hardware/gps_simple/topology.dot @@ -0,0 +1,22 @@ +graph "1x1" { + layout="neato"; + overlap="false"; + esep="+80"; + + node [shape=record, fontname="DejaVu Sans Mono, Book"]; + edge [color="cornflowerblue", penwidth="2", fontname="DejaVu Serif, Book"]; + + host [ + label="host | { mgmt }", + pos="0,12!", + requires="controller", + ]; + + target [ + label="{ mgmt } | target", + pos="10,12!", + requires="infix gps", + ]; + + host:mgmt -- target:mgmt [requires="mgmt", color="lightgray"] +} diff --git a/test/case/hardware/gps_simple/topology.svg b/test/case/hardware/gps_simple/topology.svg new file mode 100644 index 00000000..6fc6f47a --- /dev/null +++ b/test/case/hardware/gps_simple/topology.svg @@ -0,0 +1,33 @@ + + + + + + +1x1 + + + +host + +host + +mgmt + + + +target + +mgmt + +target + + + +host:mgmt--target:mgmt + + + + diff --git a/test/infamy/gps.py b/test/infamy/gps.py new file mode 100644 index 00000000..d3743a50 --- /dev/null +++ b/test/infamy/gps.py @@ -0,0 +1,210 @@ +"""GPS/NMEA test helpers + +Provides an NMEAGenerator class that writes NMEA sentences to a QEMU +pipe chardev FIFO, simulating a GPS receiver. Also provides helpers +for querying GPS operational state via YANG. +""" + +import os +import errno +import threading +import time +import pynmea2 +class NMEAGenerator: + """Write NMEA sentences to a QEMU pipe chardev FIFO. + + Sends a full cycle of NMEA sentences (like a real u-blox receiver) + continuously at 1 Hz in a background thread. + + The pipe_path should be the base path (without .in/.out suffix), + matching the QEMU ``-chardev pipe,path=...`` argument. + + Usage:: + + with NMEAGenerator("/tmp/node-gps") as nmea: + # NMEA data is being sent in background + time.sleep(10) + """ + + def __init__(self, pipe_path, lat=48.1173, lon=11.5167, alt=545.4): + self.pipe_path = pipe_path + self._fd = -1 + self.lat = lat + self.lon = lon + self.alt = alt + self._thread = None + self._stop = threading.Event() + + def __enter__(self): + self.start() + return self + def __exit__(self, _, __, ___): + self.close() + + def start(self): + self._stop.clear() + self._thread = threading.Thread(target=self._send_loop, daemon=True) + self._thread.start() + print(f"NMEA: started sender thread for {self.pipe_path}") + + def close(self): + self._stop.set() + if self._thread: + self._thread.join(timeout=5) + self._thread = None + if self._fd >= 0: + try: + os.close(self._fd) + except OSError: + pass + self._fd = -1 + + def _send_loop(self): + print(f"NMEA: send_loop entered for {self.pipe_path}") + while not self._stop.is_set(): + try: + self._fd = os.open(f"{self.pipe_path}.in", + os.O_WRONLY | os.O_NONBLOCK) + print(f"NMEA: opened {self.pipe_path}.in (fd={self._fd})") + except OSError as e: + if e.errno in (errno.ENXIO, errno.ENOENT): + print(f"NMEA: {self.pipe_path}.in not ready ({e}), retrying ...") + self._stop.wait(0.5) + continue + print(f"NMEA: {self.pipe_path}.in open failed: {e}") + raise + cycles = 0 + while not self._stop.is_set(): + try: + self._send_cycle() + cycles += 1 + if cycles <= 3 or cycles % 10 == 0: + print(f"NMEA: {self.pipe_path} sent cycle {cycles}") + except BlockingIOError: + print(f"NMEA: {self.pipe_path} write blocked (cycle {cycles})") + except OSError as e: + print(f"NMEA: {self.pipe_path} write error: {e}, reconnecting") + break + self._stop.wait(1) + try: + os.close(self._fd) + except OSError: + pass + self._fd = -1 + + def _send(self, sentence): + data = (str(sentence) + "\r\n").encode() + try: + os.write(self._fd, data) + except BlockingIOError: + pass + + def _send_cycle(self): + """Send a full NMEA cycle matching real u-blox GPS output.""" + now = time.gmtime() + utc = time.strftime("%H%M%S.00", now) + date = time.strftime("%d%m%y", now) + + lat_deg = int(abs(self.lat)) + lat_min = (abs(self.lat) - lat_deg) * 60 + lat_str = f"{lat_deg:02d}{lat_min:07.4f}" + lat_ns = "N" if self.lat >= 0 else "S" + + lon_deg = int(abs(self.lon)) + lon_min = (abs(self.lon) - lon_deg) * 60 + lon_str = f"{lon_deg:03d}{lon_min:07.4f}" + lon_ew = "E" if self.lon >= 0 else "W" + + # RMC - Recommended Minimum + self._send(pynmea2.RMC("GP", "RMC", ( + utc, "A", + lat_str, lat_ns, + lon_str, lon_ew, + "0.0", "0.0", + date, + "0.0", "E", + "A", + ))) + + # VTG - Track Made Good and Ground Speed + self._send(pynmea2.VTG("GP", "VTG", ( + "0.0", "T", + "", "M", + "0.0", "N", + "0.0", "K", + "A", + ))) + + # GGA - Fix Data + self._send(pynmea2.GGA("GP", "GGA", ( + utc, + lat_str, lat_ns, + lon_str, lon_ew, + "1", "08", "0.9", + f"{self.alt:.1f}", "M", + "47.0", "M", + "", "", + ))) + + # GSA - DOP and Active Satellites (3D fix, 8 sats) + self._send(pynmea2.GSA("GP", "GSA", ( + "A", "3", + "01", "02", "03", "04", "05", "06", "07", "08", + "", "", "", "", + "1.5", "0.9", "1.2", + ))) + + # GSV - Satellites in View (4 sats per message, 2 messages) + self._send(pynmea2.GSV("GP", "GSV", ( + "2", "1", "08", + "01", "45", "045", "40", + "02", "30", "090", "38", + "03", "60", "135", "42", + "04", "15", "180", "35", + ))) + self._send(pynmea2.GSV("GP", "GSV", ( + "2", "2", "08", + "05", "50", "225", "41", + "06", "25", "270", "36", + "07", "70", "315", "44", + "08", "20", "000", "33", + ))) + + # GLL - Geographic Position + self._send(pynmea2.GLL("GP", "GLL", ( + lat_str, lat_ns, + lon_str, lon_ew, + utc, + "A", + "A", + ))) + +def _get_hardware(target): + data = target.get_data("/ietf-hardware:hardware") + if not data or "hardware" not in data: + return {} + return data["hardware"] + + +def get_gps_state(target, name="gps0"): + """Get GPS receiver operational state for a named component.""" + hardware = _get_hardware(target) + for component in hardware.get("component", []): + if component.get("name") == name: + return component.get("infix-hardware:gps-receiver", + component.get("gps-receiver")) + return None + + +def is_activated(target, name="gps0"): + """Check if gpsd has activated the GPS device.""" + state = get_gps_state(target, name) + return state.get("activated", False) if state else False + + +def has_fix(target, name="gps0"): + """Check if GPS reports a fix (2D or 3D).""" + state = get_gps_state(target, name) + if not state: + return False + return state.get("fix-mode") in ("2d", "3d") diff --git a/test/infamy/transport.py b/test/infamy/transport.py index 0d0db27c..e660e3e4 100644 --- a/test/infamy/transport.py +++ b/test/infamy/transport.py @@ -79,6 +79,13 @@ class Transport(ABC): """Check if the device has the given YANG model loaded.""" return model_name in self.modules + def has_feature(self, model_name, feature_name): + """Check if a specific feature is enabled on the device for a given YANG model.""" + if model_name not in self.modules: + return False + features = self.modules[model_name].get("feature", []) + return feature_name in features + def reachable(self): """Check if the device reachable on ll6""" neigh = ll6ping(self.location.interface, flags=["-w1", "-c1", "-L", "-n"]) diff --git a/test/templates/infix-bios-x86_64.mustache b/test/templates/infix-bios-x86_64.mustache index 09a87e0c..033023bf 100644 --- a/test/templates/infix-bios-x86_64.mustache +++ b/test/templates/infix-bios-x86_64.mustache @@ -15,6 +15,11 @@ truncate -s 0 $imgdir/{{name}}.mactab echo "{{qn_name}} {{qn_mac}}" >>$imgdir/{{name}}.mactab {{/links}} +{{#qn_gps}} +mkfifo /tmp/{{name}}-gps.in /tmp/{{name}}-gps.out 2>/dev/null +mkfifo /tmp/{{name}}-gps1.in /tmp/{{name}}-gps1.out 2>/dev/null +{{/qn_gps}} + exec qemu-system-x86_64 -M pc,accel=kvm:tcg -cpu max \ -m {{#qn_mem}}{{qn_mem}}{{/qn_mem}}{{^qn_mem}}256M{{/qn_mem}} \ {{> ../qeneth/templates/inc/qemu-links}} @@ -24,3 +29,9 @@ exec qemu-system-x86_64 -M pc,accel=kvm:tcg -cpu max \ $usb_cmd \ {{> ../qeneth/templates/inc/infix-common}} {{> ../qeneth/templates/inc/qemu-console}} +{{#qn_gps}} + -chardev pipe,id=gps0,path=/tmp/{{name}}-gps \ + -device virtserialport,chardev=gps0,name=gps0 \ + -chardev pipe,id=gps1,path=/tmp/{{name}}-gps1 \ + -device virtserialport,chardev=gps1,name=gps1 \ +{{/qn_gps}} diff --git a/test/virt/quad/topology.dot.in b/test/virt/quad/topology.dot.in index 609cdced..a0c363fa 100644 --- a/test/virt/quad/topology.dot.in +++ b/test/virt/quad/topology.dot.in @@ -23,11 +23,12 @@ graph "quad" { dut1 [ label="{ e1 | e2 | e3 | e4 } | dut1 | { e5 | e6 | e7 | e8}", pos="10,30!", - provides="infix watchdog", + provides="infix watchdog gps", expected_boot="primary", qn_console=9001, qn_mem="384M", - qn_usb="dut1.usb" + qn_usb="dut1.usb", + qn_gps="true" ]; dut2 [ label="{ e1 | e2 | e3 | e4 } | dut2 | { e5 | e6 | e7 | e8}",