128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
# Copyright (c) 2026 Umber
|
|
#
|
|
# Licensed under the Apache License, Version 2.0; see LICENSE.
|
|
|
|
"""Integration: remote host (e.g. Pi) has fiwicontrol.power stack for discovery.
|
|
|
|
Opt-in: set ``FIWI_RUN_REMOTE_TESTS=1`` (and optionally ``FIWI_REMOTE_IP``) so these
|
|
tests run. Without that, the class is skipped so default ``pytest`` runs do not
|
|
require a reachable rig.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fiwicontrol.commands.node_control import ssh_node
|
|
|
|
# Same default as tests/conftest.py so `python -m unittest …` still avoids broken system ssh_config.d.
|
|
_minimal_ssh = Path(__file__).resolve().parent / "fixtures" / "ssh_config_minimal"
|
|
if _minimal_ssh.is_file():
|
|
os.environ.setdefault("FIWI_SSH_CONFIG", str(_minimal_ssh))
|
|
|
|
|
|
def _remote_tests_enabled() -> bool:
|
|
v = os.environ.get("FIWI_RUN_REMOTE_TESTS", "").strip().lower()
|
|
return v in ("1", "true", "yes")
|
|
|
|
|
|
_REMOTE_TESTS_SKIP = unittest.skipUnless(
|
|
_remote_tests_enabled(),
|
|
"set FIWI_RUN_REMOTE_TESTS=1 to run SSH integration tests against FIWI_REMOTE_IP",
|
|
)
|
|
|
|
_REMOTE_INSTALL_HINT = (
|
|
"On the Raspberry Pi (as root), from your FiWiControl clone, run:\n"
|
|
" cd ~/Code/FiWiControl # or your path\n"
|
|
" python3 -m pip install -e \".[power]\" # use the SAME interpreter you want for SSH\n"
|
|
"Then: python3 -c \"import fiwicontrol.power\"\n"
|
|
"If `pip show` lists python3.11 site-packages but `python3` is another version, set on Fedora:\n"
|
|
" export FIWI_REMOTE_PYTHON=python3.11\n"
|
|
"See docs/install.md (Remote host setup)."
|
|
)
|
|
|
|
|
|
def _remote_py() -> str:
|
|
return os.environ.get("FIWI_REMOTE_PYTHON", "python3")
|
|
|
|
|
|
class _RemoteNodeMixin:
|
|
def _make_node(self, *, silent_mode: bool = False) -> ssh_node:
|
|
remote_ip = os.getenv("FIWI_REMOTE_IP", "192.168.1.39")
|
|
return ssh_node(
|
|
name="remote-power-check",
|
|
ipaddr=remote_ip,
|
|
ssh_controlmaster=False,
|
|
silent_mode=silent_mode,
|
|
)
|
|
|
|
|
|
@_REMOTE_TESTS_SKIP
|
|
class TestRemotePowerDependencies(_RemoteNodeMixin, unittest.IsolatedAsyncioTestCase):
|
|
async def test_remote_imports_fiwicontrol_power_brainstem_serial(self) -> None:
|
|
"""Remote must have fiwicontrol, brainstem, and pyserial (serial) for full discovery."""
|
|
node = self._make_node(silent_mode=True)
|
|
py = _remote_py()
|
|
cmd = (
|
|
f"{py} -c "
|
|
"\"import fiwicontrol.power, brainstem, serial; print('FIWI_REMOTE_POWER_IMPORTS_OK')\""
|
|
)
|
|
session = await node.rexec(
|
|
cmd=cmd,
|
|
IO_TIMEOUT=20.0,
|
|
CMD_TIMEOUT=45,
|
|
CONNECT_TIMEOUT=25.0,
|
|
repeat=None,
|
|
)
|
|
out = session.results.decode("utf-8", errors="replace")
|
|
self.assertIn(
|
|
"FIWI_REMOTE_POWER_IMPORTS_OK",
|
|
out,
|
|
"Remote Python is missing fiwicontrol (and/or brainstem/pyserial).\n{}\n--- remote output ---\n{}".format(
|
|
_REMOTE_INSTALL_HINT,
|
|
out,
|
|
),
|
|
)
|
|
|
|
async def test_remote_power_module_discovery_json(self) -> None:
|
|
"""Remote must run `python -m fiwicontrol.power --discovery-json` (used by inventory verify)."""
|
|
node = self._make_node(silent_mode=True)
|
|
py = _remote_py()
|
|
session = await node.rexec(
|
|
cmd=f"{py} -m fiwicontrol.power --discovery-json",
|
|
IO_TIMEOUT=30.0,
|
|
CMD_TIMEOUT=90,
|
|
CONNECT_TIMEOUT=30.0,
|
|
repeat=None,
|
|
)
|
|
text = session.results.decode("utf-8", errors="replace").strip()
|
|
self.assertTrue(
|
|
text,
|
|
"Expected non-empty JSON from remote `{} -m fiwicontrol.power --discovery-json`".format(py),
|
|
)
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError as exc:
|
|
if "No module named 'fiwicontrol'" in text or "fiwicontrol.power" in text:
|
|
self.fail(
|
|
"Remote did not return JSON (usually fiwicontrol not installed on the Pi).\n{}\n--- remote output ---\n{}".format(
|
|
_REMOTE_INSTALL_HINT,
|
|
text,
|
|
)
|
|
)
|
|
raise AssertionError("Invalid JSON from remote discovery: {}\n{}".format(exc, text[:2000])) from exc
|
|
self.assertIn("acroname", data, "discovery JSON missing acroname: {!r}".format(text[:500]))
|
|
self.assertIn("monsoon", data, "discovery JSON missing monsoon: {!r}".format(text[:500]))
|
|
self.assertIsInstance(data["acroname"], list)
|
|
self.assertIsInstance(data["monsoon"], list)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Direct unittest entry implies a deliberate remote run (pytest still requires FIWI_RUN_REMOTE_TESTS=1).
|
|
os.environ.setdefault("FIWI_RUN_REMOTE_TESTS", "1")
|
|
unittest.main(module=__name__, argv=[sys.argv[0], *sys.argv[1:]])
|