diff --git a/tests/check_concentrator.py b/tests/check_concentrator.py index b15acc3..df61d8a 100644 --- a/tests/check_concentrator.py +++ b/tests/check_concentrator.py @@ -1,23 +1,20 @@ #!/usr/bin/env python3 """ -Smoke test: construct :class:`fiwi.concentrator.FiWiConcentrator`, then print Adnacom host-card -catalog, local USB power-control hubs (``show_hostcards``), and **remote** hub hosts from -``[remote_hubs]`` / ``FIWI_REMOTE_HUBS`` (each probed with ``ssh … show_hostcards`` when possible). +Smoke check: :class:`fiwi.concentrator.FiWiConcentrator` plus **consolidated** USB hub inventory +for **this machine** and every configured **remote** hub host (``show_hostcards`` / same discovery +path as ``discover``). Optional Adnacom PCIe host-card catalog at the end. -Requires the BrainStem Python package (``brainstem`` in requirements.txt). USB discovery may print -“no link specs” if no compatible hubs are attached — that is normal off the bench. Remote SSH must -reach the Pi (or host) where ``fiwi.py`` and BrainStem are installed. +Requires BrainStem (``brainstem``) for local ``show_hostcards`` when this host has hubs attached. +Remote hosts need ``fiwi.py`` + venv on the Pi as for ``check_remote``. -**Standalone** (any current working directory):: +**Standalone**:: python tests/check_concentrator.py python tests/check_concentrator.py --config uax24 - python tests/check_concentrator.py -c uax4 - python tests/check_concentrator.py --config /path/to/custom.ini -``--config`` is the same as ``FIWI_CONFIG``: a profile name (``config/.ini``) or an absolute path to a ``.ini`` file. +``--config`` matches ``FIWI_CONFIG`` (profile or absolute ``*.ini``). -With pytest, set the profile in the environment (argv is not parsed):: +With pytest:: FIWI_CONFIG=uax24 pytest tests/check_concentrator.py """ @@ -26,6 +23,7 @@ from __future__ import annotations import argparse import os +import socket import sys from typing import TYPE_CHECKING @@ -36,11 +34,15 @@ _ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), if _ROOT not in sys.path: sys.path.insert(0, _ROOT) +_WIDTH = 62 _paths_configured = False +def _rule(char: str = "-") -> str: + return char * _WIDTH + + def _ensure_paths_configured() -> None: - """Apply ``fiwi.paths.configure`` once (after ``FIWI_CONFIG`` is set for script runs).""" global _paths_configured if _paths_configured: return @@ -59,41 +61,71 @@ def _instantiate_concentrator() -> FiWiConcentrator: return c -def _print_remote_hub_hostcards() -> None: - """List configured remote hub SSH targets and run ``show_hostcards`` on each (if any).""" - from fiwi.ssh import SshNode, SshNodeConfig +def _print_banner(config_label: str) -> None: + print(_rule("="), flush=True) + print(f"Fi-Wi concentrator check [config: {config_label}]", flush=True) + print(_rule("="), flush=True) + print(flush=True) + + +def _print_ssh_and_hosts_summary() -> None: + from fiwi.ssh import SshNodeConfig cfg = SshNodeConfig.load() raw = (cfg.calibrate_remotes or "").strip() - rh_only = (os.environ.get("FIWI_REMOTE_HUBS") or "").strip() - cr_only = (os.environ.get("FIWI_CALIBRATE_REMOTES") or "").strip() + print("SSH / remote hub routing (after paths + INI)", flush=True) + print(_rule(), flush=True) + print(f" FIWI_REMOTE_PYTHON → {cfg.python}", flush=True) + print(f" FIWI_REMOTE_SCRIPT → {cfg.script}", flush=True) + print(f" FIWI_SSH_BIN {cfg.ssh_bin}", flush=True) + if cfg.ssh_extra_argv: + print(f" FIWI_SSH_OPTS {' '.join(cfg.ssh_extra_argv)}", flush=True) + print(f" merged hub hosts {raw or '(none)'}", flush=True) + print(_rule(), flush=True) + print(flush=True) - print("Remote hub hosts (from config)", flush=True) - print("-" * 60, flush=True) - if not raw: - print( - " (none) — set [remote_hubs] hosts or remote_ssh.calibrate_remotes in config/*.ini,\n" - " or FIWI_REMOTE_HUBS / FIWI_CALIBRATE_REMOTES in the environment.", - flush=True, - ) - print("-" * 60, flush=True) - print() - return - if rh_only: - print(f" FIWI_REMOTE_HUBS: {rh_only}", flush=True) - if cr_only: - print(f" FIWI_CALIBRATE_REMOTES: {cr_only}", flush=True) - print(f" merged for calibrate: {raw}", flush=True) - print("-" * 60, flush=True) +def _merged_hub_hosts() -> list[str]: + from fiwi.ssh import SshNodeConfig + raw = (SshNodeConfig.load().calibrate_remotes or "").strip() seen: list[str] = [] for part in raw.split(","): - host = part.strip() - if not host or host in seen: - continue - seen.append(host) - print(f"\nRemote USB power-control hubs via SSH → {host}\n", flush=True) + h = part.strip() + if h and h not in seen: + seen.append(h) + return seen + + +def _print_local_usb_section(c: FiWiConcentrator) -> None: + hn = socket.gethostname() + print(f"--- This machine — {hn} (local USB) ---", flush=True) + print(flush=True) + c.show_hostcards() + print(flush=True) + + +def _print_remote_usb_sections() -> int: + """ + Run ``show_hostcards`` over SSH for each merged hub host. Returns 0 if all ok, else 1. + """ + from fiwi.ssh import SshNode + + hosts = _merged_hub_hosts() + if not hosts: + print("--- Remote USB hubs ---", flush=True) + print( + " (no remote hosts) — set [remote_hubs] hosts or FIWI_REMOTE_HUBS / " + "FIWI_CALIBRATE_REMOTES.", + flush=True, + ) + print(flush=True) + return 0 + + rc = 0 + for host in hosts: + print(f"--- Remote USB hubs — SSH → {host} ---", flush=True) + print(flush=True) try: node = SshNode.parse(host) code, out, err = node.invoke_capture(["show_hostcards"], timeout=90, defer=False) @@ -102,23 +134,37 @@ def _print_remote_hub_hostcards() -> None: if err.strip(): print(err.rstrip(), file=sys.stderr, flush=True) if code != 0: - print(f" (remote show_hostcards exit {code})", flush=True) + print(f" (show_hostcards exit {code})", flush=True) + rc = 1 except Exception as exc: print(f" SSH / remote run failed: {exc}", flush=True) - print() + rc = 1 + print(flush=True) + return rc -def _print_inventory(c: FiWiConcentrator) -> None: - """Adnacom PCIe catalog + local and remote USB power-control hub views.""" +def _print_pcie_catalog_section() -> None: from fiwi.adnacom_pcie_catalog import print_adnacom_host_card_table - print() + print("--- PCIe host-card catalog (reference) ---", flush=True) + print(flush=True) print_adnacom_host_card_table() - print() - print("USB power-control hubs — local (hostcards)", flush=True) - c.show_hostcards() - print() - _print_remote_hub_hostcards() + print(flush=True) + + +def _print_consolidated_report(c: FiWiConcentrator) -> int: + """ + One flow: config summary → USB hubs (local + all remotes) → PCIe catalog. + Returns 0 if remote SSH sections all succeeded, else 1 (local errors still raise). + """ + _print_ssh_and_hosts_summary() + print("USB power-control hubs (BrainStem hostcards)", flush=True) + print(_rule(), flush=True) + print(flush=True) + _print_local_usb_section(c) + remote_rc = _print_remote_usb_sections() + _print_pcie_catalog_section() + return remote_rc def test_concentrator() -> None: @@ -128,11 +174,10 @@ def test_concentrator() -> None: def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser( - description="Smoke test: instantiate FiWiConcentrator using a config profile.", + description="Check FiWiConcentrator: consolidated local + remote USB hub inventory.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "PROFILE selects /config/.ini (e.g. uax24, uax4, default).\n" - "An absolute path to a .ini file is used as-is.\n" "Same as environment variable FIWI_CONFIG." ), ) @@ -154,10 +199,15 @@ def main() -> int: args = _parse_args() if args.config: os.environ["FIWI_CONFIG"] = args.config.strip() + + label = os.environ.get("FIWI_CONFIG", "default (config/default.ini if present)") + _print_banner(label) + c = None + remote_fail = 0 try: c = _instantiate_concentrator() - _print_inventory(c) + remote_fail = _print_consolidated_report(c) except AssertionError as exc: print(f"FAIL: {exc}", file=sys.stderr) return 1 @@ -170,9 +220,13 @@ def main() -> int: c.disconnect() except Exception: pass - label = os.environ.get("FIWI_CONFIG", "default (config/default.ini if present)") - print(f"FiWiConcentrator() OK [config: {label}]") - return 0 + + print(_rule("="), flush=True) + print(f"FiWiConcentrator() OK [config: {label}]", flush=True) + if remote_fail != 0: + print(" (one or more remote show_hostcards runs failed — see above)", flush=True) + print(_rule("="), flush=True) + return remote_fail if __name__ == "__main__":