Version 0.4.13

pytest_minio_mock/__init__.py

@@ -17,7 +17,7 @@
 
 __title__ = "pytest-minio-mock"
 __description__ = "A pytest plugin for mocking Minio S3 interactions"
-__version__ = "0.3.13"
+__version__ = "0.4.13"
 __status__ = "Production"
 __license__ = "MIT"
 __author__ = "Oussama Jarrousse"

pytest_minio_mock/plugin.py

@@ -21,7 +21,6 @@
 application interacts correctly with Minio, without the overhead of connecting to an actual Minio
 server.
 """
-
 import copy
 import datetime
 import io
@@ -33,7 +32,8 @@
 import validators
 from minio import Minio
 from minio.commonconfig import ENABLED
-from minio.datatypes import Object, Bucket
+from minio.datatypes import Bucket
+from minio.datatypes import Object
 from minio.error import S3Error
 from minio.versioningconfig import OFF
 from minio.versioningconfig import SUSPENDED
@@ -221,6 +221,33 @@ def put_object(
         self.put_object_version(version_id, obj)
         return obj
 
+    def stat_object(
+        self,
+        version_id,
+        versioning: VersioningConfig,
+        ssec=None,
+        extra_headers=None,
+        extra_query_params=None,
+    ):
+        """
+        Returns
+            the stat of the object if versioning is disabled
+            the stat of the latest version of the object if versioning is enabled
+        """
+        obj_version = self.get_object(version_id=version_id, versioning=versioning)
+        the_stat_object_version = Object(
+            bucket_name=self.bucket_name,
+            object_name=self.object_name,
+            last_modified=obj_version.last_modified,
+            version_id=None
+            if obj_version.version_id == "null"
+            else obj_version.version_id,
+            is_latest="true" if obj_version.is_latest else "false",
+            is_delete_marker=obj_version.is_delete_marker,
+        )
+
+        return the_stat_object_version
+
     def get_object(self, version_id, versioning: VersioningConfig):
         """
         Returns
@@ -451,7 +478,7 @@ def put_object(
     def remove_object(self, object_name, version_id=None):
         """ """
         if object_name not in self.objects:
-            # object does not exist, so nothing to do
+            # Object does not exist, so nothing to do
             return
         try:
             if self.versioning.status == OFF:
@@ -495,6 +522,63 @@ def get_object(self, object_name, version_id):
             )
         return the_object_version
 
+    def stat_object(
+        self,
+        object_name,
+        ssec=None,
+        version_id=None,
+        extra_headers=None,
+        extra_query_params=None,
+    ) -> Object:
+        """
+        Get object information and metadata of an object in the mock Minio server
+
+        Args:
+            object_name (str): The name of the object to remove.
+            ssec (SseCustomerKey| None, optional): Server-side encryption customer key.
+            version_id (str | None, optional): The version about which to retrieve information and metadata.
+            extra_headers ( dict | None, optional ):
+            extra_query_params ( dict | None, optional): Extra query parameters for advanced usage.
+
+        Returns:
+            Object: Object information as Object.
+        """
+
+        try:
+            the_object = self.objects[object_name]
+        except KeyError as e:
+            raise S3Error(
+                message="Object does not exist",
+                resource=f"/{self.bucket_name}/{object_name}",
+                request_id=None,
+                host_id=None,
+                response="mocked_response",
+                code="NoSuchKey",
+                bucket_name=self.bucket_name,
+                object_name=object_name,
+            )
+
+        try:
+            the_object_version_stat = the_object.stat_object(
+                version_id=version_id,
+                versioning=self.versioning,
+                ssec=ssec,
+                extra_headers=extra_headers,
+                extra_query_params=extra_query_params,
+            )
+        except S3Error as e:
+            raise S3Error(
+                message=e.message,
+                response=e.response,
+                resource=f"/{self.bucket_name}/{object_name}",
+                host_id=None,
+                request_id=None,
+                code=e.code,
+                bucket_name=self.bucket_name,
+                object_name=object_name,
+            )
+        return the_object_version_stat
+
     def list_objects(
         self,
         prefix=None,
@@ -537,12 +621,12 @@ def list_objects(
                     # Minio API always sort versions by time,
                     # it also includes delete markers at the end newwst first
                     versions_list = obj.list_versions()
-                    for version, obj_version in versions_list:
+                    for version_id, obj_version in versions_list:
                         yield Object(
                             bucket_name=self.bucket_name,
                             object_name=object_name,
                             last_modified=obj_version.last_modified,
-                            version_id=version,
+                            version_id=version_id,
                             is_latest="true" if obj_version.is_latest else "false",
                             is_delete_marker=obj_version.is_delete_marker,
                         )
@@ -774,7 +858,6 @@ def fget_object(
             object_name,
             version_id=version_id,
             request_headers=request_headers,
-            sse=sse,
             extra_query_params=extra_query_params,
         )
         with open(file_path, "wb") as f:
@@ -787,7 +870,7 @@ def get_object(
         offset: int = 0,
         length: int = 0,
         request_headers=None,
-        sse=None,
+        ssec=None,
         version_id=None,
         extra_query_params=None,
     ):
@@ -805,7 +888,7 @@ def get_object(
             length (int, optional): The number of bytes of object data to retrieve. Defaults to 0,
                 which means the whole object.
             request_headers (dict, optional): Additional headers for the request. Defaults to None.
-            sse (optional): Server-side encryption option. Defaults to None.
+            ssec ( SseCustomerKey | None, optional): Server-side encryption option. Defaults to None.
             version_id (str | None, optional): The version ID of the object. Defaults to None.
             extra_query_params (dict, optional): Additional query parameters. Defaults to None.
 
@@ -1075,7 +1158,10 @@ def list_buckets(self):
         """
         try:
             self._health_check()
-            return [Bucket(name, bucket._creation_date) for (name, bucket) in list(self.buckets.items())]
+            return [
+                Bucket(name, bucket._creation_date)
+                for (name, bucket) in list(self.buckets.items())
+            ]
         except Exception as e:
             logging.error(e)
             raise e
@@ -1261,6 +1347,48 @@ def remove_object(self, bucket_name, object_name, version_id=None):
             object_name, version_id=version_id
         )
 
+    def stat_object(
+        self,
+        bucket_name,
+        object_name,
+        ssec=None,
+        version_id=None,
+        extra_headers=None,
+        extra_query_params=None,
+    ) -> Object:
+        """
+        Get object information and metadata of an object in the mock Minio server
+
+        Args:
+            bucket_name (str): The name of the bucket.
+            object_name (str): The name of the object to remove.
+            ssec (SseCustomerKey| None, optional): Server-side encryption customer key.
+            version_id (str | None, optional): The version about which to retrieve information and metadata.
+            extra_headers ( dict | None, optional ):
+            extra_query_params ( dict | None, optional): Extra query parameters for advanced usage.
+
+        Raises:
+            S3Error:
+
+        Returns:
+            Object: Object information as Object.
+
+        """
+        self._health_check()
+
+        the_stat_object = self.buckets[bucket_name].stat_object(
+            object_name=object_name,
+            ssec=ssec,
+            version_id=version_id,
+            extra_headers=extra_headers,
+            extra_query_params=extra_query_params,
+        )
+
+        if not the_stat_object:
+            raise RuntimeError("Implementation Error")
+
+        return the_stat_object
+
 
 @pytest.fixture
 def minio_mock_servers():

setup.py

@@ -21,7 +21,7 @@
     long_description=open("README.md").read(),
     keywords="pytest minio mock",
     extras_require={"dev": ["pre-commit", "tox"]},
-    version="0.3.13",
+    version="0.4.13",
     long_description_content_type="text/markdown",
     classifiers=[
         "Framework :: Pytest",

tests/test_minio_mock.py

@@ -4,11 +4,11 @@
 import validators
 from minio import Minio
 from minio.commonconfig import ENABLED
+from minio.datatypes import Bucket
 from minio.error import S3Error
 from minio.versioningconfig import OFF
 from minio.versioningconfig import SUSPENDED
 from minio.versioningconfig import VersioningConfig
-from minio.datatypes import Bucket
 
 from pytest_minio_mock.plugin import MockMinioBucket
 from pytest_minio_mock.plugin import MockMinioObject
@@ -447,3 +447,49 @@ def test_connecting_to_the_same_endpoint(minio_mock):
     client_2 = Minio("http://local.host:9000")
     client_2_buckets = client_2.list_buckets()
     assert client_2_buckets == client_1_buckets
+
+
+@pytest.mark.UNIT
+def test_stat_object(minio_mock):
+    bucket_name = "test-bucket"
+    object_name = "test-object"
+    file_path = "tests/fixtures/maya.jpeg"
+
+    client = Minio("http://local.host:9000")
+    client.make_bucket(bucket_name)
+    client.fput_object(bucket_name, object_name, file_path)
+
+    object_stat = client.stat_object(bucket_name=bucket_name, object_name=object_name)
+
+    assert object_stat.bucket_name == bucket_name
+    assert object_stat.object_name == object_name
+    assert object_stat.version_id is None
+
+    client.remove_object(bucket_name, object_name)
+
+    with pytest.raises(S3Error) as error:
+        _ = client.stat_object(bucket_name=bucket_name, object_name=object_name)
+    assert error.value.code == "NoSuchKey"
+    assert error.value.message == "Object does not exist"
+
+    client.fput_object(bucket_name, object_name, file_path)
+    client.set_bucket_versioning(bucket_name, VersioningConfig(ENABLED))
+
+    object_stat = client.stat_object(bucket_name=bucket_name, object_name=object_name)
+    assert object_stat.bucket_name == bucket_name
+    assert object_stat.object_name == object_name
+    assert object_stat.version_id is None
+    object_stat = client.stat_object(
+        bucket_name=bucket_name, object_name=object_name, version_id="null"
+    )
+    assert object_stat.bucket_name == bucket_name
+    assert object_stat.object_name == object_name
+    assert object_stat.version_id is None
+    client.fput_object(bucket_name, object_name, file_path)
+    objects = list(client.list_objects(bucket_name=bucket_name, include_version=True))
+    object_stat = client.stat_object(
+        bucket_name=bucket_name,
+        object_name=object_name,
+        version_id=objects[1].version_id,
+    )
+    assert object_stat.version_id is None