ben
22b035dbee
Some checks failed
Build Sphinx Docs Set / Build Docs (pull_request) Has been cancelled
Test with tox / Test with tox (3.10) (pull_request) Has been cancelled
Test with tox / Test with tox (3.11) (pull_request) Has been cancelled
Test with tox / Test with tox (3.12) (pull_request) Has been cancelled
Build Project / Build Project (3.12) (pull_request) Has been cancelled
Build Project / Build Project (3.11) (pull_request) Has been cancelled
Build Project / Build Project (3.10) (pull_request) Has been cancelled
116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
"""CLI flags for TX opt-in and interlocks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from unittest.mock import patch
|
|
|
|
from ria_toolkit_oss.agent import cli as agent_cli
|
|
from ria_toolkit_oss.agent import config as agent_config
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, payload: dict):
|
|
self._payload = payload
|
|
|
|
def read(self) -> bytes:
|
|
return json.dumps(self._payload).encode()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_a):
|
|
return False
|
|
|
|
|
|
def _run_register(argv: list[str], cfg_path) -> int:
|
|
fake_resp = _FakeResp({"agent_id": "agent-1", "token": "tok-abc"})
|
|
with (
|
|
patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False),
|
|
patch("urllib.request.urlopen", return_value=fake_resp),
|
|
patch.object(sys, "argv", ["ria-agent", *argv]),
|
|
):
|
|
try:
|
|
agent_cli.main()
|
|
except SystemExit as exc:
|
|
return int(exc.code or 0)
|
|
return 0
|
|
|
|
|
|
def test_register_without_allow_tx_keeps_tx_disabled(tmp_path):
|
|
cfg_path = tmp_path / "agent.json"
|
|
_run_register(
|
|
["register", "--hub", "http://hub:3005", "--api-key", "K"],
|
|
cfg_path,
|
|
)
|
|
cfg = agent_config.load(path=cfg_path)
|
|
assert cfg.agent_id == "agent-1"
|
|
assert cfg.tx_enabled is False
|
|
assert cfg.tx_max_gain_db is None
|
|
|
|
|
|
def test_register_with_allow_tx_and_caps(tmp_path):
|
|
cfg_path = tmp_path / "agent.json"
|
|
_run_register(
|
|
[
|
|
"register",
|
|
"--hub",
|
|
"http://hub:3005",
|
|
"--api-key",
|
|
"K",
|
|
"--allow-tx",
|
|
"--tx-max-gain-db",
|
|
"-10",
|
|
"--tx-max-duration-s",
|
|
"60",
|
|
"--tx-freq-range",
|
|
"2.4e9",
|
|
"2.5e9",
|
|
"--tx-freq-range",
|
|
"5.7e9",
|
|
"5.8e9",
|
|
],
|
|
cfg_path,
|
|
)
|
|
cfg = agent_config.load(path=cfg_path)
|
|
assert cfg.tx_enabled is True
|
|
assert cfg.tx_max_gain_db == -10.0
|
|
assert cfg.tx_max_duration_s == 60.0
|
|
assert cfg.tx_allowed_freq_ranges == [[2.4e9, 2.5e9], [5.7e9, 5.8e9]]
|
|
|
|
|
|
def test_stream_allow_tx_does_not_persist(tmp_path):
|
|
# Pre-register with tx_enabled=False, then simulate `stream --allow-tx`.
|
|
# The on-disk config must remain unchanged; the runtime flag is process-local.
|
|
cfg_path = tmp_path / "agent.json"
|
|
base = agent_config.AgentConfig(
|
|
hub_url="http://hub:3005",
|
|
agent_id="agent-1",
|
|
token="tok-abc",
|
|
tx_enabled=False,
|
|
)
|
|
agent_config.save(base, path=cfg_path)
|
|
|
|
captured: dict = {}
|
|
|
|
async def _fake_run_streamer(url, token, *, cfg):
|
|
captured["cfg"] = cfg
|
|
return None
|
|
|
|
with (
|
|
patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False),
|
|
patch("ria_toolkit_oss.agent.streamer.run_streamer", new=_fake_run_streamer),
|
|
patch.object(sys, "argv", ["ria-agent", "stream", "--allow-tx"]),
|
|
):
|
|
try:
|
|
agent_cli.main()
|
|
except SystemExit:
|
|
pass
|
|
|
|
# Runtime cfg had TX flipped on
|
|
assert captured["cfg"].tx_enabled is True
|
|
# But the persisted file is untouched
|
|
on_disk = agent_config.load(path=cfg_path)
|
|
assert on_disk.tx_enabled is False
|