Compare commits

...

4 Commits

Author SHA1 Message Date
ruv 5402b070f6 docs: add ADR-051 sensing server decomposition plan (Sprint 2)
14-module extraction plan for 3,765-line main.rs god object.
6 phases, each independently committable and testable.
Target: no file over 500 lines, AppStateInner split into domain sub-states.

Refs #174

Co-Authored-By: claude-flow <ruv@ruv.net>
2026-03-06 15:18:52 -05:00
rUv 7659b0bbe2 feat: cross-platform WiFi collector factory (ADR-049) (#173)
feat: cross-platform WiFi collector factory (ADR-049)
2026-03-06 15:10:26 -05:00
ruv 75d4685d25 feat: cross-platform WiFi collector factory with graceful degradation (ADR-049)
- Add create_collector() factory function that auto-detects platform and never raises
- Add LinuxWifiCollector.is_available() classmethod for probe-without-exception
- Refactor ws_server.py to use create_collector(), removing ~30 lines of duplicated platform detection
- Add 10 unit tests covering all platform paths and edge cases
- Add ADR-049 documenting the cross-platform detection and fallback chain

Docker, WSL, and headless users now get SimulatedCollector automatically
with a clear WARNING log instead of a RuntimeError crash.

Closes #148
Closes #155

Co-Authored-By: claude-flow <ruv@ruv.net>
2026-03-06 15:09:32 -05:00
rUv 45c15b77a5 fix: ADR-050 security hardening — HMAC, path traversal, OTA auth (#172)
fix: ADR-050 security hardening — HMAC, path traversal, OTA auth
2026-03-06 14:02:50 -05:00
5 changed files with 475 additions and 60 deletions
@@ -0,0 +1,122 @@
# ADR-049: Cross-Platform WiFi Interface Detection and Graceful Degradation
| Field | Value |
|-------|-------|
| Status | Proposed |
| Date | 2026-03-06 |
| Deciders | ruv |
| Depends on | ADR-013 (Feature-Level Sensing), ADR-025 (macOS CoreWLAN) |
| Issue | [#148](https://github.com/ruvnet/wifi-densepose/issues/148) |
## Context
Users report `RuntimeError: Cannot read /proc/net/wireless` when running WiFi DensePose in environments where the Linux wireless proc filesystem is unavailable:
- **Docker containers** on macOS/Windows (Linux kernel detected, but no wireless subsystem)
- **WSL2** without USB WiFi passthrough
- **Headless Linux servers** without WiFi hardware
- **Embedded Linux** boards without wireless-extensions support
The current architecture has two layers of defense:
1. **`ws_server.py`** (line 345-355) checks `os.path.exists("/proc/net/wireless")` before instantiating `LinuxWifiCollector` and falls back to `SimulatedCollector` if missing.
2. **`rssi_collector.py`** `LinuxWifiCollector._validate_interface()` (line 178-196) raises a hard `RuntimeError` if `/proc/net/wireless` is missing or the interface isn't listed.
However, there are gaps:
- **Direct usage**: Any code that instantiates `LinuxWifiCollector` directly (outside `ws_server.py`) hits the unguarded `RuntimeError` with no fallback.
- **Error message**: The RuntimeError message tells users to "use SimulatedCollector instead" but doesn't explain how.
- **No auto-detection**: The collector selection logic is duplicated between `ws_server.py` and `install.sh` with no shared platform-detection utility.
- **Partial `/proc/net/wireless`**: The file may exist (e.g., kernel module loaded) but contain no interfaces, producing a confusing "interface not found" error instead of a clean fallback.
## Decision
### 1. Platform-Aware Collector Factory
Introduce a `create_collector()` factory function in `rssi_collector.py` that encapsulates the platform detection and fallback chain:
```python
def create_collector(
preferred: str = "auto",
interface: str = "wlan0",
sample_rate_hz: float = 10.0,
) -> BaseCollector:
"""
Create the best available WiFi collector for the current platform.
Resolution order (when preferred="auto"):
1. ESP32 CSI (if UDP port 5005 is receiving frames)
2. Platform-native WiFi:
- Linux: LinuxWifiCollector (requires /proc/net/wireless + active interface)
- Windows: WindowsWifiCollector (netsh wlan)
- macOS: MacosWifiCollector (CoreWLAN)
3. SimulatedCollector (always available)
Raises nothing — always returns a usable collector.
"""
```
### 2. Soft Validation in LinuxWifiCollector
Replace the hard `RuntimeError` in `_validate_interface()` with a class method that returns availability status without raising:
```python
@classmethod
def is_available(cls, interface: str = "wlan0") -> tuple[bool, str]:
"""Check if Linux WiFi collection is possible. Returns (available, reason)."""
if not os.path.exists("/proc/net/wireless"):
return False, "/proc/net/wireless not found (Docker, WSL, or no wireless subsystem)"
with open("/proc/net/wireless") as f:
content = f.read()
if interface not in content:
names = cls._parse_interface_names(content)
return False, f"Interface '{interface}' not in /proc/net/wireless. Available: {names}"
return True, "ok"
```
The existing `_validate_interface()` continues to raise `RuntimeError` for direct callers who need fail-fast behavior, but `create_collector()` uses `is_available()` to probe without exceptions.
### 3. Structured Fallback Logging
When auto-detection skips a collector, log at `WARNING` level with actionable context:
```
WiFi collector: LinuxWifiCollector unavailable (/proc/net/wireless not found — likely Docker/WSL).
WiFi collector: Falling back to SimulatedCollector. For real sensing, connect ESP32 nodes via UDP:5005.
```
### 4. Consolidate Platform Detection
Remove duplicated platform-detection logic from `ws_server.py` and `install.sh`. Both should use `create_collector()` (Python) or a shared `detect_wifi_platform()` shell function.
## Consequences
### Positive
- **Zero-crash startup**: `create_collector("auto")` never raises — Docker, WSL, and headless users get `SimulatedCollector` automatically with a clear log message.
- **Single detection path**: Platform logic lives in one place (`rssi_collector.py`), reducing drift between `ws_server.py`, `install.sh`, and future entry points.
- **Better DX**: Error messages explain *why* a collector is unavailable and *what to do* (connect ESP32, install WiFi driver, etc.).
### Negative
- **SimulatedCollector may mask hardware issues**: Users with real WiFi hardware that fails detection might unknowingly run on simulated data. Mitigated by the `WARNING`-level log.
- **Breaking change for direct `LinuxWifiCollector` callers**: Code that catches `RuntimeError` from `_validate_interface()` as a signal needs to migrate to `is_available()` or `create_collector()`. This is a minor change — there are no known external consumers.
### Neutral
- `_validate_interface()` behavior is unchanged for existing direct callers — this is additive.
## Implementation Notes
1. Add `create_collector()` and `BaseCollector.is_available()` to `v1/src/sensing/rssi_collector.py`
2. Refactor `ws_server.py` `_init_collector()` to call `create_collector()`
3. Update `install.sh` `detect_wifi_hardware()` to use shared detection logic
4. Add unit tests for each platform path (mock `/proc/net/wireless` presence/absence)
5. Comment on issue #148 with the fix
## References
- Issue #148: RuntimeError: Cannot read /proc/net/wireless
- ADR-013: Feature-Level Sensing on Commodity Gear
- ADR-025: macOS CoreWLAN WiFi Sensing
- [Linux /proc/net/wireless documentation](https://www.kernel.org/doc/html/latest/networking/statistics.html)
@@ -0,0 +1,109 @@
# ADR-051: Sensing Server Decomposition — main.rs God Object Breakup
| Field | Value |
|-------|-------|
| Status | Proposed |
| Date | 2026-03-06 |
| Deciders | ruv |
| Depends on | ADR-050 (Quality Engineering — Sprint 2) |
| Issue | [#174](https://github.com/ruvnet/RuView/issues/174) |
## Context
`sensing-server/src/main.rs` is 3,765 lines with cyclomatic complexity ~65. It contains 12 structs, 60+ functions, 10 constants, and a 37-field `AppStateInner` god object. This violates the project's 500-line file limit (CLAUDE.md) and makes unit testing individual components impossible.
The file mixes concerns:
- CLI argument parsing and server bootstrap
- HTTP route handlers (health, models, recordings, training, pose, vitals)
- WebSocket upgrade and client management
- UDP CSI frame receiver and parser
- Signal processing pipeline (feature extraction, classification, smoothing)
- Simulated data generator
- Windows WiFi scanning integration
- Pose estimation from WiFi signals
- Vital sign smoothing and filtering
- Model/recording file management
## Decision
Decompose `main.rs` into 14 focused modules. Each module owns its types, constants, and functions. `main.rs` retains only CLI parsing, state initialization, router construction, and server startup (~250 lines).
### Module Extraction Plan
| Module | Source Lines | Contents | Target Size |
|--------|-------------|----------|-------------|
| `cli.rs` | 59-152 | `Args` struct, CLI parsing | ~100 |
| `state.rs` | 154-370 | `AppStateInner`, all DTOs (`Esp32Frame`, `SensingUpdate`, `NodeInfo`, etc.), `SharedState` type alias | ~220 |
| `signal.rs` | 542-890 | `generate_signal_field()`, `estimate_breathing_rate_hz()`, `compute_subcarrier_variances()`, `extract_features_from_frame()`, `raw_classify()` | ~350 |
| `smoothing.rs` | 886-1060 | Classification smoothing, vital sign smoothing, `trimmed_mean()`, constants | ~180 |
| `routes_health.rs` | 1660-2005 | `/health/*`, `/api/v1/info` endpoints | ~350 |
| `routes_model.rs` | 2058-2230 | `/api/v1/models/*`, LoRA profiles, `scan_model_files()` | ~180 |
| `routes_recording.rs` | 2233-2440 | `/api/v1/recording/*`, `scan_recording_files()` | ~210 |
| `routes_training.rs` | 2443-2560 | `/api/v1/train/*`, `/api/v1/adaptive/*` | ~120 |
| `routes_sensing.rs` | 2562-2710 | Vital signs, edge vitals, WASM events, model info, SONA endpoints | ~150 |
| `routes_pose.rs` | 1701-1930, 2007-2055 | Pose estimation, `derive_single_person_pose()`, pose/stats/zones endpoints | ~280 |
| `websocket.rs` | 1492-1660 | WS upgrade handlers, `handle_ws_client()`, `handle_ws_pose_client()` | ~170 |
| `udp_receiver.rs` | 2725-2890 | UDP CSI frame receiver task, frame parsing | ~170 |
| `data_sources.rs` | 1063-1465, 2888-3020 | Windows WiFi task, simulated data task, `probe_windows_wifi()`, `parse_netsh_interfaces_output()` | ~400 |
| `router.rs` | (new) | `build_router()` function assembling all routes | ~80 |
### Extraction Order (6 Phases)
1. **Phase 1**: `cli.rs` + `state.rs` — Zero behavioral change, just move types
2. **Phase 2**: `signal.rs` + `smoothing.rs` — Pure functions, easy to test
3. **Phase 3**: `routes_health.rs` + `routes_model.rs` + `routes_recording.rs` — Stateless-ish handlers
4. **Phase 4**: `routes_training.rs` + `routes_sensing.rs` + `routes_pose.rs` — Remaining HTTP handlers
5. **Phase 5**: `websocket.rs` + `udp_receiver.rs` + `data_sources.rs` — Async tasks
6. **Phase 6**: `router.rs` — Assemble all routes, slim `main.rs` to ~250 lines
### State Refactoring
`AppStateInner` (37 fields) will be split into domain-specific sub-states:
```rust
pub struct AppStateInner {
pub config: ServerConfig, // CLI args, ports, paths
pub sensing: SensingState, // CSI frames, features, classification
pub vitals: VitalsState, // Vital sign buffers, smoothing state
pub models: ModelState, // Active model, discovered models, LoRA
pub recording: RecordingState, // Active recording, file handles
pub training: TrainingState, // Training status, adaptive model
pub pose: PoseState, // Person detections, pose history
pub broadcast_tx: broadcast::Sender<SensingUpdate>,
}
```
## Consequences
### Positive
- Each module is independently unit-testable
- No file exceeds 500 lines
- Domain boundaries are explicit (state sub-structs)
- New developers can find code by domain
- Merge conflict surface reduced (parallel module edits)
### Negative
- Large refactor with ~3,700 lines touched — high merge conflict risk
- `pub(crate)` visibility needed for cross-module state access
- Some functions share mutable state, requiring careful `&mut` threading
### Neutral
- No behavioral change — all endpoints, WebSocket, UDP behavior stays identical
- Existing integration tests (if any) continue to pass unchanged
## Implementation Notes
1. Each phase is a separate commit for easy revert
2. Run `cargo test` and `cargo check` after each phase
3. Use `pub(crate)` for internal types, keep public API surface minimal
4. Add `#[cfg(test)] mod tests` to each new module with at least smoke tests
5. Consider adding `tower` middleware for auth (Sprint 1 remaining item) during Phase 3
## References
- ADR-050: Quality Engineering Response (Sprint 2 plan)
- Issue #170: Quality Engineering Analysis
- CLAUDE.md: 500-line file limit rule
+123 -18
View File
@@ -12,13 +12,15 @@ from __future__ import annotations
import logging
import math
import os
import platform
import re
import subprocess
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, List, Optional, Protocol
from typing import Deque, List, Optional, Protocol, Union
import numpy as np
@@ -173,27 +175,47 @@ class LinuxWifiCollector:
"""Collect a single sample right now (blocking)."""
return self._read_sample()
# -- availability check --------------------------------------------------
@classmethod
def is_available(cls, interface: str = "wlan0") -> tuple[bool, str]:
"""Check if Linux WiFi collection is possible without raising.
Returns
-------
(available, reason) : tuple[bool, str]
``available`` is True when /proc/net/wireless exists and lists
the requested interface. ``reason`` is a human-readable
explanation when unavailable.
"""
if not os.path.exists("/proc/net/wireless"):
return False, (
"/proc/net/wireless not found. "
"This environment has no Linux wireless subsystem "
"(common in Docker, WSL, or headless servers)."
)
try:
with open("/proc/net/wireless", "r") as f:
content = f.read()
except OSError as exc:
return False, f"Cannot read /proc/net/wireless: {exc}"
if interface not in content:
names = cls._parse_interface_names(content)
return False, (
f"Interface '{interface}' not listed in /proc/net/wireless. "
f"Available: {names or '(none)'}. "
f"Ensure the interface is up and associated with an AP."
)
return True, "ok"
# -- internals -----------------------------------------------------------
def _validate_interface(self) -> None:
"""Check that the interface exists on this machine."""
try:
with open("/proc/net/wireless", "r") as f:
content = f.read()
if self._interface not in content:
raise RuntimeError(
f"WiFi interface '{self._interface}' not found in "
f"/proc/net/wireless. Available interfaces may include: "
f"{self._parse_interface_names(content)}. "
f"Ensure the interface is up and associated with an AP."
)
except FileNotFoundError:
raise RuntimeError(
"Cannot read /proc/net/wireless. "
"This collector requires a Linux system with wireless-extensions support. "
"If running in a container or VM without WiFi hardware, use "
"SimulatedCollector instead."
)
available, reason = self.is_available(self._interface)
if not available:
raise RuntimeError(reason)
@staticmethod
def _parse_interface_names(proc_content: str) -> List[str]:
@@ -736,3 +758,86 @@ class MacosWifiCollector:
if self._running:
logger.error("macOS WiFi utility exited unexpectedly. Collector stopped.")
self._running = False
# ---------------------------------------------------------------------------
# Collector factory (ADR-049)
# ---------------------------------------------------------------------------
CollectorType = Union[LinuxWifiCollector, WindowsWifiCollector, MacosWifiCollector, SimulatedCollector]
def create_collector(
preferred: str = "auto",
interface: str = "wlan0",
sample_rate_hz: float = 10.0,
) -> CollectorType:
"""Create the best available WiFi collector for the current platform.
Resolution order (when ``preferred="auto"``):
1. Platform-native WiFi:
- Linux: LinuxWifiCollector (requires /proc/net/wireless + active interface)
- Windows: WindowsWifiCollector (netsh wlan)
- macOS: MacosWifiCollector (CoreWLAN)
2. SimulatedCollector (always available)
This function never raises -- it always returns a usable collector.
Parameters
----------
preferred : str
``"auto"`` for platform detection, or one of ``"linux"``,
``"windows"``, ``"macos"``, ``"simulated"`` to force a specific
collector.
interface : str
WiFi interface name (Linux/Windows only).
sample_rate_hz : float
Target sampling rate.
"""
_VALID_PREFERRED = {"auto", "linux", "windows", "macos", "simulated"}
if preferred not in _VALID_PREFERRED:
logger.warning(
"WiFi collector: unknown preferred=%r (valid: %s). Falling back to auto.",
preferred, ", ".join(sorted(_VALID_PREFERRED)),
)
preferred = "auto"
system = platform.system()
if preferred == "auto":
if system == "Linux":
available, reason = LinuxWifiCollector.is_available(interface)
if available:
logger.info("WiFi collector: using LinuxWifiCollector on %s", interface)
return LinuxWifiCollector(interface=interface, sample_rate_hz=sample_rate_hz)
logger.warning("WiFi collector: LinuxWifiCollector unavailable (%s).", reason)
elif system == "Windows":
try:
win_iface = interface if interface != "wlan0" else "Wi-Fi"
collector = WindowsWifiCollector(interface=win_iface, sample_rate_hz=min(sample_rate_hz, 2.0))
collector.collect_once()
logger.info("WiFi collector: using WindowsWifiCollector on '%s'", interface)
return collector
except Exception as exc:
logger.warning("WiFi collector: WindowsWifiCollector unavailable (%s).", exc)
elif system == "Darwin":
try:
collector = MacosWifiCollector(sample_rate_hz=sample_rate_hz)
logger.info("WiFi collector: using MacosWifiCollector")
return collector
except Exception as exc:
logger.warning("WiFi collector: MacosWifiCollector unavailable (%s).", exc)
elif preferred == "linux":
return LinuxWifiCollector(interface=interface, sample_rate_hz=sample_rate_hz)
elif preferred == "windows":
return WindowsWifiCollector(interface=interface, sample_rate_hz=min(sample_rate_hz, 2.0))
elif preferred == "macos":
return MacosWifiCollector(sample_rate_hz=sample_rate_hz)
elif preferred == "simulated":
return SimulatedCollector(seed=42, sample_rate_hz=sample_rate_hz)
logger.info(
"WiFi collector: falling back to SimulatedCollector. "
"For real sensing, connect ESP32 nodes via UDP:5005 or install platform WiFi drivers."
)
return SimulatedCollector(seed=42, sample_rate_hz=sample_rate_hz)
+18 -42
View File
@@ -24,7 +24,6 @@ import asyncio
import json
import logging
import math
import platform
import signal
import socket
import struct
@@ -38,10 +37,6 @@ import numpy as np
# Sensing pipeline imports
from v1.src.sensing.rssi_collector import (
LinuxWifiCollector,
SimulatedCollector,
WindowsWifiCollector,
MacosWifiCollector,
WifiSample,
RingBuffer,
)
@@ -321,7 +316,13 @@ class SensingWebSocketServer:
self._running = False
def _create_collector(self):
"""Auto-detect data source: ESP32 UDP > Windows WiFi > Linux WiFi > simulated."""
"""Auto-detect data source: ESP32 UDP > platform WiFi > simulated.
Uses the ``create_collector`` factory (ADR-049) for platform WiFi
detection, which never raises and logs actionable fallback messages.
"""
from .rssi_collector import create_collector
# 1. Try ESP32 UDP first
print(" Probing for ESP32 on UDP :5005 ...")
if probe_esp32_udp(ESP32_UDP_PORT, timeout=2.0):
@@ -329,43 +330,18 @@ class SensingWebSocketServer:
self.source = "esp32"
return Esp32UdpCollector(port=ESP32_UDP_PORT, sample_rate_hz=10.0)
# 2. Platform-specific WiFi
system = platform.system()
if system == "Windows":
try:
collector = WindowsWifiCollector(sample_rate_hz=2.0)
collector.collect_once() # test that it works
logger.info("Using WindowsWifiCollector")
self.source = "windows_wifi"
return collector
except Exception as e:
logger.warning("Windows WiFi unavailable (%s), falling back", e)
elif system == "Linux":
# In Docker on Mac, Linux is detected but no wireless extensions exist.
# Force SimulatedCollector if /proc/net/wireless doesn't exist.
import os
if os.path.exists("/proc/net/wireless"):
try:
collector = LinuxWifiCollector(sample_rate_hz=10.0)
self.source = "linux_wifi"
return collector
except RuntimeError:
logger.warning("Linux WiFi unavailable, falling back")
else:
logger.warning("Linux detected but /proc/net/wireless missing (likely Docker). Falling back.")
elif system == "Darwin":
try:
collector = MacosWifiCollector(sample_rate_hz=10.0)
logger.info("Using MacosWifiCollector")
self.source = "macos_wifi"
return collector
except Exception as e:
logger.warning("macOS WiFi unavailable (%s), falling back", e)
# 2. Platform-specific WiFi (auto-detect with graceful fallback)
collector = create_collector(preferred="auto", sample_rate_hz=10.0)
# 3. Simulated
logger.info("Using SimulatedCollector")
self.source = "simulated"
return SimulatedCollector(seed=42, sample_rate_hz=10.0)
# Map collector class to source label
source_map = {
"LinuxWifiCollector": "linux_wifi",
"WindowsWifiCollector": "windows_wifi",
"MacosWifiCollector": "macos_wifi",
"SimulatedCollector": "simulated",
}
self.source = source_map.get(type(collector).__name__, "unknown")
return collector
def _build_message(self, features: RssiFeatures, result: SensingResult) -> str:
"""Build the JSON message to broadcast."""
+103
View File
@@ -702,3 +702,106 @@ class TestBandPower:
# Band 0.21-0.39 has no power
p = _band_power(freqs, psd, 0.21, 0.39)
assert p == 0.0
# ===========================================================================
# LinuxWifiCollector.is_available() tests (ADR-049)
# ===========================================================================
from unittest.mock import patch, mock_open
from v1.src.sensing.rssi_collector import LinuxWifiCollector, create_collector
class TestLinuxWifiCollectorAvailability:
def test_unavailable_when_proc_missing(self):
"""is_available returns False when /proc/net/wireless doesn't exist."""
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "/proc/net/wireless not found" in reason
def test_unavailable_when_interface_not_listed(self):
"""is_available returns False when the interface isn't in proc."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan1: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "wlan0" in reason
assert "wlan1" in reason
def test_available_when_interface_listed(self):
"""is_available returns True when the interface is present."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan0: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is True
assert reason == "ok"
def test_unavailable_when_file_unreadable(self):
"""is_available returns False when /proc/net/wireless exists but can't be read."""
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", side_effect=PermissionError("Permission denied")):
available, reason = LinuxWifiCollector.is_available("wlan0")
assert available is False
assert "Cannot read" in reason
# ===========================================================================
# create_collector() factory tests (ADR-049)
# ===========================================================================
class TestCreateCollector:
def test_returns_simulated_when_no_wifi(self):
"""On Linux without /proc/net/wireless, should return SimulatedCollector."""
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Linux"):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
collector = create_collector(preferred="auto")
assert isinstance(collector, SimulatedCollector)
def test_returns_simulated_for_explicit_preference(self):
"""preferred='simulated' always returns SimulatedCollector."""
collector = create_collector(preferred="simulated")
assert isinstance(collector, SimulatedCollector)
def test_returns_linux_collector_when_available(self):
"""On Linux with /proc/net/wireless, should return LinuxWifiCollector."""
proc_content = (
"Inter-| sta-| Quality | Discarded packets\n"
" face | tus | link level noise | nwid crypt frag retry misc\n"
" wlan0: 0000 60. -50. -95. 0 0 0 0 0\n"
)
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Linux"):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=True):
with patch("builtins.open", mock_open(read_data=proc_content)):
collector = create_collector(preferred="auto", interface="wlan0")
assert isinstance(collector, LinuxWifiCollector)
def test_never_raises(self):
"""create_collector should never raise, regardless of platform."""
for plat in ["Linux", "Windows", "Darwin", "FreeBSD", "SunOS"]:
with patch("v1.src.sensing.rssi_collector.platform.system", return_value=plat):
with patch("v1.src.sensing.rssi_collector.os.path.exists", return_value=False):
with patch("subprocess.run", side_effect=FileNotFoundError("not found")):
try:
collector = create_collector(preferred="auto")
assert collector is not None
except Exception as exc:
pytest.fail(f"create_collector raised on {plat}: {exc}")
def test_windows_default_interface_mapping(self):
"""On Windows with default interface='wlan0', should map to 'Wi-Fi'."""
with patch("v1.src.sensing.rssi_collector.platform.system", return_value="Windows"):
with patch("subprocess.run", side_effect=FileNotFoundError("netsh not found")):
collector = create_collector(preferred="auto", interface="wlan0")
# Should fall back to SimulatedCollector since netsh isn't available
assert isinstance(collector, SimulatedCollector)