Skip to content

Support for libremesh wifi simulation #212

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
francoriba wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Support for libremesh wifi simulation #212

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
029ec5c
targets: add qemu-x86_64 specific for LibreMesh
francoriba Jan 18, 2026
194b4c9
Makefile: enable qemu-libremesh-x86_64 testing
francoriba Jan 18, 2026
ce41289
conftest: add vwifi support for virtualized mesh testing
francoriba Jan 18, 2026
a73fd99
tests: add LibreMesh shared-state-async tests
francoriba Jan 18, 2026
2f0d969
refactor: apply formatting
francoriba Jan 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,24 @@ $(curdir)/x86-64:
--lg-env $(TESTSDIR)/targets/qemu_x86-64.yaml \
--firmware $(FIRMWARE:.gz=)

# LibreMesh x86-64 target with vwifi support for mesh simulation
$(curdir)/x86-64-libremesh: QEMU_BIN ?= qemu-system-x86_64
$(curdir)/x86-64-libremesh: FIRMWARE ?= $(TOPDIR)/bin/targets/x86/64/libremesh-x86-64-generic-squashfs-combined.img.gz
$(curdir)/x86-64-libremesh:

[ -f $(FIRMWARE) ]

gzip \
--force \
--keep \
--decompress \
$(FIRMWARE) || true

LG_QEMU_BIN=$(QEMU_BIN) \
$(pytest) \
--lg-env $(TESTSDIR)/targets/qemu_libremesh-x86-64.yaml \
--firmware $(FIRMWARE:.gz=)

$(curdir)/armsr-armv8: QEMU_BIN ?= qemu_system-aarch64
$(curdir)/armsr-armv8: FIRMWARE ?= $(TOPDIR)/bin/targets/armsr/armv8/openwrt-armsr-armv8-generic-initramfs-kernel.bin
$(curdir)/armsr-armv8:
Expand Down
45 changes: 45 additions & 0 deletions targets/qemu_libremesh-x86-64.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
# QEMU x86-64 target for LibreMesh testing with virtualized WiFi (vwifi/mac80211_hwsim)
#
# This target enables testing LibreMesh features in a virtualized mesh network.
# It requires:
# - A LibreMesh firmware image built for x86-64
# - vwifi-server running on the host
# - vwifi-client binary to be uploaded to the VM
#
# Usage:
# make tests/x86-64-libremesh FIRMWARE=/path/to/libremesh-x86-64.img

targets:
main:
features: [hwsim, libremesh]
resources:
- NetworkService:
# The actual address will be filled in by the strategy
address: ""
port: 22
username: root

drivers:
- QEMUDriver:
qemu_bin: qemu_bin
machine: pc
cpu: max
memory: 128M
extra_args: "-device virtio-rng-pci -netdev user,id=wan -device virtio-net-pci,netdev=wan"
nic: "user,model=virtio-net-pci,net=10.13.0.0/16,id=lan"
disk: firmware
- ShellDriver:
login_prompt: Please press Enter to activate this console.
prompt: 'root@'
await_login_timeout: 30
username: root
- SSHDriver:
username: root
explicit_scp_mode: True
- QEMUNetworkStrategy: {}

tools:
qemu_bin: qemu-system-x86_64

imports:
- ../strategies/qemunetworkstrategy.py
78 changes: 78 additions & 0 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@

import json
import logging
import re
import shlex
import subprocess
import time
from os import getenv

import pytest
Expand Down Expand Up @@ -63,3 +67,77 @@ def shell_command(strategy):
def ssh_command(shell_command, target):
ssh = target.get_driver("SSHDriver")
return ssh


def _host_ipv4_from_hostname_I() -> str:
"""Get the host's IPv4 address using hostname -I.

This is needed for vwifi-client in the VM to connect back to the
vwifi-server running on the host.
"""
out = subprocess.check_output("hostname -I", shell=True, text=True).strip()
if not out:
raise RuntimeError("hostname -I returned nothing")
# take the first token; if it's not IPv4, fall back to first IPv4 token
first = out.split()[0]
if ":" in first:
first = next(
(t for t in out.split() if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", t)), ""
)
if not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", first or ""):
raise RuntimeError(f"Could not determine IPv4 from: {out!r}")
return first


@pytest.fixture
def upload_vwifi(shell_command, target):
"""Upload vwifi-client to the VM and connect to the virtualized mesh network.

This fixture:
1. Uploads vwifi-client binary to the target
2. Configures mac80211_hwsim for virtual WiFi interfaces
3. Starts vwifi-client to connect to the host's vwifi-server
4. Waits for mesh interfaces to come up and establish connections

Prerequisites:
- vwifi-server must be running on the host
- vwifi/vwifi-client binary must exist in the test directory
"""
ssh = target.get_driver("SSHDriver")
ssh.scp(src="vwifi/vwifi-client", dst=":/usr/bin/vwifi-client")
path = "\n".join(ssh.run("which vwifi-client")[0])
assert path == "/usr/bin/vwifi-client"

# compute HOST IPv4 once (on the host)
host_ip = _host_ipv4_from_hostname_I()
host_ip_q = shlex.quote(host_ip)

ssh.run_check("rmmod mac80211_hwsim")
ssh.run_check("insmod mac80211_hwsim radios=0")
cmd = f"""sh -lc '
if command -v start-stop-daemon >/dev/null; then
start-stop-daemon -S -b -m -p /tmp/vwifi.pid \
-x /usr/bin/vwifi-client -- {host_ip_q} --number 2 \
>/tmp/vwifi.log 2>&1
else
nohup /usr/bin/vwifi-client {host_ip_q} --number 2 \
</dev/null >/tmp/vwifi.log 2>&1 & echo $! >/tmp/vwifi.pid
fi
'"""
ssh.run_check(cmd)
assert "\n".join(ssh.run("ps | grep vwifi")[0]) != ""
time.sleep(5)
ssh.run("wifi reload")
ssh.run("wifi up")
time.sleep(10)
phy_devices = ssh.run("iw phy | grep phy")[0]
assert len(phy_devices) == 4 # labgrid tokenizes on \t
iw_devices = "\n".join(ssh.run("iw dev")[0])
while "wlan0-mesh" not in iw_devices:
iw_devices = "\n".join(ssh.run("iw dev")[0])
time.sleep(2)
stations = "\n".join(ssh.run("iw dev wlan0-mesh station dump")[0])
assert "02:00:00:00:00:01" in stations
assert "02:00:00:00:00:02" in stations
assert "02:00:00:00:00:03" in stations
return ssh
96 changes: 96 additions & 0 deletions tests/test_libremesh_shared_state_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
# LibreMesh shared-state-async tests using virtualized mesh (vwifi)
#
# These tests verify that LibreMesh's shared-state-async mechanism works
# correctly in a virtualized mesh network environment.
#
# Prerequisites:
# - vwifi-server running on the host with multiple simulated nodes
# - LibreMesh firmware image with shared-state-async package
# - vwifi-client binary available in vwifi/ directory
#
# Usage:
# pytest tests/test_libremesh_shared_state_async.py -k libremesh

import json
import logging
import time

import pytest

# Expected node hostnames and MAC addresses in the virtualized mesh
N1 = "LiMe-000001"
N2 = "LiMe-000002"
N3 = "LiMe-000003"
N1234 = "LiMe-123456"
MAC1 = "02:58:47:00:00:01"
MAC2 = "02:58:47:00:00:02"
MAC3 = "02:58:47:00:00:03"
MAC1234 = "02:58:47:12:34:56"


logger = logging.getLogger(__name__)


def _join_stdout(stdout):
"""Join stdout lines into a single string."""
if isinstance(stdout, (list, tuple)):
return "\n".join(stdout)
return stdout or ""


def _extract_json_from_mixed(text):
"""Extract JSON object from mixed text output."""
i = text.find("{")
j = text.rfind("}")
assert i != -1 and j != -1 and j > i, f"Could not find JSON in output:\n{text}"
return json.loads(text[i : j + 1])


def _strip_mac(mac):
"""Remove colons from MAC address and lowercase."""
return mac.replace(":", "").lower()


def _canonical_link_key(mac_a, mac_b):
"""Create canonical link key from two MAC addresses."""
a = _strip_mac(mac_a)
b = _strip_mac(mac_b)
return "".join(sorted([a, b]))


@pytest.mark.lg_feature("libremesh")
def test_bat_links_info(upload_vwifi):
"""Test that shared-state-async bat_links_info shows all mesh nodes and links.

This test verifies that:
1. shared-state-async can publish and sync bat_links_info
2. All expected mesh nodes are visible in the shared state
3. Links between the test node and other nodes are recorded
"""
ssh_command = upload_vwifi
link_key_N1234_N1 = _canonical_link_key(MAC1234, MAC1)
link_key_N1234_N2 = _canonical_link_key(MAC1234, MAC2)
link_key_N1234_N3 = _canonical_link_key(MAC1234, MAC3)

ssh_command.run_check("shared-state-async-publish-all")
ssh_command.run_check("shared-state-async sync bat_links_info")
time.sleep(15)
out, err, rc = ssh_command.run("shared-state-async get bat_links_info")
assert rc == 0, f"shared-state-async failed (rc={rc}) stderr={_join_stdout(err)}"
data = _extract_json_from_mixed(_join_stdout(out))

assert isinstance(data, dict) and data, "bat_links_info must be a non-empty dict"
logger.warning(out)
assert N1234 in data, f"Expected {N1234} in shared-state keys: {list(data.keys())}"
assert N1 in data, f"Expected {N1} in shared-state keys: {list(data.keys())}"
assert N2 in data, f"Expected {N2} in shared-state keys: {list(data.keys())}"
assert N3 in data, f"Expected {N3} in shared-state keys: {list(data.keys())}"
assert link_key_N1234_N1 in data[N1234]["links"], (
f"Expected {link_key_N1234_N1} in shared-state keys: {list(data[N1234]['links'])}"
)
assert link_key_N1234_N2 in data[N1234]["links"], (
f"Expected {link_key_N1234_N2} in shared-state keys: {list(data[N1234]['links'])}"
)
assert link_key_N1234_N3 in data[N1234]["links"], (
f"Expected {link_key_N1234_N3} in shared-state keys: {list(data[N1234]['links'])}"
)
Loading