Tests: Use original CName and use dataclass instead of tuple for fixture

Author: ByteOtter

tests/s3/conftest.py

@@ -1,17 +1,39 @@
 import pytest
-from pathlib import Path
+from dataclasses import dataclass
 from moto import mock_aws
 from hashlib import md5, sha256
 import boto3
 
+from gardenlinux.features.cname import CName as RealCName
+
+BUCKET_NAME = "test-bucket"
+REGION = "us-east-1"
+
 
 # Dummy CName replacement
-class DummyCName:
-    def __init__(self, cname):  # pylint: disable=unused-argument
-        self.platform = "aws"
-        self.arch = "amd64"
-        self.version = "1234.1"
-        self.commit_id = "abc123"
+class DummyCName(RealCName):
+    def __init__(self, cname: str):  # pylint: disable=unused-argument
+        super().__init__(cname)
+
+
+@dataclass(frozen=True)
+class S3Env:
+    s3: object
+    bucket_name: str
+    tmp_path: str
+    cname: str
+
+
+def make_cname(
+    flavor: str = "testcname",
+    arch: str = "amd64",
+    version: str = "1234.1",
+    commit: str = "abc123",
+) -> str:
+    """
+    Helper function to build cname. Can be used to customized the cname.
+    """
+    return f"{flavor}-{arch}-{version}-{commit}"
 
 
 # Helpers to compute digests for fake files
@@ -36,11 +58,11 @@ def s3_setup(tmp_path, monkeypatch):
     Provides a clean S3 setup for each test.
     """
     with mock_aws():
-        s3 = boto3.resource("s3", region_name="us-east-1")
-        bucket_name = "test-bucket"
-        s3.create_bucket(Bucket=bucket_name)
+        s3 = boto3.resource("s3", region_name=REGION)
+        s3.create_bucket(Bucket=BUCKET_NAME)
 
         monkeypatch.setattr("gardenlinux.s3.s3_artifacts.CName", DummyCName)
         monkeypatch.setattr("gardenlinux.s3.s3_artifacts.file_digest", dummy_digest)
 
-        yield s3, bucket_name, tmp_path
+        cname = make_cname()
+        yield S3Env(s3, BUCKET_NAME, tmp_path, cname)

tests/s3/test_bucket.py

@@ -20,10 +20,10 @@ def test_objects_empty(s3_setup):
     List objects from empty bucket.
     """
     # Arrange
-    s3, bucket_name, _ = s3_setup
+    env = s3_setup
 
     # Act
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
 
     # Assert
     assert list(bucket.objects) == []
@@ -35,13 +35,13 @@ def test_upload_file_and_list(s3_setup):
     to list it
     """
     # Arrange
-    s3, bucket_name, tmp_path = s3_setup
+    env = s3_setup
 
-    test_file = tmp_path / "example.txt"
+    test_file = env.tmp_path / "example.txt"
     test_file.write_text("hello moto")
 
     # Act
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
     bucket.upload_file(str(test_file), "example.txt")
 
     all_keys = [obj.key for obj in bucket.objects]
@@ -55,12 +55,12 @@ def test_download_file(s3_setup):
     Try to download a file pre-existing in the bucket
     """
     # Arrange
-    s3, bucket_name, tmp_path = s3_setup
-    s3.Object(bucket_name, "file.txt").put(Body=b"some data")
+    env = s3_setup
+    env.s3.Object(env.bucket_name, "file.txt").put(Body=b"some data")
 
     # Act
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
-    target_path = tmp_path / "downloaded.txt"
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
+    target_path = env.tmp_path / "downloaded.txt"
     bucket.download_file("file.txt", str(target_path))
 
     # Assert
@@ -72,15 +72,15 @@ def test_upload_fileobj(s3_setup):
     Upload a file-like in-memory object to the bucket
     """
     # Arrange
-    s3, bucket_name, _ = s3_setup
+    env = s3_setup
 
     # Act
     # Create in-memory binary stream (file content)
     data = io.BytesIO(b"Test Data")
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
     bucket.upload_fileobj(data, "binary.obj")
 
-    obj = s3.Object(bucket_name, "binary.obj").get()
+    obj = env.s3.Object(env.bucket_name, "binary.obj").get()
 
     # Assert
     assert obj["Body"].read() == b"Test Data"
@@ -91,12 +91,12 @@ def test_download_fileobj(s3_setup):
     Download data into a in-memory object
     """
     # Arange
-    s3, bucket_name, _ = s3_setup
+    env = s3_setup
     # Put some object in the bucket
-    s3.Object(bucket_name, "somekey").put(Body=b"123abc")
+    env.s3.Object(env.bucket_name, "somekey").put(Body=b"123abc")
 
     # Act
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
     # Create empty in-memory bytestream to act as a writable file
     output = io.BytesIO()
     bucket.download_fileobj("somekey", output)
@@ -115,11 +115,11 @@ def test_getattr_delegates(s3_setup):
     the call to the real boto3 bucket.
     """
     # Arrange
-    _, bucket_name, _ = s3_setup
+    env = s3_setup
 
     # Act
-    bucket = Bucket(bucket_name, s3_resource_config={"region_name": REGION})
+    bucket = Bucket(env.bucket_name, s3_resource_config={"region_name": REGION})
 
     # Assert
     # __getattr__ should delegate this to the underlying boto3 Bucket object
-    assert bucket.name == bucket_name
+    assert bucket.name == env.bucket_name

tests/s3/test_s3_artifacts.py

@@ -4,7 +4,6 @@
 
 from gardenlinux.s3.s3_artifacts import S3Artifacts
 
-CNAME = "testcname"
 RELEASE_DATA = """
     GARDENLINUX_VERSION = 1234.1
     GARDENLINUX_COMMIT_ID = abc123
@@ -15,13 +14,13 @@
 
 def test_s3artifacts_init_success(s3_setup):
     # Arrange
-    _, bucket_name, _ = s3_setup
+    env = s3_setup
 
     # Act
-    s3_artifacts = S3Artifacts(bucket_name)
+    s3_artifacts = S3Artifacts(env.bucket_name)
 
     # Assert
-    assert s3_artifacts._bucket.name == bucket_name
+    assert s3_artifacts._bucket.name == env.bucket_name
 
 
 def tets_s3artifacts_invalid_bucket():
@@ -35,22 +34,22 @@ def test_download_to_directory_success(s3_setup):
     Test download of multiple files to a directory on disk.
     """
     # Arrange
-    s3, bucket_name, _ = s3_setup
-    bucket = s3.Bucket(bucket_name)
+    env = s3_setup
+    bucket = env.s3.Bucket(env.bucket_name)
 
-    bucket.put_object(Key=f"meta/singles/{CNAME}", Body=b"metadata")
-    bucket.put_object(Key=f"objects/{CNAME}/file1", Body=b"data1")
-    bucket.put_object(Key=f"objects/{CNAME}/file2", Body=b"data2")
+    bucket.put_object(Key=f"meta/singles/{env.cname}", Body=b"metadata")
+    bucket.put_object(Key=f"objects/{env.cname}/file1", Body=b"data1")
+    bucket.put_object(Key=f"objects/{env.cname}/file2", Body=b"data2")
 
     with TemporaryDirectory() as tmpdir:
         outdir = Path(tmpdir)
 
         # Act
-        artifacts = S3Artifacts(bucket_name)
-        artifacts.download_to_directory(CNAME, outdir)
+        artifacts = S3Artifacts(env.bucket_name)
+        artifacts.download_to_directory(env.cname, outdir)
 
         # Assert
-        assert (outdir / f"{CNAME}.s3_metadata.yaml").read_bytes() == b"metadata"
+        assert (outdir / f"{env.cname}.s3_metadata.yaml").read_bytes() == b"metadata"
         assert (outdir / "file1").read_bytes() == b"data1"
         assert (outdir / "file2").read_bytes() == b"data2"
 
@@ -60,67 +59,67 @@ def test_download_to_directory_invalid_path(s3_setup):
     Test proper handling of download attempt to invalid path.
     """
     # Arrange
-    _, bucket_name, _ = s3_setup
-    artifacts = S3Artifacts(bucket_name)
+    env = s3_setup
+    artifacts = S3Artifacts(env.bucket_name)
 
     # Act / Assert
     with pytest.raises(RuntimeError):
-        artifacts.download_to_directory({CNAME}, "/invalid/path/does/not/exist")
+        artifacts.download_to_directory({env.cname}, "/invalid/path/does/not/exist")
 
 
 def test_upload_from_directory_success(s3_setup):
     """
     Test upload of multiple artifacts from disk to bucket
     """
     # Arrange
-    s3, bucket_name, tmp_path = s3_setup
+    env = s3_setup
 
-    release_path = tmp_path / f"{CNAME}.release"
+    release_path = env.tmp_path / f"{env.cname}.release"
     release_path.write_text(RELEASE_DATA)
 
-    for filename in [f"{CNAME}-file1", f"{CNAME}-file2"]:
-        (tmp_path / filename).write_bytes(b"dummy content")
+    for filename in [f"{env.cname}-file1", f"{env.cname}-file2"]:
+        (env.tmp_path / filename).write_bytes(b"dummy content")
 
     # Act
-    artifacts = S3Artifacts(bucket_name)
-    artifacts.upload_from_directory(CNAME, tmp_path)
+    artifacts = S3Artifacts(env.bucket_name)
+    artifacts.upload_from_directory(env.cname, env.tmp_path)
 
     # Assert
-    bucket = s3.Bucket(bucket_name)
+    bucket = env.s3.Bucket(env.bucket_name)
     keys = [obj.key for obj in bucket.objects.all()]
-    assert f"objects/{CNAME}/{CNAME}-file1" in keys
-    assert f"objects/{CNAME}/{CNAME}-file2" in keys
-    assert f"meta/singles/{CNAME}" in keys
+    assert f"objects/{env.cname}/{env.cname}-file1" in keys
+    assert f"objects/{env.cname}/{env.cname}-file2" in keys
+    assert f"meta/singles/{env.cname}" in keys
 
 
 def test_upload_from_directory_with_delete(s3_setup):
     """
     Test that upload_from_directory deletes existing files before uploading
     when delete_before_push=True.
     """
-    s3, bucket_name, tmp_path = s3_setup
-    bucket = s3.Bucket(bucket_name)
+    env = s3_setup
+    bucket = env.s3.Bucket(env.bucket_name)
 
     # Arrange: create release and artifact files locally
-    release = tmp_path / f"{CNAME}.release"
+    release = env.tmp_path / f"{env.cname}.release"
     release.write_text(RELEASE_DATA)
 
-    artifact = tmp_path / f"{CNAME}.kernel"
+    artifact = env.tmp_path / f"{env.cname}.kernel"
     artifact.write_bytes(b"fake")
 
     # Arrange: put dummy existing objects to be deleted
-    bucket.put_object(Key=f"objects/{CNAME}/{artifact.name}", Body=b"old data")
-    bucket.put_object(Key=f"meta/singles/{CNAME}", Body=b"old metadata")
+    bucket.put_object(Key=f"objects/{env.cname}/{artifact.name}", Body=b"old data")
+    bucket.put_object(Key=f"meta/singles/{env.cname}", Body=b"old metadata")
 
-    artifacts = S3Artifacts(bucket_name)
+    artifacts = S3Artifacts(env.bucket_name)
 
     # Act
-    artifacts.upload_from_directory(CNAME, tmp_path, delete_before_push=True)
+    artifacts.upload_from_directory(env.cname, env.tmp_path, delete_before_push=True)
 
     # Assert
     keys = [obj.key for obj in bucket.objects.all()]
 
     # The old key should no longer be present as old data (no duplicates)
     # but the new upload file key should exist (artifact uploaded)
-    assert f"objects/{CNAME}/{artifact.name}" in keys
-    assert f"meta/singles/{CNAME}" in keys
+    assert f"objects/{env.cname}/{artifact.name}" in keys
+    assert f"meta/singles/{env.cname}" in keys