[#347] add ape bearer test acl 0
Signed-off-by: Ekaterina Chernitsyna <e.chernitsyna@yadro.com>
This commit is contained in:
parent
590e6508d9
commit
96047aecdc
1 changed files with 80 additions and 0 deletions
80
pytest_tests/testsuites/ape/test_ape_bearer.py
Normal file
80
pytest_tests/testsuites/ape/test_ape_bearer.py
Normal file
|
@ -0,0 +1,80 @@
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
|
||||||
|
import allure
|
||||||
|
import pytest
|
||||||
|
from frostfs_testlib import reporter
|
||||||
|
from frostfs_testlib.credentials.interfaces import User
|
||||||
|
from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
|
||||||
|
from frostfs_testlib.resources.error_patterns import NO_RULE_FOUND_OBJECT, RULE_ACCESS_DENIED_OBJECT
|
||||||
|
from frostfs_testlib.steps.cli.object import put_object
|
||||||
|
from frostfs_testlib.storage.dataclasses.ape import Operations
|
||||||
|
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
|
||||||
|
from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper
|
||||||
|
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
|
||||||
|
from frostfs_testlib.testing.test_control import expect_not_raises
|
||||||
|
from frostfs_testlib.utils import datetime_utils
|
||||||
|
from frostfs_testlib.utils.file_utils import generate_file
|
||||||
|
from frostfs_testlib.utils.string_utils import unique_name
|
||||||
|
from frostfs_testlib_plugin_to.storage.capi.capi_client import CApiClient
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.ape
|
||||||
|
@pytest.mark.ape_bearer
|
||||||
|
class TestApeBearer(ClusterTestBase):
|
||||||
|
@allure.title("BT with container basic-acl=0 root")
|
||||||
|
def test_BT_with_container_basic_acl_0_root(
|
||||||
|
self,
|
||||||
|
grpc_client: CliClientWrapper,
|
||||||
|
simple_object_size: ObjectSize,
|
||||||
|
temp_directory: str,
|
||||||
|
default_user: User,
|
||||||
|
):
|
||||||
|
test_file = generate_file(simple_object_size.value)
|
||||||
|
|
||||||
|
with reporter.step("Create container"):
|
||||||
|
cid = grpc_client.container.create(
|
||||||
|
endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
|
||||||
|
policy="REP 2 IN X CBF 1 SELECT 2 FROM * AS X",
|
||||||
|
await_mode=True,
|
||||||
|
basic_acl="0",
|
||||||
|
)
|
||||||
|
|
||||||
|
chain_file = os.path.join(temp_directory, "serialized_chain.json")
|
||||||
|
bt_file = os.path.join(temp_directory, "bt.json")
|
||||||
|
bt_sign_file = os.path.join(temp_directory, "bt-sign.json")
|
||||||
|
|
||||||
|
with reporter.step("Generate APE override by target and APE chains"):
|
||||||
|
grpc_client.cli.bearer.generate_ape_override(
|
||||||
|
rule="allow Object.Put *",
|
||||||
|
chain_id="allowPutObjBT",
|
||||||
|
cid=cid,
|
||||||
|
output=chain_file,
|
||||||
|
)
|
||||||
|
|
||||||
|
with reporter.step("Create bearer token"):
|
||||||
|
grpc_client.cli.bearer.create(
|
||||||
|
rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
|
||||||
|
ape=chain_file,
|
||||||
|
issued_at=2,
|
||||||
|
expire_at=1000,
|
||||||
|
out=bt_file,
|
||||||
|
)
|
||||||
|
|
||||||
|
with reporter.step("Sign bearer token"):
|
||||||
|
grpc_client.cli.util.sign_bearer_token(from_file=bt_file, to_file=bt_sign_file)
|
||||||
|
|
||||||
|
with reporter.step("[NEGATIVE] Put object without token"):
|
||||||
|
with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT.format(operation=Operations.PUT_OBJECT.value)):
|
||||||
|
put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
|
||||||
|
|
||||||
|
with reporter.step("Put object with bearer token"):
|
||||||
|
with expect_not_raises():
|
||||||
|
put_object(
|
||||||
|
default_user.wallet,
|
||||||
|
test_file,
|
||||||
|
cid,
|
||||||
|
self.shell,
|
||||||
|
self.cluster.storage_nodes[0].get_rpc_endpoint(),
|
||||||
|
bearer=bt_sign_file,
|
||||||
|
)
|
Loading…
Reference in a new issue