From fbc486a44cc69418fac2f6bd0509ed87c7bd5a9d Mon Sep 17 00:00:00 2001 From: shenxianpeng Date: Mon, 1 Jun 2026 23:18:55 +0300 Subject: [PATCH] fix: handle ansible encoding errors and update rocky9 package assertion - ansible.py: handle ansible error responses containing only 'msg' key (no stdout/stderr) to avoid KeyError when commands fail at transport layer due to encoding issues - test_backends.py: update encoding test to accept both old-style ansible error messages and new surrogate-rejection errors from newer ansible - test_modules.py: update rocky linux 9 openssh-server version prefix from 8. to 9. (new image ships openssh-server 9.9p1) --- test/test_backends.py | 3 ++- test/test_modules.py | 2 +- testinfra/backend/ansible.py | 9 +++++++-- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/test/test_backends.py b/test/test_backends.py index 94d9295b..7c5809e3 100644 --- a/test/test_backends.py +++ b/test/test_backends.py @@ -87,7 +87,8 @@ def test_encoding(host): elif host.backend.get_connection_type() == "ansible" and host.backend.force_ansible: # XXX: this encoding issue comes directly from ansible # not sure how to handle this... - assert ( + assert cmd.failed + assert "surrogates not allowed" in cmd.stderr or ( cmd.stderr == "ls: impossible d'acc\udce9der \udce0 '/é': Aucun fichier ou dossier de ce type" ) diff --git a/test/test_modules.py b/test/test_modules.py index 1057c646..6be6c5d0 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -39,7 +39,7 @@ def test_package(host, docker_image): assert not host.package("zsh").is_installed ssh = host.package("openssh-server") version = { - "rockylinux9": "8.", + "rockylinux9": "9.", "debian_bookworm": "1:9.2", }[docker_image] assert ssh.is_installed diff --git a/testinfra/backend/ansible.py b/testinfra/backend/ansible.py index 70079f76..36e3c62c 100644 --- a/testinfra/backend/ansible.py +++ b/testinfra/backend/ansible.py @@ -61,11 +61,16 @@ def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult: data = json.loads(out["module_stdout"]) stdout = data["stdout"] stderr = data["stderr"] - else: + elif "stdout" in out: # bw compat stdout = out["stdout"] stderr = out["stderr"] - return self.result(out["rc"], self.encode(command), stdout, stderr) + else: + # Ansible command failed (e.g. encoding error) with only a msg + stdout = "" + stderr = out.get("msg", "") + rc = out.get("rc", 1) + return self.result(rc, self.encode(command), stdout, stderr) def run_ansible( self, module_name: str, module_args: Optional[str] = None, **kwargs: Any