From fb1c154acf6570c567571fcb84ea0bafdd0a24a0 Mon Sep 17 00:00:00 2001 From: David Salvisberg Date: Tue, 3 Sep 2024 12:26:21 +0200 Subject: [PATCH] Only sets `local` connection if the port is `22` Adds a Quick Start section to README and sets proper. Gets some of the container tests working again. --- README.rst | 14 ++++++++++++- src/suitable/inventory.py | 11 +++++----- tests/conftest.py | 44 +++++++++++++++++++++++++++++++++------ tests/test_api.py | 33 +++++++++++++++++++---------- 4 files changed, 79 insertions(+), 23 deletions(-) diff --git a/README.rst b/README.rst index 4996ee5..9794a5a 100644 --- a/README.rst +++ b/README.rst @@ -10,6 +10,18 @@ Documentation ``_ +Quick Start +------------- + +Suitable provides a simple wrapper over Ansible's internal API, that allows you to use Ansible programmatically. + +.. code-block:: pycon + + >>> from suitable import Api + >>> api = Api('localhost') + >>> api.command('whoami').stdout() + 'myuser' + Warning ------- @@ -32,7 +44,7 @@ are favored over old ones. Run Tests --------- -.. code-block:: python +.. code-block:: console pip install tox tox diff --git a/src/suitable/inventory.py b/src/suitable/inventory.py index 112e0ec..42bd92f 100644 --- a/src/suitable/inventory.py +++ b/src/suitable/inventory.py @@ -27,15 +27,15 @@ def add_host(self, server: str, host_variables: HostVariables) -> None: # [ipv6]:port if server.startswith('['): - host, port = server.rsplit(':', 1) + host, port_str = server.rsplit(':', 1) self[server]['ansible_host'] = host = host.strip('[]') - self[server]['ansible_port'] = int(port) + self[server]['ansible_port'] = int(port_str) # host:port elif server.count(':') == 1: - host, port = server.split(':', 1) + host, port_str = server.split(':', 1) self[server]['ansible_host'] = host - self[server]['ansible_port'] = int(port) + self[server]['ansible_port'] = int(port_str) # Add vars self[server].update(host_variables) @@ -44,7 +44,8 @@ def add_host(self, server: str, host_variables: HostVariables) -> None: if not self.ansible_connection: # Get hostname (either ansible_host or server) host = self[server].get('ansible_host', server) - if host in ('localhost', '127.0.0.1', '::1'): + port = self[server].get('ansible_port', 22) + if host in ('localhost', '127.0.0.1', '::1') and port == 22: self[server]['ansible_connection'] = 'local' def add_hosts(self, servers: Hosts) -> None: diff --git a/tests/conftest.py b/tests/conftest.py index 152d103..05ff04f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,10 @@ +import paramiko import port_for import pytest import shutil import subprocess import tempfile +import time from uuid import uuid4 from suitable import Api @@ -37,7 +39,7 @@ def spawn_api(self, api_class, **kwargs): options.update(kwargs) return api_class( - '%s:%s' % (self.host, self.port), + f'{self.host}:{self.port}', ** options ) @@ -55,16 +57,46 @@ def tempdir(): shutil.rmtree(tempdir) +def wait_for_sshd(host, port): + client = paramiko.client.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + timeout = 5 + started = time.monotonic() + while time.monotonic() - started < timeout: + try: + client.connect( + host, + port, + allow_agent=False, + look_for_keys=False + ) + except paramiko.ssh_exception.SSHException as e: + # socket is open, but no SSH service responded + if not e.args[0].startswith('Error reading SSH protocol banner'): + return + except paramiko.ssh_exception.NoValidConnectionsError: + pass + + finally: + client.close() + + time.sleep(.5) + + raise RuntimeError('Failed to initalize sshd in docker container') + + @pytest.fixture(scope="function") def container(): port = port_for.select_random() - name = 'suitable-container-%s' % uuid4().hex + name = f'suitable-container-{uuid4().hex}' subprocess.check_call(( - 'docker', 'run', '-d', '--rm', '-it', '--name', name, - '-p', '127.0.0.1:%d:22/tcp' % port, - 'rastasheep/ubuntu-sshd:18.04' + 'docker', 'run', '-d', '--rm', '-it', + '--name', name, + '-p', f'127.0.0.1:{port}:22/tcp', + 'takeyamajp/ubuntu-sshd:ubuntu22.04' )) + wait_for_sshd('127.0.0.1', port) yield Container('127.0.0.1', port, 'root', 'root') @@ -73,4 +105,4 @@ def container(): @pytest.fixture(scope="function", params=APIS) def api(request, container): - yield getattr(container, '%s_api' % request.param)(connection='paramiko') + yield getattr(container, f'{request.param}_api')(connection='paramiko') diff --git a/tests/test_api.py b/tests/test_api.py index 69e1b91..66bba96 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -15,6 +15,15 @@ def test_auto_localhost(): host = Api('localhost') assert host.inventory['localhost']['ansible_connection'] == 'local' + +def test_auto_localhost_different_port(): + host = Api('localhost:8888') + assert host.inventory['localhost:8888']['ansible_host'] == 'localhost' + assert host.inventory['localhost:8888']['ansible_port'] == 8888 + assert 'ansible_connection' not in host.inventory['localhost:8888'] + + +def test_smart_connection(): host = Api('localhost', connection='smart') assert 'ansible_connection' not in host.inventory['localhost'] assert host.options.connection == 'smart' @@ -290,26 +299,28 @@ def test_dict_args(tempdir): api.set_stats(data={'foo': 'bar'}) -@pytest.mark.skip() +def test_assert_alias(): + api = Api('localhost') + api.assert_(that=[ + "'bar' != 'foo'", + "'bar' == 'bar'" + ]) + + def test_disable_hostkey_checking(api): api.host_key_checking = False assert api.command('whoami').stdout() == 'root' -@pytest.mark.skip() -def test_enable_hostkey_checking_vanilla(container): - # if we do not use 'paramiko' here, we get the following error: - # > Using a SSH password instead of a key is not possible because Host Key - # > checking is enabled and sshpass does not support this. - # > Please add this host's fingerprint to your known_hosts file to - # > manage this host. - api = container.vanilla_api(connection='paramiko') - +def test_enable_hostkey_checking(api): with pytest.raises(UnreachableError): assert api.command('whoami').stdout() == 'root' -@pytest.mark.skip() +@pytest.mark.skip( + 'opening multiple connections to the same server ' + 'does not appear to currently work' +) def test_interleaving(container): # make sure we can interleave calls of different API objects password = token_hex(16)