EliChin/frostfs-testcases
1
0
Fork 0
forked from TrueCloudLab/frostfs-testcases
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add drop locked objects tests

Browse source 
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
This commit is contained in:
Andrey Berezin 2022-12-12 11:51:11 +03:00 committed by Julia Kovshova
parent 25a9443209
commit de6434934d
2 changed files with 109 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
pytest_tests/pytest.ini
View file

@ -13,6 +13,7 @@ markers =
# functional markers
container: tests for container creation
grpc_api: standard gRPC API tests
grpc_control: tests related to using neofs-cli control commands
grpc_object_lock: gRPC lock tests
http_gate: HTTP gate contract
s3_gate: All S3 gate tests

111
pytest_tests/testsuites/object/test_object_lock.py
View file

@ -18,17 +18,17 @@ from grpc_responses import (
OBJECT_NOT_FOUND,
)
from neofs_testlib.shell import Shell
from node_management import drop_object
from pytest import FixtureRequest
from python_keywords.container import create_container
from python_keywords.neofs_verbs import delete_object, head_object, lock_object
from storage_policy import get_nodes_with_object
from test_control import expect_not_raises, wait_for_success
from utility import parse_time, wait_for_gc_pass_on_storage_nodes
import steps
from helpers.container import StorageContainer, StorageContainerInfo
from helpers.storage_object_info import LockObjectInfo, StorageObjectInfo
from helpers.wallet import WalletFactory, WalletFile
from steps.cluster_test_base import ClusterTestBase
from steps.storage_object import delete_objects
logger = logging.getLogger("NeoLogger")
@ -65,6 +65,9 @@ def locked_storage_object(
cluster: Cluster,
request: FixtureRequest,
):
"""
Intention of this fixture is to provide storage object which is NOT expected to be deleted during test act phase
"""
with allure.step("Creating locked object"):
current_epoch = ensure_fresh_epoch(client_shell, cluster)
expiration_epoch = current_epoch + FIXTURE_LOCK_LIFETIME
@ -117,7 +120,32 @@ def locked_storage_object(
@pytest.mark.sanity
@pytest.mark.grpc_object_lock
class TestObjectLockWithGrpc(ClusterTestBase):
def get_storage_object_chunks(self, storage_object: StorageObjectInfo):
@pytest.fixture()
def new_locked_storage_object(
self, user_container: StorageContainer, request: FixtureRequest
) -> StorageObjectInfo:
"""
Intention of this fixture is to provide new storage object for tests which may delete or corrupt the object or it's complementary objects
So we need a new one each time we ask for it
"""
with allure.step("Creating locked object"):
current_epoch = self.get_epoch()
storage_object = user_container.generate_object(
request.param, expire_at=current_epoch + FIXTURE_OBJECT_LIFETIME
)
lock_object(
storage_object.wallet_file_path,
storage_object.cid,
storage_object.oid,
self.shell,
self.cluster.default_rpc_endpoint,
lifetime=FIXTURE_LOCK_LIFETIME,
)
return storage_object
def get_storage_object_chunks(self, storage_object: StorageObjectInfo) -> list[str]:
with allure.step(f"Get complex object chunks (f{storage_object.oid})"):
split_object_id = get_link_object(
storage_object.wallet_file_path,
@ -539,6 +567,83 @@ class TestObjectLockWithGrpc(ClusterTestBase):
self.cluster.default_rpc_endpoint,
)
@allure.title("Link object of locked complex object can be dropped")
@pytest.mark.grpc_control
@pytest.mark.parametrize(
"new_locked_storage_object",
# Only complex object is required
[pytest.lazy_fixture("complex_object_size")],
indirect=True,
)
def test_link_object_of_locked_complex_object_can_be_dropped(
self, new_locked_storage_object: StorageObjectInfo
):
link_object_id = get_link_object(
new_locked_storage_object.wallet_file_path,
new_locked_storage_object.cid,
new_locked_storage_object.oid,
self.shell,
self.cluster.storage_nodes,
)
with allure.step(f"Drop link object with id {link_object_id} from nodes"):
nodes_with_object = get_nodes_with_object(
new_locked_storage_object.cid,
link_object_id,
shell=self.shell,
nodes=self.cluster.storage_nodes,
)
for node in nodes_with_object:
with expect_not_raises():
drop_object(node, new_locked_storage_object.cid, link_object_id)
@allure.title("Chunks of locked complex object can be dropped")
@pytest.mark.grpc_control
@pytest.mark.parametrize(
"new_locked_storage_object",
# Only complex object is required
[pytest.lazy_fixture("complex_object_size")],
indirect=True,
)
def test_chunks_of_locked_complex_object_can_be_dropped(
self, new_locked_storage_object: StorageObjectInfo
):
chunk_objects = self.get_storage_object_chunks(new_locked_storage_object)
for chunk_object_id in chunk_objects:
with allure.step(f"Drop chunk object with id {chunk_object_id} from nodes"):
nodes_with_object = get_nodes_with_object(
new_locked_storage_object.cid,
chunk_object_id,
shell=self.shell,
nodes=self.cluster.storage_nodes,
)
for node in nodes_with_object:
with expect_not_raises():
drop_object(node, new_locked_storage_object.cid, chunk_object_id)
@pytest.mark.grpc_control
@pytest.mark.parametrize(
"new_locked_storage_object",
[pytest.lazy_fixture("simple_object_size"), pytest.lazy_fixture("complex_object_size")],
ids=["simple object", "complex object"],
indirect=True,
)
def test_locked_object_can_be_dropped(
self, new_locked_storage_object: StorageObjectInfo, request: FixtureRequest
):
allure.dynamic.title(f"Locked {request.node.callspec.id} can be dropped")
nodes_with_object = get_nodes_with_object(
new_locked_storage_object.cid,
new_locked_storage_object.oid,
shell=self.shell,
nodes=self.cluster.storage_nodes,
)
for node in nodes_with_object:
with expect_not_raises():
drop_object(node, new_locked_storage_object.cid, new_locked_storage_object.oid)
@allure.title("Link object of complex object should also be protected from deletion")
@pytest.mark.parametrize(
# Only complex objects are required for this test