Feature update

HISTORY.rst

@@ -1,6 +1,15 @@
 =======
 History
 =======
+0.0.9 (2023-08-21)
+------------------
+* Support DLRN results, local composes
+* Allow et:: to look at one specific advisory instead of a release
+* Allow skipping downgrades or upgrades
+* Improve help output
+* Allow specifying verify parameter to requests
+* Allow multi-arch non-RPM builds when cleaning tags
+
 0.0.8 (2020-09-10)
 ------------------
 * Use argparse for tag-over

PKG-INFO

@@ -1,6 +1,6 @@
 Metadata-Version: 1.1
 Name: tag_utils
-Version: 0.0.7
+Version: 0.0.9
 Summary: Pungi/Koji tag utilities for Release Depot
 Author: release-depot
 Author-email: [email protected]

setup.py

@@ -14,7 +14,7 @@
 
 setup(
     name='tag_utils',
-    version='0.0.8',
+    version='0.0.9',
     install_requires=['pyyaml', 'koji', 'toolchest>=0.0.6', 'koji_wrapper'],
     tests_require=TEST_REQUIRES,
     extras_require={'test': TEST_REQUIRES,

tag_utils/cli/tag_delta.py

@@ -19,6 +19,7 @@ def main():
                '<koji_tag> | file:<filename> '
                '|  http[s]://<url to compose> '
                '| et:<release name or id>  (Errata Tool Release name or id) '
+               '| et::<id>                 (Single Errata Tool advisory id) '
                '| -   (use stdin as list of nvr to compare against)')
 
     parser.add_argument('--new', action='store_true', default=False,

tag_utils/compose.py

@@ -19,6 +19,20 @@ def check_compose(url, verify=True):
     raise ValueError('Failed to check compose; HTTP code ' + str(ret.status_code))
 
 
+def fetch_or_read(url):
+    if url.startswith('file://'):
+        filename = url[7:]
+        fp = open(filename, 'r')
+        ret = fp.read()
+        fp.close()
+        return ret
+    else:
+        ret = requests.get(url)
+        if ret.status_code == 200:
+            return ret.text
+        raise ValueError('Failed to check compose; HTTP code ' + str(ret.status_code))
+
+
 def fetch_rpm_metadata(url, verify=True):
     check_compose(url)
     full_url = os.path.join(url, 'compose/metadata/rpms.json')

tag_utils/delta.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python
 
+from urllib.parse import urlparse
+
 import koji_wrapper
 from koji_wrapper.base import KojiWrapperBase
 from koji_wrapper.tag import KojiTag
@@ -9,7 +11,8 @@
 
 from .basic import file_as_nevr
 from .compose import compose_as_nevr
-from .et import get_build_for_release
+from .dlrn import json_api_as_nevr
+from .et import get_build_for_release, get_build_for_advisory
 from .koji import latest_tagged_as_nevr
 
 __koji_session = None
@@ -29,6 +32,20 @@ def release_set_as_nevr(release_name_or_id, koji_session, **kwargs):
     return get_build_for_release(release_name_or_id.strip('et:'), __koji_session)
 
 
+def erratum_set_as_nevr(errata_id, koji_session, **kwargs):
+    global __koji_session
+
+    if 'session' in kwargs:
+        __koji_session = kwargs['session']
+
+    if __koji_session is None:
+        __koji_session = KojiWrapperBase(profile='brew')
+    if __koji_session is None:
+        raise Exception('Could not connect to koji')
+
+    return get_build_for_advisory(errata_id.strip('et::'), __koji_session)
+
+
 def tag_to_latest_builds(tag, **kwargs):
     global __koji_session
     inherit = False
@@ -55,20 +72,39 @@ def tag_to_latest_builds(tag, **kwargs):
     return latest_tagged_as_nevr(koji_tag)
 
 
+def _url_as_nevr(inp):
+    parsed_url = urlparse(inp)
+
+    if not parsed_url.path:
+        raise ValueError('Invalid URL: ' + inp)
+
+    if parsed_url.path.startswith('/dlrn/versions/'):
+        # drln API (JSON) - preferred
+        ret = json_api_as_nevr(inp)
+    else:
+        # fetch compose data
+        ret = compose_as_nevr(inp)
+    return ret
+
+
 # Automatically resolve input class to dict of:
 #   {'n': 'n-e:v-r', 'm': 'm-e:v-r'}
 # Supports:
 #    - pungi compose
 #    - koji tag
+#    - flat file (list of NVRs
+#    - errata tool build lists
+#    - JSON-parsed DLRN results
 def input_to_nevr_dict(inp, **kwargs):
     ret = None
 
     if isinstance(inp, str):
         if ( inp.startswith('http://') or   # NOQA
              inp.startswith('https://') or  # NOQA
-             inp.startswith('/')):
-            # fetch compose data
-            ret = compose_as_nevr(inp)
+             inp.startswith('file://')):
+            ret = _url_as_nevr(inp)
+        elif inp.startswith('et::'):
+            ret = erratum_set_as_nevr(inp, __koji_session, **kwargs)
         elif inp.startswith('et:'):
             ret = release_set_as_nevr(inp, __koji_session, **kwargs)
         elif inp.startswith('file:'):

tag_utils/dlrn.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+
+import json
+
+import requests
+
+
+from .common import tidy_nevra
+
+
+def fetch_json_data(url):
+    ret = requests.get(url)
+    if ret.status_code == 200:
+        return json.loads(ret.text)
+
+
+def json_api_as_nevr(url):
+    md = fetch_json_data(url)
+    ret = {}
+    for p in md['packages']:
+        ret[p['name']] = tidy_nevra(p['nvr'])
+    return ret

tag_utils/et.py

@@ -33,18 +33,7 @@ def get_advisory_list(errata_tool_release, debug=False):
     return [Erratum(errata_id=advisory_id) for advisory_id in advisory_ids]
 
 
-def get_build_for_release(release_name_or_id, koji_session, debug=False):
-    if debug:
-        print('got in "{0}"'.format(release_name_or_id))
-    try:
-        rel = errata_tool.release.Release(id=int(release_name_or_id))
-        advisories = [i for i in rel.advisories() if i["status"] not in ('SHIPPED_LIVE', 'DROP_NO_SHIP')]
-        advisory_list = [Erratum(errata_id=i['id']) for i in advisories]
-        if debug:
-            print("Got list of advisories:", advisories)
-    except ValueError:
-        advisory_list = get_advisory_list(release_name_or_id, debug=debug)
-
+def _get_builds_for_advisory_list(advisory_list, koji_session, debug=False):
     all_builds = set()
     for advisory in advisory_list:
         if debug:
@@ -61,3 +50,25 @@ def get_build_for_release(release_name_or_id, koji_session, debug=False):
         return_data[build_data['name']] = koji_build_to_nevr(build_data)
 
     return return_data
+
+
+def get_build_for_release(release_name_or_id, koji_session, debug=False):
+    if debug:
+        print('got in "{0}"'.format(release_name_or_id))
+    try:
+        rel = errata_tool.release.Release(id=int(release_name_or_id))
+        advisories = [i for i in rel.advisories() if i["status"] not in ('SHIPPED_LIVE', 'DROP_NO_SHIP')]
+        advisory_list = [Erratum(errata_id=i['id']) for i in advisories]
+        if debug:
+            print("Got list of advisories:", advisories)
+    except ValueError:
+        advisory_list = get_advisory_list(release_name_or_id, debug=debug)
+
+    return _get_builds_for_advisory_list(advisory_list, koji_session, debug=debug)
+
+
+def get_build_for_advisory(advisory_id, koji_session, debug=False):
+    if debug:
+        print('got advisory "{0}"'.format(advisory_id))
+    erratum = Erratum(errata_id=advisory_id)
+    return _get_builds_for_advisory_list([erratum], koji_session, debug=debug)