Add Tier-1 integration tests; fix bugs; inline result collector

README.md

@@ -28,7 +28,7 @@ Ansible has a great executor and a huge module ecosystem, but the existing Pytho
 pip install ansible-host
 ```
 
-Requires Python 3.10+ and `ansible-core>=2.16,<2.20`. Like Ansible itself, the library runs on POSIX systems (Linux, macOS, WSL) — it is not supported on native Windows.
+Requires Python 3.10+ and `ansible-core>=2.16,<2.22`. Like Ansible itself, the library runs on POSIX systems (Linux, macOS, WSL) — it is not supported on native Windows.
 
 ## Development
 
@@ -58,32 +58,117 @@ uv run pytest
 
 ## Quickstart
 
+### 30-second try — no SSH, no inventory
+
+`AnsibleLocalhost` runs modules in-process on the current machine via Ansible's `local` connection plugin. No inventory file, no SSH, no setup.
+
 ```python
-from ansible_host import AnsibleHost, AnsibleHosts
+from ansible_host import AnsibleLocalhost
+
+host = AnsibleLocalhost()
+result = host.ping()
+assert result["ping"] == "pong"
 
-# Single host
-host = AnsibleHost(inventory="inventory.yml", pattern="vlab-01")
-result = host.shell("uptime")
+result = host.command("uname -a")
 print(result["stdout"])
+```
+
+### Running against a real host
+
+Drop an inventory file alongside your script:
 
-# Group of hosts (returns dict keyed by hostname)
-hosts = AnsibleHosts(inventory="inventory.yml", pattern="vms_1")
-results = hosts.ping()
+```ini
+# inventory.ini
+[switches]
+sw-01 ansible_host=10.0.0.1 ansible_user=admin
+```
+
+```python
+from ansible_host import AnsibleHost
+
+host = AnsibleHost(inventory="inventory.ini", pattern="sw-01")
+result = host.shell("show version")
+print(result["stdout"])
+```
+
+### Multi-host fanout
+
+```python
+from ansible_host import AnsibleHosts
+
+hosts = AnsibleHosts(inventory="inventory.ini", pattern="switches")
+results = hosts.ping()                       # parallelism via forks=
 for hostname, r in results.items():
     print(hostname, r["ping"])
 
-# Any Ansible module is callable as a method via __getattr__
+# Container API: index, iterate, len
+print(len(hosts), hosts.hostnames)
+first = hosts[0]                             # -> AnsibleHost
+by_name = hosts["sw-01"]                     # -> AnsibleHost
+for h in hosts:
+    print(h.hostname)
+```
+
+### Dynamic dispatch and task directives
+
+Any Ansible module is callable as a method via `__getattr__` (`host.<module_name>(...)`):
+
+```python
 host.copy(src="/etc/hosts", dest="/tmp/hosts.bak")
-host.shell("ls /tmp", task_directives={"ignore_errors": True})
+host.command("rm /tmp/maybe-missing", task_directives={"ignore_errors": True})
+host.shell("echo $TOKEN", task_directives={"no_log": True})
+```
+
+Common `task_directives`: `ignore_errors`, `no_log`, `when`, `failed_when`, `changed_when`, `become`.
+
+### Batch mode — queue tasks, run them in a single play
 
-# Batch mode: queue tasks, run them in a single play
+```python
 with host:
     host.shell("uptime")
     host.shell("df -h")
     host.shell("free -m")
-results = host.results  # list of task results
+results = host.results
+```
+
+### Building a queue across functions
+
+`with host:` is lexically scoped. If you need to assemble a batch across multiple functions, use the explicit form — same machinery, no scope limit:
+
+```python
+host.load_module("ansible.builtin.command", args=["uptime"])
+host.load_module("ansible.builtin.command", args=["df -h"])
+results = host.run_loaded_modules()
+```
+
+### Result shapes
+
+|              | single task             | batch (`with` block)                |
+| ------------ | ----------------------- | ----------------------------------- |
+| single host  | `dict`                  | `list[dict]`                        |
+| multi host   | `{hostname: dict}`      | `{hostname: list[dict]}`            |
+
+### Failures
+
+A failing module raises `AnsibleModuleFailed`:
+
+```python
+from ansible_host import AnsibleModuleFailed
+
+try:
+    host.command("false")
+except AnsibleModuleFailed as e:
+    print("module failed:", e)
+
+# Or suppress and inspect the result:
+result = host.command("false", task_directives={"ignore_errors": True})
+assert result["failed"] is True
 ```
 
+### More examples
+
+See [`tests/test_local_integration.py`](tests/test_local_integration.py) for 30 runnable examples covering ping, command, shell, batch mode, dynamic dispatch, `no_log`, `forks`, multi-host fanout, per-host failure aggregation, and the container protocol.
+
 ## Compatibility
 
 This library uses Ansible's internal Python API (`TaskQueueManager`, `InventoryManager`, `VariableManager`, `Play`, `DataLoader`). Those APIs are not officially stable across `ansible-core` releases — expect occasional updates when `ansible-core` introduces breaking internal changes. The current support range is declared in `pyproject.toml` and in the matrix CI.

src/ansible_host/__init__.py

@@ -30,6 +30,7 @@
 from ansible.module_utils.common.collections import ImmutableDict
 from ansible.parsing.dataloader import DataLoader
 from ansible.playbook.play import Play
+from ansible.plugins.callback import CallbackBase
 from ansible.plugins.loader import init_plugin_loader, module_loader
 from ansible.utils.display import Display
 from ansible.vars.hostvars import HostVars
@@ -42,6 +43,103 @@
 logger = logging.getLogger("ansible_pyapi")
 
 
+class _JsonResultsCallback(CallbackBase):
+    """Internal result collector for AnsibleHostsBase._run.
+
+    Not registered via Ansible's plugin loader: it is instantiated directly
+    by _run and attached to the TaskQueueManager's _callback_plugins list.
+    The leading underscore marks this as package-private; downstream users
+    should rely on the structured results returned by run_module() / .results
+    rather than instantiating this class themselves.
+
+    Derived from the json_results callback in sonic-net/sonic-mgmt (Apache 2.0).
+    """
+
+    CALLBACK_VERSION = 2.0
+    # Deliberately NOT 'stdout' — Ansible's built-in `null` stdout callback
+    # handles terminal output (silently); this collector is purely a result sink.
+    CALLBACK_TYPE = "notification"
+    CALLBACK_NAME = "ansible_host._json_results"
+
+    TASK_FIELDS = (
+        "action",
+        "become",
+        "become_method",
+        "become_user",
+        "connection",
+        "ignore_errors",
+        "ignore_unreachable",
+        "register",
+        "retries",
+        "timeout",
+    )
+
+    def __init__(self):
+        super().__init__()
+        self._results: dict[str, list[dict]] = {}
+
+    def _get_module_name(self, result):
+        if ansible.__version__ >= "2.19.0":
+            return result.task
+        return result.task_name
+
+    def _get_task_fields(self, result):
+        return {
+            field: result._task_fields[field]
+            for field in self.TASK_FIELDS
+            if field in result._task_fields
+        }
+
+    def _log_res(self, hostname, module_name, res):
+        if display.verbosity == 0:
+            return
+        if display.verbosity == 1:
+            log_func = display.v
+            brief = json.dumps(
+                {"module_name": module_name, "reachable": res["reachable"], "failed": res["failed"]},
+                default=str,
+            )
+            msg = f"[{hostname}] => {brief}"
+        elif display.verbosity == 2:
+            log_func = display.vv
+            msg = f"[{hostname}] => {json.dumps(res, default=str)}"
+        elif display.verbosity == 3:
+            log_func = display.vvv
+            msg = f"[{hostname}] => {json.dumps(res, indent=4, default=str)}"
+        else:
+            log_func = display.vvvv
+            msg = f"[{hostname}] => {json.dumps(res, indent=4, default=str)}"
+        log_func(msg)
+
+    def _record(self, result, *, reachable, failed):
+        hostname = str(result._host.get_name())
+        module_name = self._get_module_name(result)
+        self._results.setdefault(hostname, [])
+        res = dict(hostname=hostname, reachable=reachable, failed=failed)
+        # deepcopy preserves non-serializable values (e.g. Task instances in
+        # _task_fields on ansible-core 2.19+); the prior json round-trip lost
+        # or crashed on those.
+        res.update(copy.deepcopy(result._result))
+        if "invocation" in res and isinstance(res["invocation"], dict):
+            res["invocation"]["module_name"] = module_name
+        res["_task_fields"] = self._get_task_fields(result)
+        self._log_res(hostname, module_name, res)
+        self._results[hostname].append(res)
+
+    def v2_runner_on_ok(self, result):
+        self._record(result, reachable=True, failed=False)
+
+    def v2_runner_on_failed(self, result, *args, **kwargs):
+        self._record(result, reachable=True, failed=True)
+
+    def v2_runner_on_unreachable(self, result):
+        self._record(result, reachable=False, failed=True)
+
+    @property
+    def results(self) -> dict[str, list[dict]]:
+        return self._results
+
+
 def _to_native_type(value: Any) -> Any:
     """Convert Ansible types (AnsibleUnicode, AnsibleUnsafeText, etc.) to native Python types.
 
@@ -261,7 +359,8 @@ def _check_failed_results(self, results):
         if failed_results:
             raise AnsibleModuleFailed(
                 f"Ansible module failed. If failure is expected, use `task_directives={{'ignore_errors': True}}` "
-                f"to avoid raising an exception. Details: {json.dumps(failed_results, indent=4)}"
+                f"to avoid raising an exception. Details: "
+                f"{json.dumps(failed_results, indent=4, default=str)}"
             )
 
     def _run(
@@ -344,13 +443,18 @@ def _run(
                 variable_manager=self.vm,
                 loader=self.loader
             )
+            # TQM requires a registered stdout callback name to satisfy
+            # load_callbacks(); we pass 'minimal' (one of Ansible's built-ins)
+            # then evict it from _callback_plugins before run() so nothing
+            # prints to stdout. Our own _JsonResultsCallback is attached as
+            # the sole receiver of task events.
             if ansible.__version__ >= '2.19.0':
                 tqm = TaskQueueManager(
                     inventory=self.im,
                     variable_manager=self.vm,
                     loader=self.loader,
                     passwords={},
-                    stdout_callback_name='json_results',
+                    stdout_callback_name='minimal',
                     run_tree=False,
                     forks=self.options.get("forks")
                 )
@@ -360,16 +464,30 @@ def _run(
                     variable_manager=self.vm,
                     loader=self.loader,
                     passwords={},
-                    stdout_callback='json_results',
+                    stdout_callback='minimal',
                     run_tree=False,
                     forks=self.options.get("forks")
                 )
             tqm.load_callbacks()
 
+            # Remove the built-in stdout callback (suppresses terminal output)
+            # and attach our explicit result collector in its place.
+            tqm._callback_plugins[:] = [
+                cb for cb in tqm._callback_plugins
+                if getattr(cb, 'CALLBACK_TYPE', None) != 'stdout'
+            ]
+            results_collector = _JsonResultsCallback()
+            # _init_callback_methods populates _implemented_callback_methods,
+            # which TQM.send_callback gates dispatch on. The plugin loader calls
+            # this implicitly when loading callbacks via the loader; since we
+            # construct our collector directly, we have to invoke it ourselves.
+            if hasattr(results_collector, '_init_callback_methods'):
+                results_collector._init_callback_methods()
+            tqm._callback_plugins.append(results_collector)
+
             tqm.run(play)
 
-            stdout_callback = tqm._stdout_callback
-            results = stdout_callback.results
+            results = results_collector.results
 
             # results is a dict: {hostname: [task_result_dict, ...], ...}
             # It makes sense to return this format of results for multiple hosts and multiple tasks