Fix code that constructs paths

Signed-off-by: Vladimir Domnich <v.domnich@yadro.com>

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:

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)}")

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:

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

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):

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}
 

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()

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"):

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")

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}"