From 89804ba79500b7f3d7e8970da47e7b6b8c5db7de Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Jan 21 2026 15:10:19 +0000 Subject: fix duplicate check in add_archive_type Fixes https://pagure.io/koji/issue/4527 --- diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 9e9fb47..97ea5c9 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -7953,11 +7953,13 @@ def add_archive_type(name, description, extensions, compression_type=None): if get_archive_type(type_name=name): raise koji.GenericError(f"archivetype {name} already exists") # No invalid or duplicate extensions - for ext in extensions.split(' '): + for ext in extensions.lower().split(' '): if not ext.replace('.', '').isalnum(): raise koji.GenericError(f'No such {ext} file extension') - query = QueryProcessor(tables=['archivetypes'], columns=['id'], - clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], values={}) + query = QueryProcessor( + tables=['archivetypes'], columns=['id'], values={'ext': ext}, + clauses=[r"%(ext)s IN (SELECT lower(s)" + r" FROM unnest(regexp_split_to_array(extensions, '\s+')) AS s)"]) results = query.execute() if len(results) > 0: raise koji.GenericError(f'file extension {ext} already exists') diff --git a/tests/test_hub/test_add_archivetype.py b/tests/test_hub/test_add_archivetype.py index 7cc1d18..31a5ab8 100644 --- a/tests/test_hub/test_add_archivetype.py +++ b/tests/test_hub/test_add_archivetype.py @@ -40,10 +40,8 @@ class TestAddArchiveType(DBQueryTestCase): rv = kojihub.add_archive_type('deb', 'Debian package', ext) self.assertEqual(len(self.queries), 1) - self.assertLastQueryEqual(tables=['archivetypes'], - columns=['id'], - clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], - values={}) + query = self.queries[-1] + self.assertEqual(query.tables, ['archivetypes']) self.assertEqual(rv, None) self.assertEqual(len(self.inserts), 1) insert = self.inserts[0] @@ -64,10 +62,8 @@ class TestAddArchiveType(DBQueryTestCase): kojihub.add_archive_type('jar', 'Jar package', ext, compression_type='zip') self.assertEqual(len(self.queries), 1) - self.assertLastQueryEqual(tables=['archivetypes'], - columns=['id'], - clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], - values={}) + query = self.queries[-1] + self.assertEqual(query.tables, ['archivetypes']) self.assertEqual(len(self.inserts), 1) insert = self.inserts[0] self.assertEqual(insert.table, 'archivetypes') @@ -122,10 +118,8 @@ class TestAddArchiveType(DBQueryTestCase): self.assertEqual(f'file extension {ext} already exists', str(ex.exception)) self.assertEqual(len(self.queries), 1) - self.assertLastQueryEqual(tables=['archivetypes'], - columns=['id'], - clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], - values={}) + query = self.queries[-1] + self.assertEqual(query.tables, ['archivetypes']) self.assertEqual(len(self.inserts), 0) self.context.session.assertPerm.assert_called_with('admin') self.get_archive_type.assert_called_once_with(type_name='jar')