aoh

commit c174b962cda5e205a4d597489e2c3bf442811f78
parent a944005a50cb94d88520af530397e35eec6c3108
Author: Asher Morgan <asher@ashermorgan.net>
Date:   Wed,  2 Sep 2026 18:36:27 -0700

Minor cleanup

Diffstat:
M
aoh/client.py
 | 
5
++++-
M
aoh/connection_plugins/aoh.py
 | 
41
+++++++++++++++++++++--------------------
M
aoh/runner.py
 | 
14
++++++++------

3 files changed, 33 insertions(+), 27 deletions(-)
diff --git a/aoh/client.py b/aoh/client.py
@@ -23,6 +23,9 @@ PASSWORD_PROMPTS = {
 # Set 10s EXEC keep alive interval (see also TIMEOUT in aoh.py)
 KEEP_ALIVE_INTERVAL = 10
 
+# How long to sleep after receiving an empty responses
+WAIT_INTERVAL = 0.1
+
 
 class ClientError(Exception):
     """Raised for miscellaneous client errors."""
@@ -165,7 +168,7 @@ def runner_loop(runner_url, cookies):
             req['id'] = res['id']
         else:
             req = {}
-            time.sleep(0.1)
+            time.sleep(WAIT_INTERVAL)
 
 
 def create_runner(playbook, args):
diff --git a/aoh/connection_plugins/aoh.py b/aoh/connection_plugins/aoh.py
@@ -46,9 +46,10 @@ class Connection(ConnectionBase):
     def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
         super().__init__(*args, **kwargs)
 
-        self.sendbuf = None
-        self.recvbuf = None
-        self.recvpoll = None
+        self._sendbuf = None
+        self._recvbuf = None
+        self._recvpoll = None
+        self._connected = False
 
 
     def _connect(self) -> Connection:  # noqa: F821
@@ -75,11 +76,11 @@ class Connection(ConnectionBase):
             if not os.path.exists(SENDBUF_PATH):
                 os.mkfifo(SENDBUF_PATH)
 
-            self.recvbuf = open(RECVBUF_PATH, 'r')  # noqa: SIM115
-            self.sendbuf = open(SENDBUF_PATH, 'w')  # noqa: SIM115
+            self._recvbuf = open(RECVBUF_PATH, 'r')  # noqa: SIM115
+            self._sendbuf = open(SENDBUF_PATH, 'w')  # noqa: SIM115
 
-            self.recvpoll = select.poll()
-            self.recvpoll.register(self.recvbuf, select.POLLIN)
+            self._recvpoll = select.poll()
+            self._recvpoll.register(self._recvbuf, select.POLLIN)
 
             self._connected = True
 
@@ -89,21 +90,21 @@ class Connection(ConnectionBase):
     def _send_message(self, msg: dict) -> dict:
         """Send a message to the client and wait for a response."""
 
-        assert self.sendbuf is not None
-        assert self.recvbuf is not None
-        assert self.recvpoll is not None
+        assert self._sendbuf is not None
+        assert self._recvbuf is not None
+        assert self._recvpoll is not None
 
         msg['id'] = str(uuid4())
-        self.sendbuf.write(json.dumps(msg) + '\n')
-        self.sendbuf.flush()
+        self._sendbuf.write(json.dumps(msg) + '\n')
+        self._sendbuf.flush()
 
         while True:
-            if not self.recvpoll.poll(TIMEOUT * 1000):
+            if not self._recvpoll.poll(TIMEOUT * 1000):
                 raise AnsibleConnectionFailure('Timed out waiting for AoH '
                                                'response')
 
             try:
-                res = json.loads(self.recvbuf.readline())
+                res = json.loads(self._recvbuf.readline())
             except json.decoder.JSONDecodeError:
                 raise AnsibleError('Received corrupt AoH message')
 
@@ -117,11 +118,11 @@ class Connection(ConnectionBase):
             if not res.get('keep-alive'):
                 return res
 
-            self.sendbuf.write(json.dumps({
+            self._sendbuf.write(json.dumps({
                 'id': msg['id'],
                 'keep-alive': True,
             }) + '\n')
-            self.sendbuf.flush()
+            self._sendbuf.flush()
 
 
     def exec_command(self, cmd: str, in_data: bytes | None = None,
@@ -191,10 +192,10 @@ class Connection(ConnectionBase):
                     f'{self.get_option('runner')}',
                     host=self._play_context.remote_addr)
 
-        if self.recvbuf:
-            self.recvbuf.close()
-        if self.sendbuf:
-            self.sendbuf.close()
+        if self._recvbuf:
+            self._recvbuf.close()
+        if self._sendbuf:
+            self._sendbuf.close()
 
         self._connected = False
 
diff --git a/aoh/runner.py b/aoh/runner.py
@@ -30,6 +30,8 @@ class Runner:
         self.id = str(uuid4()) # Runner ID (set to None after teardown)
         self.playbook = playbook
         self.host = self.playbook.host or host or 'aoh_node'
+        self.args = args
+        self.passwords = passwords
 
         self._lock = Lock() # Used to protect all public methods
 
@@ -51,7 +53,7 @@ class Runner:
             with open(f'{self._DIR}/hostname', 'w') as f:
                 f.write(f'{self.host}\n')
 
-            self._start_runner(args, passwords)
+            self._start_runner()
 
             # We assume that ansible-runner will eventually create its log file
             while not os.path.exists(self._LOGS_PATH):
@@ -62,7 +64,7 @@ class Runner:
             raise
 
 
-    def _start_runner(self, args, passwords):
+    def _start_runner(self):
         """Configure and start the underlying ansible-runner."""
 
         # Set ansible_connection=aoh for the client host only. Note that we
@@ -102,9 +104,10 @@ class Runner:
             'ansible_aoh_dir': self._DIR,
         }
         pw_prompt_answers = {}
-        for pw_type in passwords:
-            pw_prompt_answers[_PASSWORD_PROMPTS[pw_type]] = passwords[pw_type]
-        cmdline = ' '.join(shlex.quote(arg) for arg in args)
+        for pw_type in self.passwords:
+            pw_prompt_answers[_PASSWORD_PROMPTS[pw_type]] = \
+                    self.passwords[pw_type]
+        cmdline = ' '.join(shlex.quote(arg) for arg in self.args)
         cmdline += ' ' + self.playbook.cmdline
 
         self._thread, self._runner = ansible_runner.run_async(
@@ -141,7 +144,6 @@ class Runner:
             if self.id is None:
                 # Runner has already been torn down
                 return {
-                    'finished': True,
                     'err': 'Client timeout',
                 }