Merge pull request #28 from Pallandos/add-test

Add unit test for Python 3.11 and 3.12

Author: Pallandos

.env

@@ -1,7 +1,7 @@
 # This is the configuration file
 
 # ==== network ====
-NODE_NUMBER=5
+NODE_NUMBER=8
 MAX_PEERS=128
 # WARNING : set this value high (don't know why, but if it's
 # under ~11 , a node will refuse all connections)

.github/workflows/python-tests.yml

@@ -0,0 +1,34 @@
+name: Python application
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        python-version: ['3.11', '3.12']
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install pytest pytest-cov codecov
+          pip install -r requirements.txt
+
+      - name: Run tests with coverage
+        run: |
+          pytest --cov=py --cov-report=xml:tests/coverage.xml --cov-report=term tests/
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v4
+        with:
+          files: ./tests/coverage.xml

.gitignore

@@ -5,4 +5,8 @@ img/
 __pycache__/*
 *.pyc
 
-/py/tests.py
+# tests
+.coverage
+
+# IDE
+.vscode/

README.md

@@ -74,9 +74,9 @@ Usage: ./bitcoin-on-local.sh start|stop|renew|draw|scenario|draw [output_file]
 This tool requires :
 
 - Docker (compose v2)
-- Python (>= 3.12)[^1]
+- Python (>= 3.11)[^1]
 
-[^1]: The tool may work on older versions but has not been tested on it. 
+[^1]: The tool has been tested on Python 3.11 and 3.12
 
 > [!NOTE]
 > Project is beeing tested right now 

py/generate_compose.py

@@ -1,6 +1,7 @@
 # this file generates a docker-compose.yml file based on the provided configuration
 
 import random
+import os
 
 from config import (
     NODE_NUMBER,
@@ -130,6 +131,9 @@ def export_data(all_ports: dict, node_names: list, output_dir: str = 'data'):
         output_dir (str): Subdirectory of /docker to store the .env files. Defaults to 'data'.
     """
 
+    # ensure the output directory exists
+    os.makedirs(f"docker/{output_dir}", exist_ok=True)
+    
     # export node names :
     output_file_names = f"docker/{output_dir}/.env.node_names"
     with open(output_file_names, 'w') as file:

py/network_info/get_info.py

@@ -20,10 +20,4 @@ def get_peer_info(node_name : str):
         return json.loads(result.stdout)
     except Exception as e:
         print(f"Error for {node_name}: {e}")
-        return []
-    
-if __name__ == "__main__":
-    # tests
-    node_1 = "node_1"
-    info = get_peer_info(node_1)
-    print(info)
+        return []

py/network_info/parse.py

@@ -66,10 +66,4 @@ def extract_connections(peers_info):
                 peer_name = peer_addr
             connections.append((peer_name,connection_type))
 
-    return connections
-
-if __name__ == "__main__":
-    #tests 
-    
-    name = _docker_dns("172.20.0.7")
-    print(name)
+    return connections

py/scenario/rpc_caller.py

@@ -60,6 +60,9 @@ def call(self, node: str, method: str, params: list = None) -> Any:
 
             return result.get('result', None)
 
+        except BitcoinRPCError:
+            # Re-raise the BitcoinRPCError to avoid the final except block
+            raise
         except requests.ConnectionError as e:
             raise requests.ConnectionError(f"Connection failed to {node}") from e
         except requests.Timeout as e:

py/scenario/runner.py

@@ -2,60 +2,66 @@
 from .loader import ScenarioLoader
 from .rpc_caller import BitcoinRPC
 from .actions import ActionExecutor
-from typing import Dict, Any, List
+from typing import Dict, Any
+
 
 # Error classes for ScenarioRunner
 class ScenarioRunnerError(Exception):
     """Base class for all scenario runner errors."""
+
     pass
+
+
 class ScenarioNotLoadedError(ScenarioRunnerError):
     """Raised when a scenario is tried to be runned but not loaded."""
+
     def __init__(self):
         super().__init__("No scenario loaded. Please load a scenario first.")
 
+
 class ScenarioRunner:
-    """ScenarioRunner is a class that manages the execution of Bitcoin scenarios.
-    """
-    def __init__(self,
-                 rpc_user : str,
-                 rpc_password : str,
-                 scenarios_dir : str = "./scenarios",
-                 base_port : int = 18443,
-                 ):
-        
+    """ScenarioRunner is a class that manages the execution of Bitcoin scenarios."""
+
+    def __init__(
+        self,
+        rpc_user: str,
+        rpc_password: str,
+        scenarios_dir: str = "./scenarios",
+        base_port: int = 18443,
+    ):
         self.loader = ScenarioLoader(scenarios_dir)
-        self.variables = {} # Store scenario variables
-        
+        self.variables = {}  # Store scenario variables
+
         rpc = BitcoinRPC(rpc_user, rpc_password, base_port)
         self.executor = ActionExecutor(rpc)
-        
+
         self.scenario = None  # Will hold the loaded scenario
-        self.config = None    # Will hold the scenario configuration
-    
-    def load_scenario(self, scenario_name : str):
+        self.config = None  # Will hold the scenario configuration
+
+    def load_scenario(self, scenario_name: str):
         """Load a scenario by its name. Required before running it.
 
         Args:
             scenario_name (str): Name of the scenario to load (without .toml extension).
         """
-        
+
         self.scenario = self.loader.load_scenario(scenario_name)
-        self.config = self.scenario["config"] # so we dont have to load it again
-        
+        self.config = self.scenario["config"]  # so we dont have to load it again
+
         # print infos :
         print("==========================================")
         print(f"Name: {self.scenario['scenario']['name']} \n")
         print(f"Description: {self.scenario['scenario']['description']}\n")
         print(f"Written by: {self.scenario['scenario']['author']}\n")
         print(f"Date: {self.scenario['scenario']['date']}")
         print("==========================================")
-    
+
     def list_scenarios(self):
         """List available scenarios"""
         print("Available scenarios:")
         scenarios = self.loader.list_scenarios()
         print("\n".join(scenarios) if scenarios else "No scenarios found.")
-    
+
     def _substitute_variables(self, params: Any) -> Any:
         """Replace ${var} with actual values"""
         if isinstance(params, str):
@@ -67,52 +73,53 @@ def _substitute_variables(self, params: Any) -> Any:
         elif isinstance(params, list):
             return [self._substitute_variables(item) for item in params]
         return params
-    
+
     # ==== runners ====
-    
-    def _run_step(self, step: Dict[str , Any]) -> None:
-        
+
+    def _run_step(self, step: Dict[str, Any]) -> None:
         # exctract actions details
         # → the scenario is valid so we can assume that the step has the required keys
         action_name = step["name"]
         action = step["action"]
         node = step.get("node", self.config["default_node"])
         args = self._substitute_variables(step.get("args", {}))
-        
+
         # execute the action
         print(f"Running step: {action_name} (on node: {node})")
         result = self.executor.execute(action, node, args)
-        
+
         # == deal with options ==
         if step.get("print", False):
             print(f"Result: \n → {result} \n")
-        if args.get("store_result",""):
+        if args.get("store_result", ""):
             # store the result in the variables dict
             var_name = args.get("store_result")
             self.variables[var_name] = result
             print(f"Stored result in variable: {var_name} = {result}")
-        
+
         # time and wait
         time.sleep(step.get("wait_after", self.config["default_wait"]))
-    
+
     def run_scenario(self) -> None:
         """Run the loaded scenario step by step"""
-        if not hasattr(self, 'scenario'):
+        if self.scenario is None:
             raise ScenarioNotLoadedError()
-        
+
         print(f"[SCENARIO] Running scenario: {self.scenario['scenario']['name']}")
-        
+
         # timeout sleep:
         time.sleep(self.config.get("timeout"))
-        
+
         for i, ststep_name in enumerate(self.scenario["steps"]):
             print(f"\n[SCENARIO] Step {i + 1}/{len(self.scenario['steps'])}")
             step = self.scenario["steps"][ststep_name]
-            
+
             try:
                 self._run_step(step)
             except Exception as e:
-                print(f"[ERROR] An error occurred while running step '{ststep_name}': {e}")
+                print(
+                    f"[ERROR] An error occurred while running step '{ststep_name}': {e}"
+                )
                 raise e
-        
-        print("[SCENARIO] Scenario execution completed.")
+
+        print("[SCENARIO] Scenario execution completed.")

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+pythonpath = py