diff --git a/pytest_tests/helpers/file_helper.py b/pytest_tests/helpers/file_helper.py index 324c307b..d5a69c32 100644 --- a/pytest_tests/helpers/file_helper.py +++ b/pytest_tests/helpers/file_helper.py @@ -19,7 +19,7 @@ def generate_file(size: int = SIMPLE_OBJ_SIZE) -> str: Returns: The path to the generated file. """ - file_path = f"{os.getcwd()}/{ASSETS_DIR}/{str(uuid.uuid4())}" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) with open(file_path, "wb") as file: file.write(os.urandom(size)) logger.info(f"File with size {size} bytes has been generated: {file_path}") @@ -48,7 +48,7 @@ def generate_file_with_content( mode = "wb" if not file_path: - file_path = f"{os.getcwd()}/{ASSETS_DIR}/{str(uuid.uuid4())}" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) else: if not os.path.exists(os.path.dirname(file_path)): os.makedirs(os.path.dirname(file_path)) @@ -91,7 +91,7 @@ def concat_files(file_paths: list, resulting_file_path: Optional[str] = None) -> Path to the resulting file. """ if not resulting_file_path: - resulting_file_path = f"{os.getcwd()}/{ASSETS_DIR}/{str(uuid.uuid4())}" + resulting_file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) with open(resulting_file_path, "wb") as f: for file in file_paths: with open(file, "rb") as part_file: diff --git a/pytest_tests/steps/aws_cli_client.py b/pytest_tests/steps/aws_cli_client.py index 8ed1ba11..7c5cc4d0 100644 --- a/pytest_tests/steps/aws_cli_client.py +++ b/pytest_tests/steps/aws_cli_client.py @@ -255,7 +255,7 @@ class AwsCliClient: return self._to_json(output) def delete_objects(self, Bucket: str, Delete: dict) -> dict: - file_path = f"{os.getcwd()}/{ASSETS_DIR}/delete.json" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, "delete.json") with open(file_path, "w") as out_file: out_file.write(json.dumps(Delete)) logger.info(f"Input file for delete-objects: {json.dumps(Delete)}") @@ -448,7 +448,7 @@ class AwsCliClient: def complete_multipart_upload( self, Bucket: str, Key: str, UploadId: str, MultipartUpload: dict ) -> dict: - file_path = f"{os.getcwd()}/{ASSETS_DIR}/parts.json" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, "parts.json") with open(file_path, "w") as out_file: out_file.write(json.dumps(MultipartUpload)) logger.info(f"Input file for complete-multipart-upload: {json.dumps(MultipartUpload)}") diff --git a/pytest_tests/steps/s3_gate_object.py b/pytest_tests/steps/s3_gate_object.py index e35229b6..a3b06a4f 100644 --- a/pytest_tests/steps/s3_gate_object.py +++ b/pytest_tests/steps/s3_gate_object.py @@ -1,5 +1,3 @@ -#!/usr/bin/python3.9 - import logging import os import uuid @@ -243,7 +241,7 @@ def get_object_acl_s3( def copy_object_s3( s3_client, bucket: str, object_key: str, bucket_dst: Optional[str] = None, **kwargs ) -> str: - filename = f"{os.getcwd()}/{uuid.uuid4()}" + filename = os.path.join(os.getcwd(), str(uuid.uuid4())) try: params = { "Bucket": bucket_dst or bucket, @@ -280,7 +278,7 @@ def get_object_s3( range: Optional[list] = None, full_output: bool = False, ): - filename = f"{ASSETS_DIR}/{uuid.uuid4()}" + filename = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) try: params = {"Bucket": bucket, "Key": object_key} if version_id: diff --git a/pytest_tests/testsuites/conftest.py b/pytest_tests/testsuites/conftest.py index 6926ab2f..f935896e 100644 --- a/pytest_tests/testsuites/conftest.py +++ b/pytest_tests/testsuites/conftest.py @@ -70,7 +70,7 @@ def check_binary_versions(request, hosting: Hosting, client_shell: Shell): @pytest.fixture(scope="session") @allure.title("Prepare tmp directory") def prepare_tmp_dir(): - full_path = f"{os.getcwd()}/{ASSETS_DIR}" + full_path = os.path.join(os.getcwd(), ASSETS_DIR) shutil.rmtree(full_path, ignore_errors=True) os.mkdir(full_path) yield full_path diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py b/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py index 332fcc38..41735225 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py @@ -169,8 +169,8 @@ class TestS3Gate(TestS3GateBase): """ Test checks sync directory with AWS CLI utility. """ - file_path_1 = f"{os.getcwd()}/{ASSETS_DIR}/test_sync/test_file_1" - file_path_2 = f"{os.getcwd()}/{ASSETS_DIR}/test_sync/test_file_2" + file_path_1 = os.path.join(os.getcwd(), ASSETS_DIR, "test_sync", "test_file_1") + file_path_2 = os.path.join(os.getcwd(), ASSETS_DIR, "test_sync", "test_file_2") key_to_path = {"test_file_1": file_path_1, "test_file_2": file_path_2} if not isinstance(self.s3_client, AwsCliClient): diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py index b3d3f614..89402e21 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py @@ -852,8 +852,8 @@ class TestS3GateObject(TestS3GateBase): @allure.title("Test S3 Sync directory") @pytest.mark.parametrize("sync_type", ["sync", "cp"]) def test_s3_sync_dir(self, sync_type): - file_path_1 = f"{os.getcwd()}/{ASSETS_DIR}/test_sync/test_file_1" - file_path_2 = f"{os.getcwd()}/{ASSETS_DIR}/test_sync/test_file_2" + file_path_1 = os.path.join(os.getcwd(), ASSETS_DIR, "test_sync", "test_file_1") + file_path_2 = os.path.join(os.getcwd(), ASSETS_DIR, "test_sync", "test_file_2") object_metadata = {f"{uuid.uuid4()}": f"{uuid.uuid4()}"} key_to_path = {"test_file_1": file_path_1, "test_file_2": file_path_2} diff --git a/robot/resources/lib/python_keywords/acl.py b/robot/resources/lib/python_keywords/acl.py index 1b97690f..d3e87708 100644 --- a/robot/resources/lib/python_keywords/acl.py +++ b/robot/resources/lib/python_keywords/acl.py @@ -148,7 +148,7 @@ def _encode_cid_for_eacl(cid: str) -> str: def create_eacl(cid: str, rules_list: List[EACLRule], shell: Shell) -> str: - table_file_path = f"{os.getcwd()}/{ASSETS_DIR}/eacl_table_{str(uuid.uuid4())}.json" + table_file_path = os.path.join(os.getcwd(), ASSETS_DIR, f"eacl_table_{str(uuid.uuid4())}.json") cli = NeofsCli(shell, NEOFS_CLI_EXEC, WALLET_CONFIG) cli.acl.extended_create(cid=cid, out=table_file_path, rule=rules_list) @@ -168,7 +168,7 @@ def form_bearertoken_file( with bearer token and writes to file """ enc_cid = _encode_cid_for_eacl(cid) - file_path = f"{os.getcwd()}/{ASSETS_DIR}/{str(uuid.uuid4())}" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) eacl = get_eacl(wif, cid, shell=shell) json_eacl = dict() diff --git a/robot/resources/lib/python_keywords/http_gate.py b/robot/resources/lib/python_keywords/http_gate.py index 66014174..fa462eb8 100644 --- a/robot/resources/lib/python_keywords/http_gate.py +++ b/robot/resources/lib/python_keywords/http_gate.py @@ -1,5 +1,3 @@ -#!/usr/bin/python3 - import logging import os import re @@ -39,10 +37,10 @@ def get_via_http_gate(cid: str, oid: str): logger.info(f"Request: {request}") _attach_allure_step(request, resp.status_code) - filename = f"{ASSETS_DIR}/{cid}_{oid}" - with open(filename, "wb") as get_file: - shutil.copyfileobj(resp.raw, get_file) - return filename + file_path = os.path.join(os.getcwd(), ASSETS_DIR, f"{cid}_{oid}") + with open(file_path, "wb") as file: + shutil.copyfileobj(resp.raw, file) + return file_path @allure.step("Get via Zip HTTP Gate") @@ -66,14 +64,14 @@ def get_via_zip_http_gate(cid: str, prefix: str): logger.info(f"Request: {request}") _attach_allure_step(request, resp.status_code) - filename = f"{ASSETS_DIR}/{cid}_archive.zip" - with open(filename, "wb") as get_file: - shutil.copyfileobj(resp.raw, get_file) + file_path = os.path.join(os.getcwd(), ASSETS_DIR, f"{cid}_archive.zip") + with open(file_path, "wb") as file: + shutil.copyfileobj(resp.raw, file) - with zipfile.ZipFile(filename, "r") as zip_ref: + with zipfile.ZipFile(file_path, "r") as zip_ref: zip_ref.extractall(ASSETS_DIR) - return f"{ASSETS_DIR}/{prefix}" + return os.path.join(os.getcwd(), ASSETS_DIR, prefix) @allure.step("Get via HTTP Gate by attribute") @@ -99,10 +97,10 @@ def get_via_http_gate_by_attribute(cid: str, attribute: dict): logger.info(f"Request: {request}") _attach_allure_step(request, resp.status_code) - filename = f"{ASSETS_DIR}/{cid}_{str(uuid.uuid4())}" - with open(filename, "wb") as get_file: - shutil.copyfileobj(resp.raw, get_file) - return filename + file_path = os.path.join(os.getcwd(), ASSETS_DIR, f"{cid}_{str(uuid.uuid4())}") + with open(file_path, "wb") as file: + shutil.copyfileobj(resp.raw, file) + return file_path @allure.step("Upload via HTTP Gate") @@ -165,12 +163,12 @@ def get_via_http_curl(cid: str, oid: str) -> str: :param oid: object OID """ request = f"{HTTP_GATE}/get/{cid}/{oid}" - filename = f"{ASSETS_DIR}/{cid}_{oid}_{str(uuid.uuid4())}" - cmd = f"curl {request} > {filename}" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, f"{cid}_{oid}_{str(uuid.uuid4())}") + cmd = f"curl {request} > {file_path}" _cmd_run(cmd) - return filename + return file_path def _attach_allure_step(request: str, status_code: int, req_type="GET"): diff --git a/robot/resources/lib/python_keywords/neofs_verbs.py b/robot/resources/lib/python_keywords/neofs_verbs.py index 21a6df08..0a8ad2c9 100644 --- a/robot/resources/lib/python_keywords/neofs_verbs.py +++ b/robot/resources/lib/python_keywords/neofs_verbs.py @@ -1,11 +1,6 @@ -#!/usr/bin/python3 - -""" - This module contains wrappers for NeoFS verbs executed via neofs-cli. -""" - import json import logging +import os import random import re import uuid @@ -55,7 +50,7 @@ def get_object( if not write_object: write_object = str(uuid.uuid4()) - file_path = f"{ASSETS_DIR}/{write_object}" + file_path = os.path.join(ASSETS_DIR, write_object) if not endpoint: endpoint = random.sample(NEOFS_NETMAP, 1)[0] @@ -254,7 +249,7 @@ def get_range( Returns: (str, bytes) - path to the file with range content and content of this file as bytes """ - range_file = f"{ASSETS_DIR}/{uuid.uuid4()}" + range_file_path = os.path.join(ASSETS_DIR, str(uuid.uuid4())) cli = NeofsCli(shell, NEOFS_CLI_EXEC, wallet_config or WALLET_CONFIG) cli.object.range( @@ -263,15 +258,15 @@ def get_range( cid=cid, oid=oid, range=range_cut, - file=range_file, + file=range_file_path, bearer=bearer, xhdr=xhdr, session=session, ) - with open(range_file, "rb") as fout: - content = fout.read() - return range_file, content + with open(range_file_path, "rb") as file: + content = file.read() + return range_file_path, content @allure.step("Search object") diff --git a/robot/resources/lib/python_keywords/session_token.py b/robot/resources/lib/python_keywords/session_token.py index 9a0fd40f..b85afd6f 100644 --- a/robot/resources/lib/python_keywords/session_token.py +++ b/robot/resources/lib/python_keywords/session_token.py @@ -30,7 +30,7 @@ def generate_session_token(owner: str, session_wallet: str, cid: str = "") -> st Returns: (str): the path to the generated session token file """ - file_path = f"{os.getcwd()}/{ASSETS_DIR}/{uuid.uuid4()}" + file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) session_wlt_content = "" with open(session_wallet) as fout: @@ -80,7 +80,7 @@ def create_session_token(owner: str, wallet_path: str, rpc: str = NEOFS_ENDPOINT Returns: (str): the path to the generated session token file """ - session_token = f"{os.getcwd()}/{ASSETS_DIR}/{uuid.uuid4()}" + session_token = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) cmd = ( f"{NEOFS_CLI_EXEC} session create --address {owner} -w {wallet_path} " f"--out {session_token} --rpc-endpoint {rpc}" @@ -99,7 +99,7 @@ def sign_session_token(session_token: str, wlt: str): Returns: (str): the path to the signed token """ - signed_token = f"{os.getcwd()}/{ASSETS_DIR}/{uuid.uuid4()}" + signed_token = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4())) cmd = ( f"{NEOFS_CLI_EXEC} util sign session-token --from {session_token} " f"-w {wlt} --to {signed_token} --config {WALLET_CONFIG}"