frostfs-testcases/pytest_tests/testsuites/object/test_object_lifetime.py

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.resources.common import STORAGE_GC_TIME
from frostfs_testlib.resources.error_patterns import OBJECT_ALREADY_REMOVED, OBJECT_NOT_FOUND
from frostfs_testlib.steps.cli.container import DEFAULT_EC_PLACEMENT_RULE, DEFAULT_PLACEMENT_RULE
from frostfs_testlib.storage.grpc_operations.interfaces_wrapper import GrpcClientWrapper
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.testing.test_control import wait_for_success
from frostfs_testlib.utils import datetime_utils
from frostfs_testlib.utils.file_utils import TestFile

from ...helpers.container_request import APE_EVERYONE_ALLOW_ALL, ContainerRequest


@wait_for_success(datetime_utils.parse_time(STORAGE_GC_TIME) * 5, datetime_utils.parse_time(STORAGE_GC_TIME))
def wait_for_object_status_change_to(status: str, grpc_client: GrpcClientWrapper, cid: str, oid: str, endpoint: str) -> None:
    with pytest.raises(Exception, match=status):
        grpc_client.object.head(cid, oid, endpoint)


@pytest.mark.nightly
@pytest.mark.sanity
@pytest.mark.grpc_api
class TestObjectApiLifetime(ClusterTestBase):
    @allure.title("Object is removed when lifetime expired (obj_size={object_size}, policy={container_request.short_name})")
    @pytest.mark.parametrize(
        "container_request",
        [
            ContainerRequest(DEFAULT_PLACEMENT_RULE, APE_EVERYONE_ALLOW_ALL, "REP 2"),
            ContainerRequest(DEFAULT_EC_PLACEMENT_RULE, APE_EVERYONE_ALLOW_ALL, "EC 3.1"),
        ],
    )
    def test_object_api_lifetime(self, grpc_client: GrpcClientWrapper, container: str, test_file: TestFile):
        """
        Test object deleted after expiration epoch.
        """

        with reporter.step("Get current epoch"):
            current_epoch = self.get_epoch()
            last_active_epoch = current_epoch + 1

        with reporter.step("Put object to random node"):
            oid = grpc_client.object.put_to_random_node(test_file, container, self.cluster, expire_at=last_active_epoch)

        with reporter.step("Ensure that expiration of object has expected value"):
            object_info: dict = grpc_client.object.head(container, oid, self.cluster.default_rpc_endpoint)
            expiration_epoch = int(object_info["header"]["attributes"]["__SYSTEM__EXPIRATION_EPOCH"])
            assert expiration_epoch == last_active_epoch, f"Expiration time set for object is not expected: {expiration_epoch}"

        with reporter.step("Tick two epoch for object expiration"):
            self.tick_epochs(2)

        with reporter.step("Wait until GC marks object as 'already removed' or 'not found'"):
            wait_for_object_status_change_to(
                f"{OBJECT_ALREADY_REMOVED}|{OBJECT_NOT_FOUND}", grpc_client, container, oid, self.cluster.default_rpc_endpoint
            )

        with reporter.step("Try to get object from random node and make sure it is really deleted"):
            with pytest.raises(Exception, match=f"{OBJECT_ALREADY_REMOVED}|{OBJECT_NOT_FOUND}"):
                grpc_client.object.get_from_random_node(container, oid, self.cluster)