use the api to create a set of reusable fixtures

requirements.txt

@@ -1,2 +1,3 @@
 pytest-testinfra
 paramiko
+apypie

tests/conftest.py

@@ -0,0 +1,187 @@
+import time
+import uuid
+
+import apypie
+import paramiko
+import pytest
+import requests.exceptions
+
+
+@pytest.fixture(scope="module")
+def ssh_config():
+    config = paramiko.SSHConfig.from_path('./.vagrant/ssh-config')
+    return config.lookup('quadlet')
+
+
+def to_native(thing):
+    return thing
+
+class ForemanApiException(Exception):
+
+    def __init__(self, msg, error=''):
+        self.msg = msg
+        self.error = error
+        return super(ForemanApiException, self).__init__()
+
+    def __repr__(self, /):
+        return f'{self.__class__.__name__}: {self.msg} - {self.error}'
+
+    def __str__(self, /):
+        return f'{self.__class__.__name__}: {self.msg} - {self.error}'
+
+    @classmethod
+    def from_exception(cls, exc, msg):
+        fail = {'msg': msg}
+        if isinstance(exc, requests.exceptions.HTTPError):
+            try:
+                response = exc.response.json()
+                if 'error' in response:
+                    fail['error'] = response['error']
+                else:
+                    fail['error'] = response
+            except Exception:
+                fail['error'] = exc.response.text
+        return cls(**fail)
+
+
+class ForemanApi(apypie.Api):
+
+    def __init__(self, **kwargs):
+        kwargs['api_version'] = 2
+        self.check_mode = False
+        self.task_timeout = 60
+        self.task_poll = 1
+        return super(ForemanApi, self).__init__(**kwargs)
+
+    def set_changed(self):
+        pass
+
+    def _resource(self, resource):
+        if resource not in self.resources:
+            raise Exception("The server doesn't know about {0}, is the right plugin installed?".format(resource))
+        return self.resource(resource)
+
+    def _resource_call(self, resource, *args, **kwargs):
+        return self._resource(resource).call(*args, **kwargs)
+
+    def _resource_prepare_params(self, resource, action, params):
+        api_action = self._resource(resource).action(action)
+        return api_action.prepare_params(params)
+
+    def resource_action(self, resource, action, params, options=None, data=None, files=None,
+                        ignore_check_mode=False, record_change=True, ignore_task_errors=False):
+        resource_payload = self._resource_prepare_params(resource, action, params)
+        if options is None:
+            options = {}
+        try:
+            result = None
+            if ignore_check_mode or not self.check_mode:
+                result = self._resource_call(resource, action, resource_payload, options=options, data=data, files=files)
+                is_foreman_task = isinstance(result, dict) and 'action' in result and 'state' in result and 'started_at' in result
+                if is_foreman_task:
+                    result = self.wait_for_task(result, ignore_errors=ignore_task_errors)
+        except Exception as e:
+            msg = 'Error while performing {0} on {1}: {2}'.format(
+                action, resource, to_native(e))
+            raise ForemanApiException.from_exception(e, msg) from e
+        if record_change and not ignore_check_mode:
+            # If we were supposed to ignore check_mode we can assume this action was not a changing one.
+            self.set_changed()
+        return result
+
+    def wait_for_task(self, task, ignore_errors=False):
+        duration = self.task_timeout
+        while task['state'] not in ['paused', 'stopped']:
+            duration -= self.task_poll
+            if duration <= 0:
+                raise ForemanApiException(msg="Timeout waiting for Task {0}".format(task['id']))
+            time.sleep(self.task_poll)
+
+            resource_payload = self._resource_prepare_params('foreman_tasks', 'show', {'id': task['id']})
+            task = self._resource_call('foreman_tasks', 'show', resource_payload)
+        if not ignore_errors and task['result'] != 'success':
+            raise ForemanApiException(msg='Task {0}({1}) did not succeed. Task information: {2}'.format(task['action'], task['id'], task['humanized']['errors']))
+        return task
+
+    def create(self, resource, desired_entity, params=None):
+        """
+        Create entity with given properties
+
+        :param resource: Plural name of the api resource to manipulate
+        :type resource: str
+        :param desired_entity: Desired properties of the entity
+        :type desired_entity: dict
+        :param params: Lookup parameters (i.e. parent_id for nested entities)
+        :type params: dict, optional
+
+        :return: The new current state of the entity
+        :rtype: dict
+        """
+        payload = desired_entity.copy()
+        if not self.check_mode:
+            if params:
+                payload.update(params)
+            return self.resource_action(resource, 'create', payload)
+        else:
+            fake_entity = desired_entity.copy()
+            fake_entity['id'] = -1
+            self.set_changed()
+            return fake_entity
+
+    def delete(self, resource, current_entity, params=None):
+        """
+        Delete a given entity
+
+        :param resource: Plural name of the api resource to manipulate
+        :type resource: str
+        :param current_entity: Current properties of the entity
+        :type current_entity: dict
+        :param params: Lookup parameters (i.e. parent_id for nested entities)
+        :type params: dict, optional
+
+        :return: The new current state of the entity
+        :rtype: Union[dict,None]
+        """
+        payload = {'id': current_entity['id']}
+        if params:
+            payload.update(params)
+        entity = self.resource_action(resource, 'destroy', payload)
+
+        # this is a workaround for https://projects.theforeman.org/issues/26937
+        if entity and isinstance(entity, dict) and 'error' in entity and 'message' in entity['error']:
+            raise ForemanApiException(msg=entity['error']['message'])
+
+        return None
+
+    def ping(self):
+        return self.resource('ping').call('ping')
+
+
+@pytest.fixture(scope="module")
+def foremanapi(ssh_config):
+    return ForemanApi(
+        uri=f'https://{ssh_config['hostname']}',
+        username='admin',
+        password='changeme',
+        verify_ssl=False,
+    )
+
+@pytest.fixture
+def organization(foremanapi):
+    org = foremanapi.create('organizations', {'name': str(uuid.uuid4())})
+    yield org
+    foremanapi.delete('organizations', org)
+
+@pytest.fixture
+def product(organization, foremanapi):
+    prod = foremanapi.create('products', {'name': str(uuid.uuid4()), 'organization_id': organization['id']})
+    yield prod
+    foremanapi.delete('products', prod)
+
+@pytest.fixture
+def repository(product, organization, foremanapi):
+    # This is broken as Katello currently does not use the right CA when talking to Pulp
+    # https://github.com/Katello/katello/blob/master/app/models/katello/concerns/smart_proxy_extensions.rb#L254-L265
+    repo = foremanapi.create('repositories', {'name': str(uuid.uuid4()), 'product_id': product['id'], 'content_type': 'yum'})
+    yield repo
+    foremanapi.delete('repositories', repo)

tests/foreman_api_test.py

@@ -0,0 +1,8 @@
+def test_foreman_organization(organization):
+    assert organization
+
+def test_foreman_product(product):
+    assert product
+
+def test_foreman_repository(repository):
+    assert repository