Clean code

Author: Potterluo

test/common/allure_utils.py

@@ -3,18 +3,17 @@
 Provides convenient Allure reporting functionality and decorators
 """
 
-import allure
 import os
-import pytest
-import subprocess
-import shutil
-import time
 import platform
+import shutil
+import subprocess
 import sys
+import time
 from pathlib import Path
-from typing import Dict, Any, ContextManager, Optional, Union, List
-
+from typing import Any, ContextManager, Dict, List, Optional, Union
 
+import allure
+import pytest
 
 
 def setup_allure(config: Dict[str, Any]) -> Optional[Path]:
@@ -26,8 +25,10 @@ def setup_allure(config: Dict[str, Any]) -> Optional[Path]:
     base_dir = Path(config.get("base_dir", "reports"))
     if config.get("use_timestamp", False) and base_dir.exists():
         timestamp_dirs = [
-            d for d in base_dir.iterdir()
-            if d.is_dir() and d.name.startswith(config.get("directory_prefix", "pytest"))
+            d
+            for d in base_dir.iterdir()
+            if d.is_dir()
+            and d.name.startswith(config.get("directory_prefix", "pytest"))
         ]
         if timestamp_dirs:
             timestamp_dirs.sort(key=lambda x: x.stat().st_mtime, reverse=True)
@@ -37,8 +38,8 @@ def setup_allure(config: Dict[str, Any]) -> Optional[Path]:
     allure_dir.mkdir(parents=True, exist_ok=True)
     os.environ["ALLURE_REPORT_DIR"] = str(allure_dir)
 
-    env_info = _get_system_info()        
-    custom_env = allure_cfg.get("environment", {})  
+    env_info = _get_system_info()
+    custom_env = allure_cfg.get("environment", {})
     env_info.update(custom_env)
     _create_environment_properties(allure_dir, env_info)
 
@@ -56,7 +57,7 @@ def check_allure_available() -> bool:
             capture_output=True,
             text=True,
             timeout=10,
-            shell=True
+            shell=True,
         )
         return result.returncode == 0
     except Exception:
@@ -67,11 +68,13 @@ def serve_allure_report(
     allure_results_dir: Union[str, Path],
     host: str = "localhost",
     port: int = 8080,
-    auto_open: bool = True
+    auto_open: bool = True,
 ) -> Optional[subprocess.Popen]:
     """Start Allure server and optionally open browser."""
     if not check_allure_available():
-        print("Allure CLI not found. Install from https://github.com/allure-framework/allure2/releases")
+        print(
+            "Allure CLI not found. Install from https://github.com/allure-framework/allure2/releases"
+        )
         return None
 
     allure_results_dir = Path(allure_results_dir)
@@ -90,7 +93,7 @@ def serve_allure_report(
         stderr=subprocess.STDOUT,
         text=True,
         bufsize=1,
-        universal_newlines=True
+        universal_newlines=True,
     )
     print(f"Allure server starting at http://{host}:{port} (PID: {process.pid})")
     print("Please press Ctrl+C to stop the server")
@@ -118,11 +121,13 @@ def generate_allure_html(
     allure_results_dir: Union[str, Path],
     html_output_dir: Optional[Union[str, Path]] = None,
     clean: bool = False,
-    auto_serve: bool = False
+    auto_serve: bool = False,
 ) -> Optional[Union[Path, subprocess.Popen]]:
     """Generate static HTML report or serve dynamically."""
     if not check_allure_available():
-        print("Allure CLI not found. Install from https://github.com/allure-framework/allure2/releases")
+        print(
+            "Allure CLI not found. Install from https://github.com/allure-framework/allure2/releases"
+        )
         return None
 
     allure_results_dir = Path(allure_results_dir)
@@ -133,13 +138,17 @@ def generate_allure_html(
     if auto_serve:
         return serve_allure_report(allure_results_dir)
 
-    html_output_dir = Path(html_output_dir or allure_results_dir.parent / "allure-report")
+    html_output_dir = Path(
+        html_output_dir or allure_results_dir.parent / "allure-report"
+    )
     if clean and html_output_dir.exists():
         shutil.rmtree(html_output_dir)
     html_output_dir.mkdir(parents=True, exist_ok=True)
 
     allure_path = shutil.which("allure")
-    cmd = f'{allure_path} generate "{allure_results_dir}" -o "{html_output_dir}" --clean'
+    cmd = (
+        f'{allure_path} generate "{allure_results_dir}" -o "{html_output_dir}" --clean'
+    )
     result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
 
     if result.returncode == 0:
@@ -150,17 +159,18 @@ def generate_allure_html(
         return None
 
 
-def _create_environment_properties(allure_results_dir: Union[str, Path], 
-                                  environment_info: Dict[str, str]) -> None:
+def _create_environment_properties(
+    allure_results_dir: Union[str, Path], environment_info: Dict[str, str]
+) -> None:
     allure_results_dir = Path(allure_results_dir)
     allure_results_dir.mkdir(parents=True, exist_ok=True)
-    
+
     env_file = allure_results_dir / "environment.properties"
-    
-    with open(env_file, 'w', encoding='utf-8') as f:
+
+    with open(env_file, "w", encoding="utf-8") as f:
         for key, value in environment_info.items():
             f.write(f"{key}={value}\n")
-    
+
     print(f"Environment properties file created: {env_file}")
 
 
@@ -173,7 +183,7 @@ def _get_system_info() -> Dict[str, str]:
     info["OS"] = os_name
 
     # ---------- Architecture ----------
-    arch = platform.architecture()[0]      # '64bit' / '32bit'
+    arch = platform.architecture()[0]  # '64bit' / '32bit'
     info["Architecture"] = "64-bit" if "64" in arch else "32-bit"
 
     # ---------- Python ----------
@@ -191,4 +201,4 @@ def _get_system_info() -> Dict[str, str]:
     else:
         info["Processor"] = proc.split()[0] if proc else "Kunpeng"
 
-    return info
+    return info

test/common/config_utils.py

@@ -1,7 +1,8 @@
-import yaml
 import os
 import threading
-from typing import Dict, Any
+from typing import Any, Dict
+
+import yaml
 
 
 class ConfigUtils:
@@ -77,4 +78,6 @@ def get_nested_config(self, key_path: str, default: Any = None) -> Any:
 
 if __name__ == "__main__":
     print("InfluxDB config:", config_utils.get_config("influxdb"))
-    print("InfluxDB host:", config_utils.get_nested_config("influxdb.host", "localhost"))
+    print(
+        "InfluxDB host:", config_utils.get_nested_config("influxdb.host", "localhost")
+    )