diff --git a/pytest_minio_mock/__init__.py b/pytest_minio_mock/__init__.py index a4ab0c9..373d1d9 100644 --- a/pytest_minio_mock/__init__.py +++ b/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" diff --git a/pytest_minio_mock/plugin.py b/pytest_minio_mock/plugin.py index 51cb855..39810ff 100644 --- a/pytest_minio_mock/plugin.py +++ b/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(): diff --git a/setup.py b/setup.py index e0bdf2a..2ec9aac 100644 --- a/setup.py +++ b/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", diff --git a/tests/test_minio_mock.py b/tests/test_minio_mock.py index 9b934fc..0395781 100644 --- a/tests/test_minio_mock.py +++ b/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