PR#1149: hub: new addArchiveType RPC

Merges #1149
https://pagure.io/koji/pull-request/1149

Fixes #1528
hub: new addArchiveType RPC
https://pagure.io/koji/issue/1528

hub/kojihub.py

@@ -6737,6 +6737,37 @@ def get_archive_type(filename=None, type_name=None, type_id=None, strict=False):
     else:
         return None
 
+def add_archive_type(name, description, extensions):
+    """
+    Add new archive type.
+
+    Use this to tell Koji about new builds' files' extensions before
+    importing the files.
+
+    :param str name: archive type name, eg. "yaml"
+    :param str description: eg. "YAML Ain't Markup Language"
+    :param str extensions: space-separated list of descriptions, eg. "yaml yml"
+    """
+    context.session.assertPerm('admin')
+    data = {'name': name,
+            'description': description,
+            'extensions': extensions,
+    }
+    if get_archive_type(type_name=name):
+        raise koji.GenericError("archivetype %s already exists" % name)
+    # No invalid or duplicate extensions
+    for ext in extensions.split(' '):
+        if not ext.replace('.', '').isalnum():
+            raise koji.GenericError('invalid %s file extension' % ext)
+        select = r"""SELECT id FROM archivetypes
+                      WHERE extensions ~* E'(\\s|^)%s(\\s|$)'""" % ext
+        results = _multiRow(select, {}, ('id',))
+        if len(results) > 0:
+            raise koji.GenericError('file extension %s already exists' % ext)
+    insert = InsertProcessor('archivetypes', data=data)
+    insert.execute()
+
+
 def new_maven_build(build, maven_info):
     """
     Add Maven metadata to an existing build.
@@ -10339,6 +10370,8 @@ class RootExports(object):
     listBTypes = staticmethod(list_btypes)
     addBType = staticmethod(add_btype)
 
+    addArchiveType = staticmethod(add_archive_type)
+
     def getChangelogEntries(self, buildID=None, taskID=None, filepath=None, author=None, before=None, after=None, queryOpts=None):
         """Get changelog entries for the build with the given ID,
            or for the rpm generated by the given task at the given path

tests/test_hub/test_add_archivetype.py

@@ -0,0 +1,53 @@
+from __future__ import absolute_import
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+import mock
+
+import koji
+import kojihub
+
+IP = kojihub.InsertProcessor
+
+
+class TestAddArchiveType(unittest.TestCase):
+
+    @mock.patch('kojihub._multiRow')
+    @mock.patch('kojihub.get_archive_type')
+    @mock.patch('kojihub.InsertProcessor')
+    def test_add_archive_type(self, InsertProcessor, get_archive_type,
+                              _multiRow):
+        # Not sure why mock can't patch kojihub.context, so we do this
+        session = kojihub.context.session = mock.MagicMock()
+        mocks = [InsertProcessor, get_archive_type, session]
+        # It seems MagicMock will not automatically handle attributes that
+        # start with "assert"
+        session.assertPerm = mock.MagicMock()
+
+        # expected case
+        get_archive_type.return_value = None
+        insert = InsertProcessor.return_value
+        kojihub.add_archive_type('deb', 'Debian package', 'deb')
+        InsertProcessor.assert_called_once()
+        insert.execute.assert_called_once()
+
+        args, kwargs = InsertProcessor.call_args
+        ip = IP(*args, **kwargs)
+        self.assertEquals(ip.table, 'archivetypes')
+        self.assertEquals(ip.data, {'name': 'deb',
+                                    'description': 'Debian package',
+                                    'extensions': 'deb'})
+        self.assertEquals(ip.rawdata, {})
+        session.assertPerm.assert_called_with('admin')
+
+        for m in mocks:
+            m.reset_mock()
+        session.assertPerm = mock.MagicMock()
+
+        # already exists
+        get_archive_type.return_value = True
+        with self.assertRaises(koji.GenericError):
+            kojihub.add_archive_type('deb', 'Debian package', 'deb')
+        InsertProcessor.assert_not_called()
+        session.assertPerm.assert_called_with('admin')