adding tests

plugins/quetz_repodata_patching/quetz_repodata_patching/main.py

@@ -88,7 +88,10 @@ def extract_from_conda(fs):
 
 
 def _load_instructions(tar, path):
-    patch_instructions = json.load(tar.extractfile(path))
+    try:
+        patch_instructions = json.load(tar.extractfile(path))
+    except KeyError:
+        return {}
     return patch_instructions
 
 
@@ -164,12 +167,16 @@ def _addfile(content, path):
 
             subdir_path = f"{subdir}/{path}"
 
-            pkgstore.add_file(
-                content,
-                channel_name,
-                subdir_path,
-            )
+            if isinstance(content, str):
+                content = content.encode("utf-8")
+
+            compressed_content = bz2.compress(content)
+
+            pkgstore.add_file(content, channel_name, subdir_path)
+            pkgstore.add_file(compressed_content, channel_name, subdir_path + ".bz2")
+
             updated_files.append((path, content))
+            updated_files.append((path + ".bz2", compressed_content))
 
         with extract_(fs) as tar:
 
@@ -186,23 +193,15 @@ def _addfile(content, path):
                     repodata_str = fs.read()
                     repodata = json.loads(repodata_str)
 
-                    if fname == 'repodata':
-                        packages.update(repodata["packages"])
-                        packages.update(repodata["packages.conda"])
-
                     _addfile(repodata_str, f"{fname}_from_packages.json")
 
-                    compressed_repodata_str = bz2.compress(repodata_str)
-                    _addfile(compressed_repodata_str, f"{fname}_from_packages.json.bz2")
-
                     patch_repodata(repodata, patch_instructions)
 
+                    if fname == 'repodata':
+                        packages.update(repodata["packages"])
+                        packages.update(repodata["packages.conda"])
+
                     patched_repodata_str = json.dumps(repodata)
                     _addfile(patched_repodata_str, f"{fname}.json")
 
-                    compressed_patched_repodata_str = bz2.compress(
-                        patched_repodata_str.encode('utf-8')
-                    )
-                    _addfile(compressed_patched_repodata_str, f"{fname}.json.bz2")
-
                 update_index(pkgstore, updated_files, channel_name, subdir, packages)

plugins/quetz_repodata_patching/tests/test_main.py

@@ -56,6 +56,11 @@ def channel(dao: "quetz.dao.Dao", channel_name, user):
     return channel
 
 
+@pytest.fixture
+def package_subdir():
+    return "noarch"
+
+
 @pytest.fixture
 def package_version(
     dao: "quetz.dao.Dao",
@@ -65,18 +70,23 @@ def package_version(
     db,
     package_file_name,
     package_format,
+    package_subdir,
 ):
+    channel_data = json.dumps({"subdirs": [package_subdir]})
     package_data = Package(name=package_name)
 
-    dao.create_package(channel.name, package_data, user.id, "owner")
+    package = dao.create_package(channel.name, package_data, user.id, "owner")
+    package.channeldata = channel_data
+    db.commit()
+
     package_info = (
         '{"run_exports": {"weak": ["otherpackage > 0.1"]}, "size": 100, "depends": []}'
     )
     version = dao.create_version(
         channel.name,
         package_name,
         package_format,
-        "noarch",
+        package_subdir,
         "0.1",
         "0",
         "0",
@@ -114,6 +124,7 @@ def remove_instructions():
 
 @pytest.fixture
 def patched_package_name(package_file_name):
+    "name of the package in patch_instructions"
     # by default the name of the package in patch_instructions is the same
     # as the name of the dummy package
     # but we will change it in the test to test if we can patch .conda files
@@ -146,8 +157,13 @@ def archive_format():
     return "tarbz2"
 
 
+@pytest.fixture()
+def patches_subdir():
+    return "noarch"
+
+
 @pytest.fixture
-def repodata_archive(repodata_file_name, patch_content, archive_format):
+def repodata_archive(repodata_file_name, patch_content, archive_format, patches_subdir):
 
     from io import BytesIO
 
@@ -158,13 +174,13 @@ def mk_tarfile(patch_instructions, compr=None):
         mode = f'w|{compr}' if compr else 'w'
         tar_content = BytesIO()
         tar = tarfile.open(repodata_file_name, mode, fileobj=tar_content)
-        info = tarfile.TarInfo(name='noarch')
+        info = tarfile.TarInfo(name=patches_subdir)
         info.type = tarfile.DIRTYPE
         info.mode = 0o755
         info.mtime = time.time()
         tar.addfile(tarinfo=info)
 
-        info = tarfile.TarInfo(name='noarch/patch_instructions.json')
+        info = tarfile.TarInfo(name=f'{patches_subdir}/patch_instructions.json')
         info.size = len(patch_instructions.getvalue())
         info.mtime = time.time()
         tar.addfile(tarinfo=info, fileobj=patch_instructions)
@@ -340,14 +356,22 @@ def get_db():
     assert not data.get("removed")
 
 
+@pytest.mark.parametrize(
+    "remove_instructions",
+    [
+        [],
+        ["mytestpackage-0.1-0.tar.bz2"],
+    ],
+)
 def test_index_html(
     pkgstore,
     package_version,
     package_repodata_patches,
     channel_name,
-    patched_package_name,
+    package_file_name,
     dao,
     db,
+    remove_instructions,
 ):
     def get_db():
         yield db
@@ -370,3 +394,116 @@ def get_db():
     assert "repodata.json.bz2" in content
     assert "repodata_from_packages.json" in content
     assert "repodata_from_packages.json.bz2" in content
+    if remove_instructions:
+        assert package_file_name not in content
+    else:
+        assert package_file_name in content
+
+
+@pytest.mark.parametrize("package_subdir", ["linux-64", "noarch"])
+@pytest.mark.parametrize("patches_subdir", ["linux-64", "noarch"])
+def test_patches_for_subdir(
+    pkgstore,
+    package_version,
+    channel_name,
+    package_file_name,
+    package_repodata_patches,
+    dao,
+    db,
+    package_subdir,
+    patches_subdir,
+):
+    def get_db():
+        yield db
+
+    with mock.patch("quetz_repodata_patching.main.get_db", get_db):
+        indexing.update_indexes(dao, pkgstore, channel_name)
+
+    index_path = os.path.join(
+        pkgstore.channels_dir,
+        channel_name,
+        package_subdir,
+        "index.html",
+    )
+
+    assert os.path.isfile(index_path)
+    with open(index_path, 'r') as fid:
+        content = fid.read()
+
+    assert "repodata.json" in content
+    assert "repodata.json.bz2" in content
+    assert "repodata_from_packages.json" in content
+    assert "repodata_from_packages.json.bz2" in content
+
+    for fname in ("repodata.json", "current_repodata.json"):
+
+        repodata_path = os.path.join(
+            pkgstore.channels_dir, channel_name, package_subdir, fname
+        )
+
+        assert os.path.isfile(repodata_path)
+
+        with open(repodata_path) as fid:
+            data = json.load(fid)
+
+        packages = data["packages"]
+
+        pkg = packages[package_file_name]
+
+        if patches_subdir == package_subdir:
+            assert pkg['run_exports'] == {"weak": ["otherpackage > 0.2"]}
+        else:
+            assert pkg['run_exports'] == {"weak": ["otherpackage > 0.1"]}
+
+
+def test_no_repodata_patches_package(
+    pkgstore,
+    package_version,
+    channel_name,
+    package_file_name,
+    dao,
+    db,
+):
+    def get_db():
+        yield db
+
+    with mock.patch("quetz_repodata_patching.main.get_db", get_db):
+        indexing.update_indexes(dao, pkgstore, channel_name)
+
+    index_path = os.path.join(
+        pkgstore.channels_dir,
+        channel_name,
+        "noarch",
+        "index.html",
+    )
+
+    assert os.path.isfile(index_path)
+    with open(index_path, 'r') as fid:
+        content = fid.read()
+
+    assert "repodata.json" in content
+    assert "repodata.json.bz2" in content
+    assert "repodata_from_packages.json" not in content
+    assert "repodata_from_packages.json.bz2" not in content
+
+    for fname in ("repodata.json", "current_repodata.json"):
+
+        repodata_path = os.path.join(
+            pkgstore.channels_dir, channel_name, "noarch", fname
+        )
+
+        assert os.path.isfile(repodata_path)
+
+        with open(repodata_path) as fid:
+            data = json.load(fid)
+
+        packages = data["packages"]
+
+        pkg = packages[package_file_name]
+
+        assert pkg['run_exports'] == {"weak": ["otherpackage > 0.1"]}
+
+        assert not pkg.get("revoked", False)
+        assert 'package_has_been_revoked' not in pkg["depends"]
+
+        assert package_file_name not in data.get("removed", ())