ErrataConnection Filter helper method - WIP

errata_tool/connector.py

@@ -310,3 +310,96 @@ def get_paginated_data(self, api_url):
             if page_number >= PAGE_LIMIT:
                 raise RuntimeError('hit pagination timeout: %d' % page_number)
         return data
+
+    def get_filter(self, endpoint, filter_arg, **kwargs):
+        """format and generate filter get request
+
+        expose a general filter helper method to format kwargs up
+        as parameters for ET filter request.  Then return generated
+        json object
+        """
+
+        if endpoint is None or filter_arg is None:
+            return None
+
+        url = endpoint + "?"
+        param_list = []
+        keys = list(kwargs)
+        keys.sort()
+        for k in keys:
+            v = kwargs[k]
+            if k in ('paginated'):
+                continue
+            if k in ('release', 'product'):
+                param_list.append("{0}[{1}][]={2}".format(filter_arg, k, v))
+            else:
+                param_list.append("{0}[{1}]={2}".format(filter_arg, k, v))
+        if self.debug:
+            print(param_list)
+        url = url + "&".join(param_list)
+        if endpoint == '/errata':
+            url = url + '&format=json'
+        if self.debug:
+            print(url)
+
+        if 'paginated' in kwargs and kwargs['paginated']:
+            return {'data': self.get_paginated_data(url)}
+
+        return self._get(url)
+
+    def get_releases_for_product(self, product_name_or_id,
+                                 return_ids_only=True):
+        """search for and return list of releases by name or id of product"""
+        args = {'is_active': 'true', 'enabled': 'true'}
+
+        try:
+            args['id'] = int(product_name_or_id)
+        except ValueError:
+            args['name'] = product_name_or_id
+
+        data = self.get_filter('/api/v1/releases', 'filter', **args)
+        if return_ids_only:
+            return [i['id'] for i in data['data']]
+
+        return data
+
+    def get_open_advisories_for_release(self, release_id):
+        data = self._get('/errata/errata_for_release/' +
+                         '{0}.json'.format(release_id))
+
+        ADVISORY_STATES = ('NEW_FILES', 'QE', 'REL_PREP', 'PUSH_READY')
+        advisory_ids = set()
+
+        for advisory_result in data:
+            if advisory_result['status'] in ADVISORY_STATES:
+                advisory_ids.add(advisory_result['id'])
+        return list(advisory_ids)
+
+    def get_open_advisories_for_release_filter(self, release_id,
+                                               return_ids_only=True):
+        """Return list of open advisories for a release either id's or json"""
+
+        data = self.get_filter(
+            '/errata', 'errata_filter[filter_params]',
+            show_type_RHBA=1, show_type_RHEA=1, show_type_RHSA=1,
+            show_state_NEW_FILES=1, show_state_QE=1, show_state_REL_PREP=1,
+            show_state_PUSH_READY=1, open_closed_option='exclude',
+            release=release_id)
+
+        if return_ids_only:
+            return [i['id'] for i in data]
+
+        return data
+
+    def get_errata_by_batch(self, batch_name_or_id):
+        """search for and return list of errata by name or id of batch"""
+        args = {}
+
+        try:
+            args['id'] = int(batch_name_or_id)
+        except ValueError:
+            args['name'] = batch_name_or_id
+
+        data = self.get_filter('/api/v1/batches', 'filter', **args)
+
+        return data

errata_tool/erratum.py

@@ -454,13 +454,10 @@ def externalTests(self, test_type=None):
         :param test_type: str, like "rpmdiff" or "covscan"
         :returns: a possibly-empty list of dicts, one per result.
         """
-        tmpl = '/api/v1/external_tests/?filter[active]=true'
-        tmpl += '&filter[errata_id]={errata_id}'
-        if test_type:
-            tmpl += '&filter[test_type]={test_type}'
-        url = tmpl.format(errata_id=self.errata_id, test_type=test_type)
-        data = self.get_paginated_data(url)
-        return data
+        response = self.get_filter('/api/v1/external_tests', 'filter',
+                                   active='true', errata_id=self.errata_id,
+                                   test_type=test_type, paginated=True)
+        return response['data']
 
     def _get_build_list(self, check_signatures=False):
         # Grab build list; store on a per-key basis

errata_tool/release.py

@@ -29,12 +29,13 @@ def __init__(self, **kwargs):
         self.refresh()
 
     def refresh(self):
-        url = self._url + '/api/v1/releases?'
+        params = {}
         if self.id is not None:
-            url += 'filter[id]=%s' % self.id
+            params['id'] = self.id
         elif self.name is not None:
-            url += 'filter[name]=%s' % self.name
-        result = self._get(url)
+            params['name'] = self.name
+        result = self.get_filter('/api/v1/releases', 'filter', **params)
+
         if len(result['data']) < 1:
             raise NoReleaseFoundError()
         if len(result['data']) > 1:

errata_tool/tests/conftest.py

@@ -76,6 +76,14 @@ def mock_put():
     return RequestRecorder()
 
 
+@pytest.fixture
+def sample_connector(monkeypatch, mock_get):
+    monkeypatch.delattr('requests.sessions.Session.request')
+    monkeypatch.setattr(ErrataConnector, '_auth', None)
+    monkeypatch.setattr(ErrataConnector, '_username', 'test')
+    monkeypatch.setattr(requests, 'get', mock_get)
+    return ErrataConnector()
+
 @pytest.fixture
 def advisory(monkeypatch, mock_get):
     monkeypatch.delattr('requests.sessions.Session.request')

errata_tool/tests/fixtures/errata.devel.redhat.com/api/v1/external_tests?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff

@@ -0,0 +1 @@
+external_tests/?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=1

errata_tool/tests/fixtures/errata.devel.redhat.com/api/v1/external_tests?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=1

@@ -0,0 +1 @@
+external_tests/?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=1

errata_tool/tests/fixtures/errata.devel.redhat.com/api/v1/external_tests?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=2

@@ -0,0 +1 @@
+external_tests/?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=2

errata_tool/tests/test_external_tests.py

@@ -6,7 +6,7 @@ class TestExternalTests(object):
     def test_external_tests_url(self, monkeypatch, mock_get, advisory):
         monkeypatch.setattr(requests, 'get', mock_get)
         advisory.externalTests(test_type='rpmdiff')
-        assert mock_get.response.url == 'https://errata.devel.redhat.com/api/v1/external_tests/?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=2'  # NOQA: E501
+        assert mock_get.response.url == 'https://errata.devel.redhat.com/api/v1/external_tests?filter[active]=true&filter[errata_id]=33840&filter[test_type]=rpmdiff&page[number]=2'  # NOQA: E501
 
     def test_external_tests_data(self, monkeypatch, mock_get, advisory):
         monkeypatch.setattr(requests, 'get', mock_get)

errata_tool/tests/test_filter.py

@@ -0,0 +1,33 @@
+import requests
+
+
+class TestFilter(object):
+
+    def test_filter_none(self, sample_connector):
+        assert sample_connector.get_filter(None, None) is None
+
+    def test_filter_sample(self, sample_connector):
+        assert sample_connector.get_filter(
+            '/api/v1/releases', 'filter', name='rhceph-3.1')
+
+    def test_filter_sample_check_url(self, monkeypatch, mock_get,
+                                     sample_connector):
+        monkeypatch.setattr(requests, 'get', mock_get)
+        assert sample_connector.get_filter(
+            '/api/v1/releases', 'filter', name='rhceph-3.1')
+        assert 'page' not in mock_get.response.url
+
+    def test_filter_url_paginated_false(self, monkeypatch,
+                                        mock_get, sample_connector):
+        monkeypatch.setattr(requests, 'get', mock_get)
+        assert sample_connector.get_filter(
+            '/api/v1/releases', 'filter', name='rhceph-3.1', paginated=False)
+        assert 'page' not in mock_get.response.url
+
+    def test_filter_sample_check_url_paginated(self, monkeypatch, mock_get,
+                                               sample_connector):
+        monkeypatch.setattr(requests, 'get', mock_get)
+        assert sample_connector.get_filter(
+            '/api/v1/external_tests', 'filter', errata_id='33840',
+            test_type='rpmdiff', active='true', paginated=True)
+        assert 'page' in mock_get.response.url