275 lines
9.3 KiB
Python
275 lines
9.3 KiB
Python
"""Tests for the `ria campaign` CLI commands."""
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import yaml
|
|
from click.testing import CliRunner
|
|
|
|
from ria_toolkit_oss_cli.cli import cli
|
|
|
|
|
|
def _write_yaml(d: dict, suffix=".yml") -> str:
|
|
f = tempfile.NamedTemporaryFile(mode="w", suffix=suffix, delete=False)
|
|
yaml.dump(d, f)
|
|
f.close()
|
|
return f.name
|
|
|
|
|
|
WIFI_PROFILE = {
|
|
"device": {"name": "iPhone_13_WiFi", "type": "wifi"},
|
|
"capture": {
|
|
"channels": [1, 6, 11],
|
|
"bandwidth": "20MHz",
|
|
"traffic_patterns": ["idle", "ping", "iperf_udp"],
|
|
"duration_per_config": "30s",
|
|
},
|
|
"recorder": {
|
|
"device": "usrp_b210",
|
|
"center_freq": "2.45GHz",
|
|
"sample_rate": "40MHz",
|
|
"gain": "auto",
|
|
},
|
|
"output": {"path": "/tmp/test_enroll", "device_id": "iphone13_wifi_001"},
|
|
}
|
|
|
|
FULL_CAMPAIGN = {
|
|
"campaign": {"name": "wifi_capture_001", "mode": "controlled_testbed"},
|
|
"transmitters": [
|
|
{
|
|
"id": "laptop_wifi",
|
|
"type": "wifi",
|
|
"control_method": "external_script",
|
|
"schedule": [
|
|
{"channel": 6, "bandwidth": "20MHz", "traffic": "iperf_udp", "duration": "30s"},
|
|
{"channel": 11, "bandwidth": "20MHz", "traffic": "idle", "duration": "30s"},
|
|
],
|
|
}
|
|
],
|
|
"recorder": {
|
|
"device": "usrp_b210",
|
|
"center_freq": "2.45GHz",
|
|
"sample_rate": "20MHz",
|
|
"gain": "40dB",
|
|
},
|
|
"qa": {"snr_threshold": "10dB", "min_duration": "25s", "flag_for_review": True},
|
|
"output": {"format": "sigmf", "path": "/tmp/test_campaign"},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ria campaign --help
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCampaignHelp:
|
|
def test_campaign_help(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "campaign" in result.output.lower()
|
|
|
|
def test_subcommands_listed(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "--help"])
|
|
assert result.exit_code == 0
|
|
for sub in ("validate", "enroll", "run"):
|
|
assert sub in result.output
|
|
|
|
def test_validate_help(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_enroll_help(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "enroll", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_run_help(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "run", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ria campaign validate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCampaignValidate:
|
|
def test_validate_device_profile(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path, "--profile"])
|
|
assert result.exit_code == 0
|
|
assert "✓" in result.output or "valid" in result.output.lower()
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_shows_campaign_name(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path, "--profile"])
|
|
assert "enroll_iphone13_wifi_001" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_shows_step_count(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path, "--profile"])
|
|
assert "9" in result.output # 9 total steps
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_shows_capture_time(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path, "--profile"])
|
|
assert "270" in result.output # 270s total
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_full_campaign(self):
|
|
path = _write_yaml(FULL_CAMPAIGN)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path])
|
|
assert result.exit_code == 0
|
|
assert "wifi_capture_001" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_shows_steps(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path, "--profile"])
|
|
assert "ch01_20mhz_idle" in result.output
|
|
assert "ch06_20mhz_ping" in result.output
|
|
assert "ch11_20mhz_iperf_udp" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_validate_missing_file(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", "/nonexistent/file.yml"])
|
|
assert result.exit_code != 0
|
|
|
|
def test_validate_bad_yaml(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False) as f:
|
|
f.write(": broken yaml [\n")
|
|
path = f.name
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "validate", "--config", path])
|
|
assert result.exit_code != 0
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ria campaign enroll --dry-run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCampaignEnrollDryRun:
|
|
def test_dry_run_exits_cleanly(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "enroll", "--config", path, "--dry-run"])
|
|
assert result.exit_code == 0
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_dry_run_shows_campaign_info(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "enroll", "--config", path, "--dry-run"])
|
|
assert "enroll_iphone13_wifi_001" in result.output
|
|
assert "9" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_dry_run_does_not_capture(self):
|
|
"""Dry run should not create any output files."""
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
runner.invoke(
|
|
cli,
|
|
["campaign", "enroll", "--config", path, "--output", tmpdir, "--dry-run"],
|
|
)
|
|
# No .sigmf-data files should have been created
|
|
sigmf_files = list(os.walk(tmpdir))
|
|
all_files = [f for _, _, files in sigmf_files for f in files]
|
|
assert not any(f.endswith(".sigmf-data") for f in all_files)
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_dry_run_output_override(self):
|
|
path = _write_yaml(WIFI_PROFILE)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
["campaign", "enroll", "--config", path, "--output", "/tmp/custom_out", "--dry-run"],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Dry run" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ria campaign run --dry-run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCampaignRunDryRun:
|
|
def test_dry_run_exits_cleanly(self):
|
|
path = _write_yaml(FULL_CAMPAIGN)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "run", "--config", path, "--dry-run"])
|
|
assert result.exit_code == 0
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_dry_run_shows_campaign_name(self):
|
|
path = _write_yaml(FULL_CAMPAIGN)
|
|
try:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "run", "--config", path, "--dry-run"])
|
|
assert "wifi_capture_001" in result.output
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_dry_run_does_not_create_report(self):
|
|
path = _write_yaml(FULL_CAMPAIGN)
|
|
try:
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
report_path = os.path.join(tmpdir, "qa_report.json")
|
|
result = runner.invoke(
|
|
cli,
|
|
["campaign", "run", "--config", path, "--dry-run", "--report", report_path],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert not os.path.exists(report_path)
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_missing_config_fails(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["campaign", "run", "--config", "/nonexistent.yml"])
|
|
assert result.exit_code != 0
|