frostfs-testcases/pytest_tests/testsuites/services/http_gate/test_http_streaming.py
2024-11-26 19:42:53 +03:00

52 lines
2 KiB
Python

import logging
import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.steps.http.http_gate import upload_via_http_gate_curl, verify_object_hash
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.utils.file_utils import generate_file
from ....helpers.container_request import REP_2_1_4_PUBLIC, requires_container
logger = logging.getLogger("NeoLogger")
@pytest.mark.http_gate
@pytest.mark.http_put
class Test_http_streaming(ClusterTestBase):
@allure.title("Put via pipe (streaming), Get over HTTP and verify hashes")
@requires_container(REP_2_1_4_PUBLIC)
def test_object_can_be_put_get_by_streaming(self, default_wallet: WalletInfo, container: str, complex_object_size: ObjectSize):
"""
Test that object can be put using gRPC interface and get using HTTP.
Steps:
1. Create big object;
2. Put object using curl with pipe (streaming);
3. Download object using HTTP gate (https://git.frostfs.info/TrueCloudLab/frostfs-http-gw#downloading);
4. Compare hashes between original and downloaded object;
Expected result:
Hashes must be the same.
"""
with reporter.step("Allocate big object"):
# Generate file
file_path = generate_file(complex_object_size.value)
with reporter.step("Put objects using curl utility"):
oid = upload_via_http_gate_curl(container, file_path, self.cluster.default_http_gate_endpoint)
with reporter.step("Get object and verify hashes [ get/$CID/$OID ]"):
verify_object_hash(
oid=oid,
file_name=file_path,
wallet=default_wallet,
cid=container,
shell=self.shell,
nodes=self.cluster.storage_nodes,
request_node=self.cluster.cluster_nodes[0],
)