Extend test to cover double upload bug

quetz/tests/api/test_main_packages.py

@@ -1,9 +1,10 @@
 import hashlib
 import json
 import os
+import shutil
 import time
 from pathlib import Path
-from typing import BinaryIO, Callable, Union
+from typing import BinaryIO, Callable, Tuple, Union
 
 import pytest
 
@@ -333,12 +334,12 @@ def _upload_file_1(
  auth_client,
  public_channel,
  public_package,
- package_filename: str,
+ filepath: Path,
  force: bool = False,
 ):
  """Upload a file using /channels/{channel_name}/packages/{package_name}/files/"""
- with open(package_filename, "rb") as fid:
- files = {"files": (package_filename, fid)}
+ with open(filepath, "rb") as fid:
+ files = {"files": (filepath.name, fid)}
  response = auth_client.post(
  f"/api/channels/{public_channel.name}/packages/"
  f"{public_package.name}/files/",
@@ -352,15 +353,15 @@ def _upload_file_2(
  auth_client,
  public_channel,
  public_package,
- package_filename: str,
+ filepath: Path,
  force: bool = False,
 ):
  """Upload a file using /channels/{channel_name}/upload/{file_name}"""
 
- with open(package_filename, "rb") as fid:
+ with open(filepath, "rb") as fid:
  body_bytes = fid.read()
  response = auth_client.post(
- f"/api/channels/{public_channel.name}/upload/{package_filename}",
+ f"/api/channels/{public_channel.name}/upload/{filepath.name}",
  content=body_bytes,
  params={"force": force, "sha256": hashlib.sha256(body_bytes).hexdigest()},
  )
@@ -384,7 +385,7 @@ def test_upload_package_version_wrong_filename(
  os.rename("test-package-0.1-0.tar.bz2", package_filename)
 
  response = upload_function(
- auth_client, public_channel, public_package, package_filename
+ auth_client, public_channel, public_package, Path(package_filename)
  )
 
  package_dir = Path(pkgstore.channels_dir) / public_channel.name / 'linux-64'
@@ -397,12 +398,14 @@ def test_upload_package_version_wrong_filename(
  assert not os.path.exists(package_dir)
 
 
-def sha256(path: Union[Path, str]) -> str:
+def sha_and_md5(path: Union[Path, str]) -> Tuple[str, str]:
  sha = hashlib.sha256()
+ md5 = hashlib.md5()
  with open(path, "rb") as f:
  for chunk in iter(lambda: f.read(2**16), b""):
  sha.update(chunk)
- return sha.hexdigest()
+ md5.update(chunk)
+ return sha.hexdigest(), md5.hexdigest()
 
 
 @pytest.mark.parametrize("upload_function", [_upload_file_1, _upload_file_2])
@@ -416,78 +419,125 @@ def test_upload_duplicate_package_version(
  config,
  remove_package_versions,
  upload_function: Callable,
+ tmp_path,
 ):
  pkgstore = config.get_package_store()
 
- package_filename = "test-package-0.1-0.tar.bz2"
- package_filename_copy = "test-package-0.1-0_copy.tar.bz2"
+ def get_repodata():
+ """Helper function to read repo data"""
+ package_dir = Path(pkgstore.channels_dir) / public_channel.name / 'linux-64'
+ return json.loads((package_dir / 'repodata.json').read_text())
+
+ # Test setup: path1 is a package we will upload
+ path1 = Path(__file__).parent.parent / "data" / "test-package-0.1-0.tar.bz2"
 
- file_sha = sha256(package_filename)
- file_copy_sha = sha256(package_filename_copy)
+ # To test duplicate uploads, we have a second file `_copy`, which is the same
+ # package and version but has a different content, so different hashes
+ # We must move this file to a temporary directory so that we can give it the same
+ # name as the first file.
+ path2 = tmp_path / path1.name
+ shutil.copyfile(
+ Path(__file__).parent.parent / "data" / "test-package-0.1-0_copy.tar.bz2", path2
+ )
+
+ # Sanity checks
+ sha1, md51 = sha_and_md5(path1)
+ sha2, md52 = sha_and_md5(path2)
+ assert (sha1 != sha2) and (
+ md51 != md52
+ ), "Sanity check failure: Test files have same hash"
  assert (
- file_sha != file_copy_sha
- ), "Sanity check: Test files must have different hashes for this test."
- file_size = os.path.getsize(package_filename)
- file_copy_size = os.path.getsize(package_filename_copy)
+ path1.name == path2.name
+ ), "Sanity check failure: Test files have different name"
+
+ size1 = os.path.getsize(path1)
+ size2 = os.path.getsize(path2)
  assert (
- file_size != file_copy_size
- ), "Sanity check: Test files must have different sizes for this test."
+ size1 != size2
+ ), "Sanity check failure: Test files must have different sizes for this test."
 
- upload_function(auth_client, public_channel, public_package, package_filename)
+ # First upload
+ # File should not exist
+ assert not pkgstore.file_exists(public_channel.name, f"linux-64/{path1.name}")
+ response = upload_function(auth_client, public_channel, public_package, path1)
 
- def get_repodata():
- """Helper function to read repo data"""
- package_dir = Path(pkgstore.channels_dir) / public_channel.name / 'linux-64'
- return json.loads((package_dir / 'repodata.json').read_text())
+ # Expect success
+ assert response.status_code == 201
 
- repodata_init = get_repodata()
- assert repodata_init["packages"][package_filename]["sha256"] == file_sha
+ # Check meta data is OK
+ repodata_after_first = get_repodata()
+ assert repodata_after_first["packages"][path1.name]["sha256"] == sha1
+ assert repodata_after_first["packages"][path1.name]["md5"] == md51
+ assert repodata_after_first["packages"][path1.name]["size"] == size1
 
- # Try submitting the same package without 'force' flag
- response = upload_function(
- auth_client, public_channel, public_package, package_filename
+ # Check that the file in the store is OK
+ file_in_store = (
+ Path(pkgstore.channels_dir) / public_channel.name / 'linux-64' / path1.name
  )
+ assert size1 == os.path.getsize(file_in_store)
+ assert (sha1, md51) == sha_and_md5(file_in_store)
 
- # Expect 409 here
+ # Second upload: File with same name but different content, without force
+ response = upload_function(auth_client, public_channel, public_package, path2)
+
+ # Expect 409 since the file already exists
  assert response.status_code == 409
  detail = response.json()['detail']
  assert "Duplicate" in detail
 
- # Change the archive to test force update
- os.remove(package_filename)
- os.rename(package_filename_copy, package_filename)
+ # Check meta data is OK: It should not have changed with respect to before
+ repodata_after_second = get_repodata()
+ assert repodata_after_second == repodata_after_first
+
+ # Check that the file in the store is OK
+ file_in_store = (
+ Path(pkgstore.channels_dir) / public_channel.name / 'linux-64' / path1.name
+ )
 
+ # File in store should not be the second file
+ assert not (
+ (sha2, md52) == sha_and_md5(file_in_store)
+ ), "Duplicate upload without force updated stored file."
+ assert not (
+ size2 == os.path.getsize(file_in_store)
+ ), "Duplicate upload without force updated stored file."
+
+ # File in store should be the first file
+ assert size1 == os.path.getsize(
+ file_in_store
+ ), "Duplicate upload without force updated stored file."
+ assert (sha1, md51) == sha_and_md5(
+ file_in_store
+ ), "Duplicate upload without force updated stored file."
+
+ # Third upload: Same as second but now with force
  # Ensure the 'time_modified' value changes in repodata.json
  time.sleep(1)
 
  # Submit the same package with 'force' flag
  response = upload_function(
- auth_client, public_channel, public_package, package_filename, force=True
+ auth_client, public_channel, public_package, path2, force=True
  )
  assert response.status_code == 201
 
  # Check that repodata content has been updated
- repodata = get_repodata()
+ repodata_after_force = get_repodata()
 
  # Info should match
- assert repodata_init["info"] == repodata["info"]
+ assert repodata_after_first["info"] == repodata_after_force["info"]
 
  # Package keys should match
- assert repodata_init["packages"].keys() == repodata["packages"].keys()
- repodata_init_pkg = repodata_init["packages"][package_filename]
- repodata_pkg = repodata["packages"][package_filename]
-
- # Hashes should match pre-upload expectation
- assert repodata_init_pkg["sha256"] == file_sha
- assert repodata_pkg["sha256"] == file_copy_sha
+ assert (
+ repodata_after_first["packages"].keys()
+ == repodata_after_force["packages"].keys()
+ )
 
- # Sizes should match pre-upload expectation
- assert repodata_init_pkg["size"] == file_size
- assert repodata_pkg["size"] == file_copy_size
+ # Hashes should now have changed to the second file
+ assert repodata_after_force["packages"][path1.name]["sha256"] == sha2
+ assert repodata_after_force["packages"][path1.name]["md5"] == md52
+ assert repodata_after_force["packages"][path1.name]["size"] == size2
 
- # Upload-related metadata should not match
- for key in "time_modified", "md5", "sha256":
- assert repodata_init_pkg[key] != repodata_pkg[key]
+ assert repodata_after_force["time_modified"] > repodata_after_first["time_modified"]
 
 
 @pytest.mark.parametrize("package_name", ["test-package"])