Feature/refactoring (#65)

* refactoring

.github/workflows/python_lint.yml

@@ -0,0 +1,24 @@
+name: Python lint
+
+on:
+  - pull_request
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.x'
+
+      - name: Install Ruff
+        run: |
+          python -m pip install --upgrade pip
+          pip install ruff
+
+      - name: Run Ruff
+        run: |
+          ruff check .

frameworks/VboxMachine/VboxMachine.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+from VBoxWrapper import VirtualMachine
+
+from frameworks.decorators import vm_is_turn_on
+from .vm_data import VmData
+from .configs import VmConfig
+
+
+class VboxMachine:
+
+    def __init__(self, name: str, config_path: str = None):
+        self.vm_config = VmConfig(config_path=config_path)
+        self.vm = VirtualMachine(name)
+        self.name = name
+        self.data = None
+
+    @vm_is_turn_on
+    def create_data(self):
+        self.data = VmData(
+            ip=self.vm.network.get_ip(),
+            user=self.vm.get_logged_user(),
+            name=self.name,
+            local_dir=self.vm.get_parameter('CfgFile')
+        )
+
+    def run(self, headless: bool = True, status_bar: bool = False, timeout: int = 600):
+        if self.vm.power_status():
+            self.vm.stop()
+
+        self.vm.snapshot.restore()
+        self.configurate()
+        self.vm.run(headless=headless)
+        self.vm.network.wait_up(status_bar=status_bar, timeout=timeout)
+        self.vm.wait_logged_user(status_bar=status_bar, timeout=timeout)
+        self.create_data()
+
+    def configurate(self):
+        self.vm.set_cpus(self.vm_config.cpus)
+        self.vm.nested_virtualization(self.vm_config.nested_virtualization)
+        self.vm.set_memory(self.vm_config.memory)
+        self.vm.audio(self.vm_config.audio)
+        self.vm.speculative_execution_control(self.vm_config.speculative_execution_control)
+
+    def stop(self):
+        self.vm.stop()

frameworks/VboxMachine/__init__.py

@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+from .VboxMachine import VboxMachine
+from .configs import VmConfig
+
+__all__ = [VboxMachine, VmConfig]

frameworks/VboxMachine/configs/__init__.py

@@ -0,0 +1,4 @@
+# -*- coding: utf-8 -*-
+from .vm_config import VmConfig
+
+__all__ = [VmConfig]

frameworks/VboxMachine/configs/vm_config.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+import json
+from rich import print
+from pydantic import BaseModel, conint
+from host_tools import singleton
+
+
+class SystemConfigModel(BaseModel):
+    """
+    A Pydantic model for validating the system configuration parameters.
+
+    Attributes:
+        cpus (int): The number of CPUs allocated for the system. Must be an integer >= 1.
+        memory (int): The amount of memory in MB. Must be an integer >= 512.
+    """
+
+    cpus: conint(ge=1)
+    memory: conint(ge=512)
+    audio: bool
+    nested_virtualization: bool
+    speculative_execution_control: bool
+
+
+@singleton
+class VmConfig:
+    """
+    Configuration class for system settings.
+
+    Attributes:
+        cpus (int): The number of CPUs allocated for the system.
+        memory (int): The amount of memory in MB.
+    """
+    def __init__(self, config_path: str):
+        self.config_path = config_path
+        self._config = self._load_config(self.config_path)
+        self.cpus = self._config.cpus
+        self.memory = self._config.memory
+        self.audio = self._config.audio
+        self.nested_virtualization = self._config.nested_virtualization
+        self.speculative_execution_control = self._config.speculative_execution_control
+
+    @staticmethod
+    def _load_config(file_path: str) -> SystemConfigModel:
+        """
+        Loads the system configuration from a JSON file and returns a SystemConfigModel instance.
+
+        :param file_path: The path to the configuration JSON file.
+        :return: An instance of SystemConfigModel containing the loaded configuration.
+        """
+        with open(file_path, 'r') as f:
+            return SystemConfigModel(**json.load(f))
+
+    def display_config(self):
+        """
+        Displays the loaded system configuration.
+        """
+        print(
+            f"[green]|INFO| System Configuration:\n"
+            f"  CPUs: {self.cpus}\n"
+            f"  Memory: {self.memory}MB\n"
+            f"  Audio Enabled: {self.audio}\n"
+            f"  Nested Virtualization: {self.nested_virtualization}\n"
+            f"  Speculative Execution Control: {self.speculative_execution_control}"
+        )

frameworks/VboxMachine/vm_data.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+from dataclasses import dataclass
+
+
+@dataclass
+class VmData:
+    user: str
+    ip: str
+    name: str
+    local_dir: str

frameworks/__init__.py

@@ -0,0 +1,11 @@
+# -*- coding: utf-8 -*-
+from .VboxMachine import VboxMachine, VmConfig
+from .console import MyConsole
+from .report import Report
+
+__all__ = [
+    VboxMachine,
+    MyConsole,
+    Report,
+    VmConfig
+]

frameworks/decorators/__init__.py

@@ -1,2 +1,4 @@
 # -*- coding: utf-8 -*-
-from .decorators import *
+from .decorators import vm_data_created, vm_is_turn_on, retry
+
+__all__ = ['vm_data_created', 'vm_is_turn_on', 'retry']

frameworks/decorators/decorators.py

@@ -1,22 +1,32 @@
 # -*- coding: utf-8 -*-
 from functools import wraps
-from multiprocessing import Process
-from time import perf_counter, sleep
+from time import sleep
 
+from VBoxWrapper import VirtualMachinException
 from rich import print
 
 
-def singleton(class_):
-    __instances = {}
+def vm_data_created(method):
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
 
-    @wraps(class_)
-    def getinstance(*args, **kwargs):
-        if class_ not in __instances:
-            __instances[class_] = class_(*args, **kwargs)
-        return __instances[class_]
+        if self.vm.data is None:
+            raise VirtualMachinException("Vm data has not been created, Please start the VM before creating data.")
 
-    return getinstance
+        return method(self, *args, **kwargs)
 
+    return wrapper
+
+def vm_is_turn_on(method):
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
+
+        if not self.vm.power_status():
+            raise VirtualMachinException("Virtual machine is not turned on. Please start the VM before creating data.")
+
+        return method(self, *args, **kwargs)
+
+    return wrapper
 
 def retry(
         max_attempts: int = 3,
@@ -44,56 +54,3 @@ def inner(*args, **kwargs):
         return inner
 
     return wrapper
-
-
-def highlighter(color: str = 'green'):
-    def wrapper(func):
-        @wraps(func)
-        def inner(*args, **kwargs):
-            print(f'[{color}]-' * 90)
-            result = func(*args, **kwargs)
-            print(f'[{color}]-' * 90)
-            return result
-
-        return inner
-
-    return wrapper
-
-
-def async_processing(target):
-    def wrapper(func):
-        @wraps(func)
-        def inner(*args, **kwargs):
-            process = Process(target=target)
-            process.start()
-            result = func(*args, **kwargs)
-            process.terminate()
-            return result
-
-        return inner
-
-    return wrapper
-
-
-def timer(func):
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        start_time = perf_counter()
-        result = func(*args, **kwargs)
-        print(f"[green]|INFO| Time existing the function `{func.__name__}`: {(perf_counter() - start_time):.02f}")
-        return result
-
-    return wrapper
-
-
-def memoize(func):
-    __cache = {}
-
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        key = (args, tuple(kwargs.items()))
-        if key not in __cache:
-            __cache[key] = func(*args, **kwargs)
-        return __cache[key]
-
-    return wrapper

frameworks/report/__init__.py

@@ -1,2 +1,4 @@
 # -*- coding: utf-8 -*-
 from .report import Report
+
+__all__ = [Report]

frameworks/report/report.py

@@ -25,23 +25,24 @@ def value_count(df: pd.DataFrame, column_name: str) -> str:
 
     def insert_column(self, path: str, location: str, column_name: str, value: str, delimiter='\t') -> pd.DataFrame:
         df = self.read(path, delimiter=delimiter)
-        if column_name  not in df.columns:
+        if column_name not in df.columns:
             df.insert(loc=df.columns.get_loc(location), column=column_name, value=value)
         else:
-            print(f"[green]|INFO| Column `{column_name}` already exists in `{path}`")
+            print(f"[cyan]|INFO| Column `{column_name}` already exists in `{path}`")
         return df
 
     def merge(self, reports: list, result_csv_path: str, delimiter='\t') -> str | None:
-        if reports:
-            merge_reports = []
-            for csv_ in reports:
-                if isfile(csv_):
-                    report = self.read(csv_, delimiter)
-                    if report is not None:
-                        merge_reports.append(report)
+        merge_reports = [
+            self.read(csv_, delimiter)
+            for csv_ in reports
+            if isfile(csv_) and self.read(csv_, delimiter) is not None
+        ]
+
+        if merge_reports:
             df = pd.concat(merge_reports, ignore_index=True)
             df.to_csv(result_csv_path, index=False, sep=delimiter)
             return result_csv_path
+
         print('[green]|INFO| No files to merge')
 
     @staticmethod

pyproject.toml

@@ -17,6 +17,8 @@ host-tools = { git = "https://github.com/l8556/host_tools.git", branch = "master
 telegram = { git = "https://github.com/l8556/TelegramApi.git", branch = "master" }
 VBoxWrapper = { git = "https://github.com/l8556/VBoxWrapper.git", branch = "master" }
 ssh-wrapper = { git = "https://github.com/l8556/SshWrapper.git", branch = "master" }
+ruff = "^0.6.4"
+pydantic = "^2.9.1"
 
 
 [build-system]

tasks.py

@@ -7,23 +7,18 @@
 from rich import print
 
 from VBoxWrapper import VirtualMachine, Vbox
-from tests.data import TestData
-from tests.desktop_tests import DesktopTest
-import tests.multiprocessing as multiprocess
-from frameworks.console import MyConsole
-from tests.tools.desktop_report import DesktopReport
+from tests.desktop_tests.tools.test_data import TestData
+from tests.desktop_tests import DesktopTest, DesktopReport
+import tests.desktop_tests.multiprocessing as multiprocess
 from host_tools import Process, Service
 from elevate import elevate
 
-console = MyConsole().console
-print = console.print
-
 
 @task
 def desktop_test(
         c,
         version=None,
-        update_from=None,
+        update_from_version=None,
         name=None,
         processes=None,
         detailed_telegram=False,
@@ -34,7 +29,7 @@ def desktop_test(
 
     data = TestData(
         version=version if version else Prompt.ask('[red]Please enter version'),
-        update_from=update_from,
+        update_from=update_from_version,
         telegram=detailed_telegram,
         config_path=join(getcwd(), 'custom_config.json') if custom_config else join(getcwd(), 'config.json'),
         custom_config_mode=custom_config
@@ -51,7 +46,6 @@ def desktop_test(
     report.get_full(data.version)
     report.send_to_tg(data.version, data.title, data.tg_token, data.tg_chat_id, data.update_from) if not name else ...
 
-
 @task
 def run_vm(c, name: str = '', headless=False):
     vm = VirtualMachine(Vbox().check_vm_names(name))
@@ -60,7 +54,6 @@ def run_vm(c, name: str = '', headless=False):
     vm.wait_logged_user(status_bar=True)
     return print(f"[green]ip: [red]{vm.network.get_ip()}[/]\nuser: [red]{vm.get_logged_user()}[/]")
 
-
 @task
 def stop_vm(c, name: str = None, group_name: str = None):
     if name:
@@ -78,19 +71,16 @@ def stop_vm(c, name: str = None, group_name: str = None):
                 print(f"[green]|INFO| Shutting down the virtual machine: [red]{vm_info[0]}[/]")
                 virtualmachine.stop()
 
-
 @task
 def vm_list(c, group_name: str = None):
     vm_names = Vbox().vm_list(group_name)
     print(vm_names)
     return vm_names
 
-
 @task
 def out_info(c, name: str = '', full: bool = False):
     print(VirtualMachine(Vbox().check_vm_names(name)).get_info(full=full))
 
-
 @task
 def group_list(c):
     group_names = Vbox().get_group_list()