Created
March 13, 2026 14:33
-
-
Save wowi42/348cea6542d4a810ee55311219d33988 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| Test scenario: gather every pyinfra fact against the target hosts and | |
| write a per-host JSON report into the ``reports/`` directory. | |
| Usage: | |
| pyinfra inventory.py test_all_facts.py | |
| After the run, generate a combined HTML report with: | |
| python generate_report.py | |
| Fully dynamic: facts and their arguments are discovered at runtime via | |
| introspection, so this script never needs updating when pyinfra adds or | |
| removes facts. | |
| """ | |
| import importlib | |
| import inspect | |
| import json | |
| import os | |
| import pkgutil | |
| import re | |
| import traceback | |
| from datetime import datetime, timezone | |
| import pyinfra | |
| import pyinfra.facts as facts_pkg | |
| from pyinfra import host, logger | |
| # --------------------------------------------------------------------------- | |
| # Parameter-name → default-value mapping. When a fact's ``command()`` | |
| # method has a required parameter whose name matches a key here, the | |
| # corresponding value is used automatically. | |
| # --------------------------------------------------------------------------- | |
| PARAM_DEFAULTS: dict[str, object] = { | |
| "boolean": "httpd_can_network_connect", | |
| "command": "uptime", | |
| "directory": "/tmp", | |
| "object_id": "test", | |
| "package": "bash", | |
| "parameter": "hostname", | |
| "path": "/etc/hostname", | |
| "pattern": ".*", | |
| "port": 22, | |
| "protocol": "tcp", | |
| "repo": "/tmp", | |
| "src": "/dev/null", | |
| "srvname": "sshd", | |
| "target": "/etc/hostname", | |
| "user": "root", | |
| } | |
| def _is_base_class(name: str) -> bool: | |
| """Detect abstract base classes / mixins by naming convention.""" | |
| return bool(re.search(r"(Base|Mixin)$", name)) | |
| def _discover_facts(): | |
| """Yield (qualified_name, fact_class) for every concrete fact.""" | |
| for _, modname, _ in pkgutil.iter_modules(facts_pkg.__path__): | |
| mod = importlib.import_module(f"pyinfra.facts.{modname}") | |
| for name, obj in inspect.getmembers(mod, inspect.isclass): | |
| if not hasattr(obj, "command") or obj.__module__ != mod.__name__: | |
| continue | |
| if _is_base_class(name): | |
| continue | |
| yield f"{modname}.{name}", obj | |
| def _build_kwargs(cls) -> dict: | |
| """Build kwargs for a fact's command() from PARAM_DEFAULTS.""" | |
| cmd = cls.command | |
| if not callable(cmd) or isinstance(cmd, str): | |
| return {} | |
| sig = inspect.signature(cmd) | |
| kwargs = {} | |
| for p in sig.parameters.values(): | |
| if p.name == "self": | |
| continue | |
| if p.default is not inspect.Parameter.empty: | |
| continue # has a default — no need to supply | |
| if p.name in PARAM_DEFAULTS: | |
| kwargs[p.name] = PARAM_DEFAULTS[p.name] | |
| else: | |
| kwargs[p.name] = "test" # fallback for unknown params | |
| return kwargs | |
| # --------------------------------------------------------------------------- | |
| # Gather every fact and collect results. | |
| # --------------------------------------------------------------------------- | |
| fact_results = {} | |
| for qname, fact_cls in sorted(_discover_facts()): | |
| kwargs = _build_kwargs(fact_cls) | |
| try: | |
| value = host.get_fact(fact_cls, **kwargs) | |
| fact_results[qname] = { | |
| "status": "ok", | |
| "args": kwargs, | |
| "value": repr(value), | |
| } | |
| logger.info(f"OK {qname}") | |
| except Exception: | |
| tb = traceback.format_exc() | |
| fact_results[qname] = { | |
| "status": "fail", | |
| "args": kwargs, | |
| "error": tb, | |
| } | |
| logger.warning(f"FAIL {qname}") | |
| # --------------------------------------------------------------------------- | |
| # Write per-host JSON report. | |
| # --------------------------------------------------------------------------- | |
| report_dir = os.path.join(os.path.dirname(__file__) or ".", "reports") | |
| os.makedirs(report_dir, exist_ok=True) | |
| hostname = host.name | |
| display_name = host.data.get("display_name", hostname) | |
| ok = sum(1 for f in fact_results.values() if f["status"] == "ok") | |
| fail = sum(1 for f in fact_results.values() if f["status"] == "fail") | |
| report = { | |
| "host": hostname, | |
| "display_name": display_name, | |
| "pyinfra_version": pyinfra.__version__, | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "summary": {"total": len(fact_results), "ok": ok, "fail": fail}, | |
| "facts": fact_results, | |
| } | |
| report_path = os.path.join(report_dir, f"{hostname}.json") | |
| with open(report_path, "w") as fh: | |
| json.dump(report, fh, indent=2, default=str) | |
| logger.info(f"Report written to {report_path}") | |
| logger.info(f"RESULTS for {hostname}: {ok} ok, {fail} failed out of {len(fact_results)}") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment