diff --git a/requirements.txt b/requirements.txt index 36f7907..a5fae92 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ pytest-testinfra paramiko +apypie diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f01ab62 --- /dev/null +++ b/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) diff --git a/tests/foreman_api_test.py b/tests/foreman_api_test.py new file mode 100644 index 0000000..9b8e449 --- /dev/null +++ b/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