diff --git a/pytest_tests/testsuites/object/test_object_api.py b/pytest_tests/testsuites/object/test_object_api.py index 700cdb4a..1370a696 100755 --- a/pytest_tests/testsuites/object/test_object_api.py +++ b/pytest_tests/testsuites/object/test_object_api.py @@ -12,7 +12,7 @@ from frostfs_testlib.resources.error_patterns import ( INVALID_RANGE_ZERO_LENGTH, OUT_OF_RANGE, ) -from frostfs_testlib.shell import Shell +from frostfs_testlib.shell import Shell, CommandResult from frostfs_testlib.steps.cli.container import create_container from frostfs_testlib.steps.cli.object import ( get_object_from_random_node, @@ -21,6 +21,7 @@ from frostfs_testlib.steps.cli.object import ( head_object, put_object_to_random_node, search_object, + delete_object, ) from frostfs_testlib.steps.complex_object_actions import get_complex_object_split_ranges from frostfs_testlib.steps.storage_object import delete_objects @@ -207,6 +208,34 @@ class TestObjectApi(ClusterTestBase): ) self.check_header_is_presented(head_info, storage_object_2.attributes) + def test_object_head_raw(self, default_wallet: str, object_size: ObjectSize, client_shell: Shell): + """ + Execute command object head --raw and check object deletion + """ + shell = client_shell + wallet = default_wallet + cid = create_container(wallet, shell, self.cluster.default_rpc_endpoint) + + file_path = generate_file(object_size.value) + oid = put_object_to_random_node(wallet, file_path, cid, shell, self.cluster) + + delete_object(wallet, cid, oid, shell, self.cluster.default_rpc_endpoint) + + with reporter.step("Execute command object head --raw"): + try: + result = head_object(wallet, cid, oid, shell, self.cluster.default_rpc_endpoint, is_raw=True) + if result.return_code != 0: + print("Expected error occurred while executing head_object, continuing test execution.") + else: + raise AssertionError("Unexpected successful execution of head_object.") + except Exception as e: + if "object already removed" in str(e): + print("Expected error occurred: object already removed") + else: + print(f"An unexpected error occurred: {e}") + raise + + @allure.title("Search objects by native API (obj_size={object_size})") def test_search_object_api(self, storage_objects: list[StorageObjectInfo]): """