From 3fd50299a657480bfaca19e02cb1d0dca91eb45b Mon Sep 17 00:00:00 2001 From: Scott Bahling Date: Mon, 3 Feb 2020 08:00:17 +0100 Subject: Add remove_flag method --- obsapi/core.py | 58 ++++++++++++++++++++++++++++++++++++++--------- test/test_api.py | 68 +++++++++++++++++++++++++++++++++++++++++++++----------- 2 files changed, 102 insertions(+), 24 deletions(-) diff --git a/obsapi/core.py b/obsapi/core.py index 748044c..8d7d995 100644 --- a/obsapi/core.py +++ b/obsapi/core.py @@ -394,10 +394,7 @@ class ObsApi(object): if srcfile.name.endswith('.spec'): yield self.source.get(prj, pkg, srcfile.name) - def set_flag(self, prj, pkg=None, flag=None): - if flag is None: - return None - + def __flag_to_params(self, flag): try: flag_type = getattr(flag, 'flag', flag.get('flag', None)) if flag_type is None: @@ -409,11 +406,50 @@ class ObsApi(object): if status not in ['enable', 'disable']: raise ValueError('flag status expected "enable" or "disable" got %s' % status) - params = dict(cmd='set_flag', - flag=flag_type, - status=status, - repository=getattr(flag, 'repository', flag.get('repository', None)), - arch=getattr(flag, 'arch', flag.get('arch', None)), - ) + return dict(flag=flag_type, + status=status, + repository=getattr(flag, 'repository', flag.get('repository', None)), + arch=getattr(flag, 'arch', flag.get('arch', None)), + ) + + def parse_status_code(self, xml): + status_code = etree.fromstring(xml) + code = status_code.get('code', None) + summary = status_code.find('summary') + if summary is not None: + summary = summary.text + + return (code, summary) + + def __check_status(self, r): + if r.status_code == 400: + code, summary = self.parse_status(r.text) + msg = '{}: {}'.format(code.replace('_', ' ').capitalize(), summary) + raise Exception(msg) + return r + + def set_flag(self, prj, pkg=None, flag=None): + if flag is None: + return None - return self.source.post(prj, pkg, **params) + params = self.__flag_to_params(flag) + params['cmd'] = 'set_flag' + + r = self.source.post(prj, pkg, **params) + try: + return self.__check_status(r) + except Exception as e: + raise e + + def remove_flag(self, prj, pkg=None, flag=None): + if flag is None: + return None + + params = self.__flag_to_params(flag) + params['cmd'] = 'remove_flag' + + r = self.source.post(prj, pkg, **params) + try: + return self.__check_status(r) + except Exception as e: + raise e diff --git a/test/test_api.py b/test/test_api.py index 3fbf349..52c6796 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -522,26 +522,68 @@ def test_get_project_spec_files(requests_mock, project_directory, package_direct assert list(specfiles) == expected_specfiles -# POST /source/?cmd=set_flag&repository=:opt&arch=:opt&flag=flag&status=status -def test_set_project_flags(requests_mock): - matcher = re.compile(r'/source/.*') - requests_mock.post(matcher) - params = dict(cmd='set_flag', flag='publish', status='enable') - - response = api.set_flag(prj, flag=params) - assert response.status_code == 200 - - url = requests_mock.request_history[0].url +def check_url_query_params(url, params): query = parse_qs(urlparse(url).query) + params = {k: v for k, v in params.items() if v is not None} assert query.keys() == params.keys() for key, values in query.items(): assert len(values) == 1 assert params.get(key, None) == query.get(key, [])[0] +# POST /source/?cmd=set_flag&repository=:opt&arch=:opt&flag=flag&status=status # POST /source//?cmd=set_flag&repository=:opt&arch=:opt&flag=flag&status=status -def test_set_package_flags(requests_mock): - pass +@pytest.mark.parametrize('target', ('project', 'package')) +@pytest.mark.parametrize('flag', ('build', 'publish', 'useforbuild', 'debuginfo')) +@pytest.mark.parametrize('status', ('enable', 'disable')) +@pytest.mark.parametrize('repository', (None, 'SLE_12_SP5'), ids=('no-repo', 'repo')) +@pytest.mark.parametrize('arch', (None, 'x86_64'), ids=('no-arch', 'arch')) +def test_set_flag(requests_mock, target, flag, status, repository, arch): + if target == 'project': + matcher = re.compile(r'/source/[^/]+\?') + else: + matcher = re.compile(r'/source/.+/[^/]*\?') + requests_mock.post(matcher) + params = dict(cmd='set_flag', + flag=flag, + status=status, + repository=repository, + arch=arch) + + if target == 'project': + response = api.set_flag(prj, flag=params) + else: + response = api.set_flag(prj, pkg, flag=params) + assert response.status_code == 200 + + url = requests_mock.request_history[0].url + check_url_query_params(url, params) + -# POST /source//?cmd=remove_flag&repository=:opt&arch=:opt&flag=flag # POST /source/?cmd=remove_flag&repository=:opt&arch=:opt&flag=flag +# POST /source//?cmd=remove_flag&repository=:opt&arch=:opt&flag=flag +@pytest.mark.parametrize('target', ('project', 'package')) +@pytest.mark.parametrize('flag', ('build', 'publish', 'useforbuild', 'debuginfo')) +@pytest.mark.parametrize('status', ('enable', 'disable')) +@pytest.mark.parametrize('repository', (None, 'SLE_12_SP5'), ids=('no-repo', 'repo')) +@pytest.mark.parametrize('arch', (None, 'x86_64'), ids=('no-arch', 'arch')) +def test_remove_flag(requests_mock, target, flag, status, repository, arch): + if target == 'project': + matcher = re.compile(r'/source/[^/]+\?') + else: + matcher = re.compile(r'/source/.+/[^/]*\?') + requests_mock.post(matcher) + params = dict(cmd='remove_flag', + flag=flag, + status=status, + repository=repository, + arch=arch) + + if target == 'project': + response = api.remove_flag(prj, flag=params) + else: + response = api.remove_flag(prj, pkg, flag=params) + assert response.status_code == 200 + + url = requests_mock.request_history[0].url + check_url_query_params(url, params) -- cgit v1.2.3