- [fedpkg-stage.pdc]

- url = https://pdc.stg.fedoraproject.org/

- 

- token = 

- [fedpkg.pdc]

- url = https://pdc.fedoraproject.org/

- 

- token = 

-         pdc_url = config.get('{0}.pdc'.format(name), 'url')

-                 *list(get_release_branches(pdc_url).values())))

-             verify_sls(pdc_url, sl_dict)

-                 *list(get_release_branches(pdc_url).values())))

-             if branch_info['name'] != name:

-                 continue

-             if branch_info['active']:

-             else:

-                 raise rpkgError('Cannot build from stream branch {0} as it is '

-                                 'inactive.'.format(name))

-         server_url = self.config.get('{0}.pdc'.format(self.name), 'url')

-         stream_branches = get_stream_branches(server_url, self.cmd.repo_name)

-             self.cmd.repo_name, [item['name'] for item in stream_branches])

-         server_url = self.config.get('{0}.pdc'.format(self.name), 'url')

- from six.moves.urllib.parse import urlencode, urlparse

- def query_pdc(server_url, endpoint, params, timeout=60):

-     api_url = '{0}/rest_api/v1/{1}/'.format(

-         server_url.rstrip('/'), endpoint.strip('/'))

-     query_args = params

-     while True:

-         try:

-             rv = requests.get(api_url, params=query_args, timeout=60)

-         except ConnectionError as error:

-             error_msg = ('The connection to PDC failed while trying to get '

-                          'the active release branches. The error was: {0}'

-                          .format(str(error)))

-             raise rpkgError(error_msg)

- 

-         if not rv.ok:

-             base_error_msg = ('The following error occurred while trying to '

-                               'get the active release branches in PDC: {0}')

-             raise rpkgError(base_error_msg.format(rv.text))

- 

-         rv_json = rv.json()

-         for item in rv_json['results']:

-             yield item

- 

-         if rv_json['next']:

-             # Clear the query_args because they are baked into the "next" URL

-             query_args = {}

-             api_url = rv_json['next']

-         else:

-             # We've gone through every page, so we can return the found

-             # branches

-             break

- 

- 

- def get_sl_type(url, sl_name):

-     """

-     Gets the service level (SL) type from PDC

-     :param url: a string of the URL to PDC

-     :param sl_name: a string of the SL name

-     :return: a dictionary representing the SL type or None

-     """

-     api_url = '{0}/rest_api/v1/component-sla-types/'.format(url.rstrip('/'))

-     api_url_w_args = '{0}?{1}'.format(api_url, urlencode({'name': sl_name}))

-         rv = requests.get(api_url_w_args, timeout=60)

-         error_msg = ('The connection to PDC failed while trying to validate '

-                      'the passed in service level. The error was: {0}'

-         base_error_msg = ('The following error occurred while validating the '

-                           'passed in service level in PDC: {0}')

-     if rv_json['count'] == 1:

-         return rv_json['results'][0]

-     else:

-         return None

-     Get the active Fedora release branches from PDC

-     :param  str url: a string of the URL to PDC

-     query_args = {

-         'fields': ['short', 'version'],

-         'active': True

-     }

- 

-     for product_version in query_pdc(

-             server_url, 'product-versions', params=query_args):

-         short_name = product_version['short']

-         version = product_version['version']

- 

-         # If the version is not a digit we can ignore it (e.g. rawhide)

-         if not version.isdigit():

-         if short_name == 'epel':

-             prefix = 'el' if version == '6' else 'epel'

-         elif short_name == 'fedora':

-             prefix = 'f'

- 

-         release = '{0}{1}'.format(prefix, version)

-         releases.setdefault(short_name, []).append(release)

- 

-     return releases

- def verify_sls(pdc_url, sl_dict):

-     Verifies that the service levels are properly formatted and exist in PDC

-     :param pdc_url: a string of the URL to PDC

-         sl_obj = get_sl_type(pdc_url, sl)

-         if not sl_obj:

-             raise rpkgError('The SL "{0}" is not in PDC'.format(sl))

- 

- def get_stream_branches(server_url, package_name):

-     :param str server_url: PDC server URL.

-     :return: a list of stream branches. Each element in the list is a dict

-         containing branch property name and active.

-     :rtype: list[dict]

-     query_args = {

-         'global_component': package_name,

-         'fields': ['name', 'active'],

-     }

-     branches = query_pdc(

-         server_url, 'component-branches', params=query_args)

-     # When write this method, endpoint component-branches contains not only

-     # stream branches, but also regular release branches, e.g. rawhide/main, f28.

-     # Please remember to review the data regularly, there are only stream

-     # branches, or some new replacement of PDC fixes the issue as well, it

-     # should be ok to remove if from this list.

-     for item in branches:

-         if item['name'] in ('rawhide', 'main'):

-         elif re.match(r'^(f|el)\d+$', item['name']):

-         elif 'epel7' == item['name']:

-         elif re.match(r'^epel\d+-next$', item['name']):

- [fedpkg-stage.pdc]

- url = https://pdc.stg.example.com/

- 

- [fedpkg.pdc]

- url = https://pdc.example.com/

- 

-     def test_stream_branch_is_inactive(self):

-         with patch('sys.argv', new=self.fake_cmd):

-             cli = self.new_cli()

- 

-         six.assertRaisesRegex(

-             self, rpkgError, 'Cannot build from stream branch',

-             cli.is_stream_branch, [{'name': '10', 'active': False}], '10')

- 

-             [{'name': '8', 'active': True}, {'name': '10', 'active': True}],

-             '8')

-             'epel': ['el6', 'epel7'],

-         get_stream_branches.return_value = [{'name': '8', 'active': True}]

-     @patch('requests.get')

-     def test_get_sl_type(self, mock_get):

-         """Test get_sl_type"""

-         sl_type = {

-             'id': 1,

-             'name': 'security_fixes',

-             'description': 'security_fixes',

-         }

-         mock_rv = Mock()

-         mock_rv.ok = True

-         mock_rv.json.return_value = {

-             'count': 1,

-             'results': [sl_type]

-         }

-         mock_get.return_value = mock_rv

-         rv = utils.get_sl_type('http://pdc.local/', 'securty_fixes')

-         self.assertEqual(rv, sl_type)

- 

-     @patch('requests.get')

-     def test_get_sl_type_pdc_error(self, mock_request_get):

-         """Test get_sl_type when PDC errors"""

-         mock_rv = Mock()

-         mock_rv.ok = False

-         mock_rv.text = 'Some error'

-         mock_request_get.return_value = mock_rv

-         try:

-             utils.get_sl_type('http://pdc.local/', 'securty_fixes')

-             assert False, 'rpkgError not raised'

-         except rpkgError as error:

-             expected_error = ('The following error occurred while validating '

-                               'the passed in service level in PDC: Some error')

-             self.assertEqual(str(error), expected_error)

- 

-     @patch('fedpkg.utils.get_sl_type')

-     def test_verify_sls(self, mock_get_sl_type):

-         mock_get_sl_type.return_value = {

-             'id': 1,

-             'name': 'security_fixes',

-             'description': 'security_fixes',

-         }

-         sls = {'security_fixes': '2222-12-01'}

-             utils.verify_sls('http://pdc.local/', sls)

-     @patch('fedpkg.utils.get_sl_type')

-     def test_verify_sls_eol_expired(self, mock_get_sl_type):

-         mock_get_sl_type.return_value = {

-             'id': 1,

-             'name': 'security_fixes',

-             'description': 'security_fixes',

-         }

-         sls = {'security_fixes': '2001-12-01'}

-             utils.verify_sls('http://pdc.local/', sls)

-             assert False, 'An rpkgError exception was not raised'

-                 sls = {'security_fixes': eol, 'bug_fixes': eol}

-                 utils.verify_sls('http://pdc.example.com/', sls)

-                 assert False, 'An rpkgError exception was not raised'

-         # This abbreviated data returned from the product-versions PDC API

-         mock_rv.json.return_value = {

-             'count': 7,

-             'next': None,

-             'previous': None,

-             'results': [

-                 {'short': 'epel', 'version': '6'},

-                 {'short': 'epel', 'version': '7'},

-                 {'short': 'fedora', 'version': '25'},

-                 {'short': 'fedora', 'version': '26'},

-                 {'short': 'fedora', 'version': '27'},

-                 {'short': 'fedora', 'version': '28'},

-                 {'short': 'fedora', 'version': 'rawhide'}

-             ]

-         }

-         expected = set(['el6', 'epel7', 'f25', 'f26', 'f27', 'f28'])

-             'epel': ['el6', 'epel7'],

-             'fedora': ['f25', 'f26', 'f27', 'f28'],

-         actual = utils.get_release_branches('http://pdc.local')

-         self.assertEqual(expected, actual)

- @patch('requests.get')

- class TestGetServiceLevelType(unittest.TestCase):

-     """Test get_sl_type"""

- 

-     def test_raise_error_if_connection_error_to_pdc(self, get):

-         get.side_effect = ConnectionError

- 

-         six.assertRaisesRegex(

-             self, rpkgError, 'The connection to PDC failed',

-             utils.get_sl_type, 'http://localhost/', 'bug_fixes:2020-12-01')

- 

-     def test_sl_type_not_exist(self, get):

-         rv = Mock(ok=True)

-         rv.json.return_value = {'count': 0}

-         get.return_value = rv

- 

-         sl_type = utils.get_sl_type('http://localhost/',

-                                     'bug_fixes:2020-12-01')

-         self.assertIsNone(sl_type)

- 

-     def test_raise_error_if_response_not_ok(self, get):

-         get.return_value = Mock(ok=False)

- 

-         six.assertRaisesRegex(

-             self, rpkgError, 'The following error occurred',

-             utils.get_sl_type, 'http://localhost/', 'bug_fixes:2020-12-01')

- 

- 

-             self, rpkgError, 'The EOL date .+ is in an invalid format',

-             utils.verify_sls, 'http://localhost/', {'bug_fixes': '2018/7/21'})

- 

-     @freeze_time('2018-01-01')

-     @patch('requests.get')

-     def test_sl_not_exist(self, get):

-         rv = Mock(ok=True)

-         rv.json.return_value = {'count': 0}

- 

-         six.assertRaisesRegex(

-             self, rpkgError, 'The SL .+ is not in PDC',

-             utils.verify_sls, 'http://localhost/', {'some_sl': '2018-06-01'})

-     @freeze_time('2018-01-01')

-     @patch('requests.get')

-             'http://localhost/',

-             {

-                 'bug_fixes': '2018-06-01',

-                 'security_fixes': '2018-12-01'

-             })

- @patch('requests.get')

- class TestQueryPDC(unittest.TestCase):

-     """Test utils.query_pdc"""

-         result = utils.query_pdc('http://localhost/', 'endpoint', {})

-             self, rpkgError, 'The connection to PDC failed',

-         result = utils.query_pdc('http://localhost/', 'endpoint', {})

-             {'results': ['item1', 'item2'],

-              'next': 'http://localhost/?page=2'},

-             {'results': ['item3'], 'next': None}

-         result = utils.query_pdc('http://localhost/', 'endpoint', {})

-         self.assertEqual(['item1', 'item2', 'item3'], list(result))

-     def test_fedora_and_epel_branches_are_filtered_out(self, get):

-         rv.json.return_value = {

-             'results': [

-                 {'name': '8'},

-                 {'name': '10'},

-                 {'name': 'f28'},

-                 {'name': 'epel7'},

-                 {'name': 'rawhide'},

-                 {'name': 'epel8'},

-             ],

-             'next': None

-         }

-         result = utils.get_stream_branches('http://localhost/', 'pkg')

-         self.assertEqual([{'name': '8'}, {'name': '10'}, {'name': 'epel8'}], list(result))