[refactor] Generate configs by peers

Signed-off-by: alexstroke <111361420+astrokov7@users.noreply.github.com>

client_cli/pytests/common/settings.py

@@ -11,11 +11,9 @@
 
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-ROOT_DIR = os.environ.get("CLIENT_CLI_DIR", BASE_DIR)
-
 PATH_CONFIG_CLIENT_CLI = os.environ["CLIENT_CLI_CONFIG"]
 CLIENT_CLI_PATH = os.environ["CLIENT_CLI_BINARY"]
-PEERS_CONFIGS_PATH = os.path.join(ROOT_DIR, "peers_configs")
+PEERS_CONFIGS_PATH = os.path.join(BASE_DIR, "peers_configs")
 
 PORT_MIN = int(os.getenv("TORII_API_PORT_MIN", "8080"))
 PORT_MAX = int(os.getenv("TORII_API_PORT_MAX", "8083"))

client_cli/pytests/src/client_cli/__init__.py

@@ -3,9 +3,9 @@
 """
 
 from common.settings import PATH_CONFIG_CLIENT_CLI, PORT_MAX, PORT_MIN
-from src.client_cli.client_cli import ClientCli
-from src.client_cli.configuration import Config
-from src.client_cli.iroha import Iroha
+from .client_cli import ClientCli
+from .configuration import Config
+from .iroha import Iroha
 
 config = Config(PORT_MIN, PORT_MAX)
 config.load(PATH_CONFIG_CLIENT_CLI)

client_cli/pytests/src/client_cli/client_cli.py

@@ -3,16 +3,24 @@
 commands for interacting with Iroha blockchain using the Iroha command-line client.
 """
 
+import json
 import shlex
 import subprocess
+import threading
+from json import JSONDecoder
 from pathlib import Path
 from time import monotonic, sleep
 from typing import Callable
 
 import allure  # type: ignore
 
 from common.helpers import extract_hash, read_isi_from_json, write_isi_to_json
-from common.settings import BASE_DIR, CLIENT_CLI_PATH, PATH_CONFIG_CLIENT_CLI, ROOT_DIR
+from common.settings import (
+    BASE_DIR,
+    CLIENT_CLI_PATH,
+    PATH_CONFIG_CLIENT_CLI,
+    PEERS_CONFIGS_PATH,
+)
 from src.client_cli.configuration import Config
 
 
@@ -35,6 +43,9 @@ def __init__(self, config: Config):
         self.stderr = None
         self.transaction_hash = None
         self._timeout = 20
+        self.event_data = {}
+        self.event_data_lock = threading.Lock()
+        self.should_continue_listening = True
 
     def __enter__(self):
         """
@@ -53,6 +64,36 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         """
         self.reset()
 
+    def start_listening_to_events(self, configs):
+        """
+        Initializes listening to events on all peers.
+        """
+        self.transaction_status = {}
+        self.threads = []
+        for config_path in configs:
+            thread = threading.Thread(target=self.listen_to_events, args=(config_path,))
+            self.threads.append(thread)
+            thread.start()
+
+    def listen_to_events(self, config_path):
+        """
+        Listens to the events using the specified configuration file and stores them.
+        """
+        command = [self.BASE_PATH] + ["--config=" + config_path, "events", "pipeline"]
+        with subprocess.Popen(command, stdout=subprocess.PIPE, text=True) as process:
+            while self.should_continue_listening:
+                output = process.stdout.readline()
+                if not output:
+                    break
+                with self.event_data_lock:
+                    if config_path in self.event_data:
+                        self.event_data[config_path] += output
+                    else:
+                        self.event_data[config_path] = output
+
+    def stop_listening(self):
+        self.should_continue_listening = False
+
     def wait_for(self, condition: Callable[[], bool], timeout=None):
         """
         Wait for a certain condition to be met, specified by the expected and actual values.
@@ -280,7 +321,7 @@ def register_trigger(self, account):
         trigger_data = read_isi_from_json(str(json_template_path))
         trigger_data[0]["Register"]["Trigger"]["action"]["authority"] = str(account)
 
-        json_temp_file_path = Path(ROOT_DIR) / "isi_register_trigger.json"
+        json_temp_file_path = Path(CLIENT_CLI_PATH) / "isi_register_trigger.json"
         write_isi_to_json(trigger_data, str(json_temp_file_path))
 
         self._execute_pipe(
@@ -308,7 +349,7 @@ def unregister_asset(self, asset_id):
         asset_data = read_isi_from_json(str(json_template_path))
         asset_data[0]["Unregister"]["Asset"]["object_id"] = str(asset_id)
 
-        json_temp_file_path = Path(ROOT_DIR) / "isi_unregister_asset.json"
+        json_temp_file_path = Path(CLIENT_CLI_PATH) / "isi_unregister_asset.json"
         write_isi_to_json(asset_data, str(json_temp_file_path))
 
         self._execute_pipe(
@@ -336,11 +377,14 @@ def execute(self, command=None):
         :return: The current ClientCli object.
         :rtype: ClientCli
         """
-        self.config.randomise_torii_url()
+        self.config.select_random_peer_config(PEERS_CONFIGS_PATH)
         if command is None:
             command = self.command
         else:
-            command = [self.BASE_PATH] + self.BASE_FLAGS + shlex.split(command)
+            if isinstance(command, str):
+                command = [self.BASE_PATH] + self.BASE_FLAGS + shlex.split(command)
+            elif isinstance(command, list):
+                command = [self.BASE_PATH] + self.BASE_FLAGS + command
 
         if "|" in command:
             pipe_index = command.index("|")
@@ -375,6 +419,7 @@ def _execute_single(self, command):
         """
         Executes a single command.
         """
+        print(" ".join(command) + "\n")
         with subprocess.Popen(
             command,
             stdout=subprocess.PIPE,
@@ -386,6 +431,47 @@ def _execute_single(self, command):
             self.transaction_hash = extract_hash(self.stdout)
             self._attach_allure_reports()
 
+    def wait_for_transaction_commit(self, transaction_hash, timeout=1):
+        """
+        Waits for the transaction with the given hash to be committed in all configs.
+        """
+        condition = lambda: self.is_transaction_committed(transaction_hash)
+        try:
+            self.wait_for(condition, timeout)
+            return True
+        except TimeoutError:
+            return False
+
+    def is_transaction_committed(self, transaction_hash):
+        """
+        Checks if the transaction with the given hash is committed in all configs.
+        """
+        with self.event_data_lock:
+            for config_path, data in self.event_data.items():
+                if not self._check_commit_in_output(transaction_hash, data):
+                    return False
+        return True
+
+    def _check_commit_in_output(self, transaction_hash, output):
+        """
+        Parses the output to check if the transaction with the given hash is committed.
+        """
+        decoder = JSONDecoder()
+        idx = 0
+        try:
+            while idx < len(output):
+                obj, idx_next = decoder.raw_decode(output[idx:])
+                if (
+                    obj.get("Pipeline", {}).get("entity_kind") == "Transaction"
+                    and obj.get("Pipeline", {}).get("status") == "Committed"
+                    and obj.get("Pipeline", {}).get("hash") == transaction_hash
+                ):
+                    return True
+                idx += idx_next
+        except json.JSONDecodeError:
+            return False
+        return False
+
     def _attach_allure_reports(self):
         """
         Attaches stdout and stderr to Allure reports.

client_cli/pytests/src/client_cli/configuration.py

@@ -4,7 +4,6 @@
 
 import tomlkit
 import glob
-import json
 import os
 import random
 from urllib.parse import urlparse
@@ -35,16 +34,20 @@ def load(self, path_config_client_cli):
 
         :param path_config_client_cli: The path to the configuration file.
         :type path_config_client_cli: str
-        :raises IOError: If the file does not exist.
+        :raises IOError: If the file does not exist or is not a file.
+        :raises ValueError: If the configuration file is invalid.
         """
         if not os.path.exists(path_config_client_cli):
             raise IOError(f"No config file found at {path_config_client_cli}")
-
         if not os.path.isfile(path_config_client_cli):
             raise IOError(f"The path is not a file: {path_config_client_cli}")
 
-        with open(path_config_client_cli, "r", encoding="utf-8") as config_file:
-            self._config = tomlkit.load(config_file)
+        try:
+            with open(path_config_client_cli, "r", encoding="utf-8") as config_file:
+                self._config = tomlkit.load(config_file)
+        except Exception as e:
+            raise ValueError(f"Error reading configuration file: {e}")
+
         self.file = path_config_client_cli
 
     def generate_by_peers(self, peers_configs_dir):
@@ -63,20 +66,25 @@ def generate_by_peers(self, peers_configs_dir):
 
         for port in range(self.port_min, self.port_max + 1):
             config_copy = self._config.copy()
-            config_copy["TORII_API_URL"] = f"http://localhost:{port}"
-            file_name = f"config_to_peer_{port}.json"
+            parsed_url = urlparse(config_copy["torii_url"])
+            updated_url = parsed_url._replace(
+                netloc=f"{parsed_url.hostname}:{port}"
+            ).geturl()
+            config_copy["torii_url"] = updated_url
+
+            file_name = f"config_to_peer_{port}.toml"
             file_path = os.path.join(peers_configs_dir, file_name)
             with open(file_path, "w", encoding="utf-8") as config_file:
-                json.dump(config_copy, config_file, indent=4)
+                tomlkit.dump(config_copy, config_file)
 
-    def select_random_peer_config(self):
+    def select_random_peer_config(self, peers_configs_dir):
         """
         Select and load a random configuration file generated by the generate_by_peers method.
         This updates the current configuration to the one chosen.
 
         :return: None
         """
-        peers_configs = glob.glob("path/to/peers/configs/*.json")
+        peers_configs = glob.glob(peers_configs_dir + "/*.toml")
         if not peers_configs:
             raise ValueError(
                 "Peer configuration files not found. First generate them using generate_by_peers."
@@ -86,20 +94,6 @@ def select_random_peer_config(self):
 
         self.load(chosen_config_file)
 
-    def randomise_torii_url(self):
-        """
-        Update Torii URL.
-        Note that in order for update to take effect,
-        `self.env` should be used when executing the client cli.
-
-        :return: None
-        """
-        parsed_url = urlparse(self._config["torii_url"])
-        random_port = random.randint(self.port_min, self.port_max)
-        self._envs["TORII_URL"] = parsed_url._replace(
-            netloc=f"{parsed_url.hostname}:{random_port}"
-        ).geturl()
-
     @property
     def torii_url(self):
         """

client_cli/pytests/test/conftest.py

@@ -14,6 +14,7 @@
     generate_random_string_with_reserved_char,
     generate_random_string_with_whitespace,
     generate_random_string_without_reserved_chars,
+    get_peers_config_files,
     key_with_invalid_character_in_key,
     name_with_uppercase_letter,
     not_existing_name,
@@ -32,6 +33,7 @@ def before_all():
     This fixture generates configurations based on peers and is automatically
     used for every test session."""
     config.generate_by_peers(PEERS_CONFIGS_PATH)
+    client_cli.start_listening_to_events(get_peers_config_files(PEERS_CONFIGS_PATH))
     yield