Repository: NucciTheBoss/cleantest (language: python)

Issue:
Enable parallel tests
Another important feature to add is the support for parallel tests. Running tests sequentially using LXD will take forever as the amount of testlets for each project grows.

This should be easily accomplished using the `multiprocessing` module but extra care will need to be taken to ensure that there is no data racing in the tests. Also, the amount of threads should be controllable via an environment variable and or with the test provider decorators.
Enable parallel tests
Another important feature to add is the support for parallel tests. Running tests sequentially using LXD will take forever as the amount of testlets for each project grows.

This should be easily accomplished using the `multiprocessing` module but extra care will need to be taken to ensure that there is no data racing in the tests. Also, the amount of threads should be controllable via an environment variable and or with the test provider decorators.
Enable parallel tests
Another important feature to add is the support for parallel tests. Running tests sequentially using LXD will take forever as the amount of testlets for each project grows.

This should be easily accomplished using the `multiprocessing` module but extra care will need to be taken to ensure that there is no data racing in the tests. Also, the amount of threads should be controllable via an environment variable and or with the test provider decorators.


Discussion / hints:
Removed from 0.2.0 milestone due to time constraints and presentations
Removed from 0.2.0 milestone due to time constraints and presentations
Removed from 0.2.0 milestone due to time constraints and presentations

Fix patch:
diff --git a/tox.ini b/tox.ini
index 08c843c..a79f073 100644
--- a/tox.ini
+++ b/tox.ini
@@ -87,6 +87,12 @@ commands =
 
 [testenv:lxd-local-parallel]
 description = Run functional tests in parallel for a locally-hosted LXD cluster.
+deps =
+  -r {toxinidir}/requirements.txt
+  pytest
+  pylxd
+commands =
+  pytest -v --tb native --log-cli-level=ERROR {[vars]tst_path}/lxd/parallel
 
 [testenv:lxd-remote]
 description = Run functional tests for a remotely-hosted LXD cluster.


Test patch:
diff --git a/src/cleantest/control/configurator.py b/src/cleantest/control/configurator.py
index 6f3f02a..af41dbc 100644
--- a/src/cleantest/control/configurator.py
+++ b/src/cleantest/control/configurator.py
@@ -30,7 +30,7 @@ class Configure:
     __hook_registry = HookRegistry()
     __metadata = set()
 
-    def __new__(cls) -> None:
+    def __new__(cls) -> "Configure":
         if not hasattr(cls, "instance"):
             cls.instance = super(Configure, cls).__new__(cls)
         return cls.instance
diff --git a/src/cleantest/pkg/handler/__init__.py b/src/cleantest/pkg/handler/__init__.py
index c751043..5fb0081 100644
--- a/src/cleantest/pkg/handler/__init__.py
+++ b/src/cleantest/pkg/handler/__init__.py
@@ -2,4 +2,4 @@
 # Copyright 2022 Jason C. Nucciarone, Canonical Ltd.
 # See LICENSE file for licensing details.
 
-import snap_handler as snap
+import cleantest.pkg.handler.snap_handler as snap
diff --git a/src/cleantest/provider/_handler/__init__.py b/src/cleantest/provider/_handler/__init__.py
index 9250f8d..0f6a42f 100644
--- a/src/cleantest/provider/_handler/__init__.py
+++ b/src/cleantest/provider/_handler/__init__.py
@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
+# Copyright 2022 Jason C. Nucciarone, Canonical Ltd.
 # See LICENSE file for licensing details.
 
-from .lxd_handler import LXDHandler
+from .base_handler import Result
+from .lxd_handler import LXDProvider
diff --git a/src/cleantest/provider/_handler/base_handler.py b/src/cleantest/provider/_handler/base_handler.py
index 78b9700..d209b6c 100644
--- a/src/cleantest/provider/_handler/base_handler.py
+++ b/src/cleantest/provider/_handler/base_handler.py
@@ -14,9 +14,8 @@
 import tempfile
 import textwrap
 from abc import ABC, abstractmethod
-from typing import Any, Callable, List
-
-from pydantic import BaseModel
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, List
 
 import cleantest
 
@@ -25,19 +24,30 @@ class HandlerError(Exception):
     ...
 
 
-class Result(BaseModel):
-    exit_code: int | None = None
-    stdout: Any | None = None
-    stderr: Any | None = None
+@dataclass
+class Result:
+    exit_code: int = None
+    stdout: Any = None
+    stderr: Any = None
 
 
-class Handler(ABC):
-    """Abstract super-class for all test environment providers."""
+class Entrypoint(ABC):
+    """Abstract super-class for test environment provider entrypoints.
+
+    Entrypoints define the tooling needed by cleantest to start a test.
+    """
 
     @abstractmethod
-    def run(self) -> Result:
+    def run(self) -> Dict[str, Result]:
         ...
 
+
+class Handler(ABC):
+    """Abstract super-class for test environment handlers.
+
+    Handlers define all the tooling needed to run tests inside the remote environment.
+    """
+
     @abstractmethod
     def _init(self) -> None:
         ...
@@ -84,11 +94,11 @@ def _construct_cleantest_injection(self, path: str) -> str:
             )
         )
 
-    def _construct_testlet(self, func: Callable, remove: List[re.Pattern] | None) -> str:
+    def _construct_testlet(self, src: str, name: str, remove: List[re.Pattern] | None) -> str:
         """Construct Python source file to be run in subroutine.
 
         Args:
-            func (Callable): TODO
+            src (str): TODO
             remove (List[re.Pattern]): TODO
 
         Returns:
@@ -98,7 +108,6 @@ def _construct_testlet(self, func: Callable, remove: List[re.Pattern] | None) ->
             This will need more advanced logic if tests accept arguments.
         """
         try:
-            src = inspect.getsource(func)
             if remove is not None:
                 for pattern in remove:
                     src = re.sub(pattern, "", src)
@@ -107,7 +116,7 @@ def _construct_testlet(self, func: Callable, remove: List[re.Pattern] | None) ->
                 content = [
                     "#!/usr/bin/env python3\n",
                     f"{src}\n",
-                    f"{func.__name__}()\n",
+                    f"{name}()\n",
                 ]
                 f.writelines(content)
                 f.seek(0)
@@ -115,7 +124,7 @@ def _construct_testlet(self, func: Callable, remove: List[re.Pattern] | None) ->
 
             return scriptlet
         except OSError:
-            raise HandlerError(f"Could not locate source code for testlet {func.__name__}.")
+            raise HandlerError(f"Could not locate source code for testlet {name}.")
 
     def _construct_pkg_installer(self, pkg: object, file_path: str, hash: str) -> str:
         src = inspect.getsourcefile(pkg.__class__)
diff --git a/src/cleantest/provider/_handler/lxd_handler.py b/src/cleantest/provider/_handler/lxd_handler.py
index 5de8221..7645925 100644
--- a/src/cleantest/provider/_handler/lxd_handler.py
+++ b/src/cleantest/provider/_handler/lxd_handler.py
@@ -1,79 +1,57 @@
 #!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
+# Copyright 2022 Jason C. Nucciarone, Canonical Ltd.
 # See LICENSE file for licensing details.
 
 """Handler for LXD-based test environments."""
 
+import inspect
 import json
 import os
 import re
+import sys
+from concurrent.futures import ProcessPoolExecutor
+from dataclasses import dataclass
 from typing import Any, Callable, Dict, List
 
-from pydantic import BaseModel
+from pylxd import Client
 
 from cleantest.pkg._base import Package
-from cleantest.provider._handler.base_handler import Handler, Result
+from cleantest.provider._handler.base_handler import Entrypoint, Handler, Result
 from cleantest.provider.data.lxd_data import LXDConfig
 
 
-class Instance(BaseModel):
+@dataclass
+class Instance:
     name: str
     image: str
     exists: bool = False
 
 
 class LXDHandler(Handler):
-    def __init__(self, attr: Dict[str, Any], func: Callable, threaded: bool) -> None:
-        [setattr(self, k, v) for k, v in attr.items()]
-        self.func = func
-        self.threaded = threaded
-
-    @classmethod
-    def serial(cls, lxd, func: Callable) -> object:
-        return cls(lxd.__dict__, func, False)
-
-    @classmethod
-    def parallel(cls, lxd, func: Callable) -> object:
-        return cls(lxd.__dict__, func, True)
-
-    def run(self) -> Result:
-        instances = []
-        for i in self._image:
-            instances.append(Instance(name=f"{self._name}-{i}", image=i))
-
-        self._build(self._check_exists(instances))
-        self._handle_start_env_hooks(instances)
-        result = self._execute(
-            self._construct_testlet(self.func, [re.compile("@lxd(.*)")]), instances
-        )
-        if self._preserve is False:
-            self._teardown(instances)
-
-        return self._process(result)
-
-    def _check_exists(self, instances: List[Instance]) -> List[Instance]:
-        exists_status = []
-        for i in instances:
-            if self._client.instances.exists(i.name):
-                i.exists = True
-                exists_status.append(i)
-            else:
-                exists_status.append(i)
-
-        return exists_status
-
-    def _build(self, instances: List[Instance]) -> None:
-        for i in instances:
-            if i.exists is False:
-                config = self._data.get_config(i.image)
-                config.name = i.name
-                self._client.instances.create(config.dict(), wait=True)
-                instance = self._client.instances.get(i.name)
-                instance.start(wait=True)
-                self._init(instance, config)
-            else:
-                if (tmp := self._client.instances.get(i.name)).status.lower() == "stopped":
-                    tmp.start(wait=True)
+    def _set_client(self) -> None:
+        if self._client_config is None:
+            self._client = Client(project="default")
+        else:
+            self._client = Client(
+                endpoint=self._client_config.endpoint,
+                version=self._client_config.version,
+                cert=self._client_config.cert,
+                verify=self._client_config.verify,
+                timeout=self._client_config.timeout,
+                project=self._client_config.project,
+            )
+
+    def _build(self, instance: Instance) -> None:
+        if instance.exists is False:
+            config = self._data.get_config(instance.image)
+            config.name = instance.name
+            self._client.instances.create(config.dict(), wait=True)
+            instance = self._client.instances.get(instance.name)
+            instance.start(wait=True)
+            self._init(instance, config)
+        else:
+            if self._client.instances.get(instance.name).status.lower() == "stopped":
+                self._client.instances.get(instance.name).start(wait=True)
 
     def _init(self, instance: Any, config: LXDConfig) -> None:
         if "ubuntu" in config.source.alias:
@@ -89,33 +67,27 @@ def _init(self, instance: Any, config: LXDConfig) -> None:
         else:
             NotImplementedError(f"{config.source.alias} injection not supported yet.")
 
-    def _execute(self, test: str, instances: List[Instance]) -> Any:
-        for i in instances:
-            instance = self._client.instances.get(i.name)
-            instance.files.put("/root/test", test)
-            instance.execute(["chmod", "+x", "/root/test"])
-            return instance.execute(["/root/test"], environment=self._env.dump())
-
-    def _process(self, result: Any) -> Result:
-        return Result(exit_code=result.exit_code, stdout=result.stdout, stderr=result.stderr)
-
-    def _teardown(self, instances: List[Instance]) -> None:
-        for i in instances:
-            instance = self._client.instances.get(i.name)
-            instance.stop(wait=True)
-            instance.delete(wait=True)
-
-    def _handle_start_env_hooks(self, instances: List[Instance]) -> None:
-        startenvhooks = self._cleanconfig.get_start_env_hooks()
-        while len(startenvhooks) > 0:
-            hook = startenvhooks.pop()
-            for i in instances:
-                instance = self._client.instances.get(i.name)
-                if hook.packages is not None:
-                    for pkg in hook.packages:
-                        self.__handle_package_install(instance, pkg)
-
-    def __handle_package_install(self, instance: Any, pkg: Package) -> None:
+    def _execute(self, test: str, instance: Instance) -> Any:
+        instance = self._client.instances.get(instance.name)
+        instance.files.put("/root/test", test)
+        instance.execute(["chmod", "+x", "/root/test"])
+        return instance.execute(["/root/test"], environment=self._env.dump())
+
+    def _teardown(self, instance: Instance) -> None:
+        instance = self._client.instances.get(instance.name)
+        instance.stop(wait=True)
+        instance.delete(wait=True)
+
+    def _handle_start_env_hooks(self, instance: Instance) -> None:
+        start_env_hooks = self._clean_config.get_start_env_hooks()
+        while len(start_env_hooks) > 0:
+            hook = start_env_hooks.pop()
+            instance = self._client.instances.get(instance.name)
+            if hook.packages is not None:
+                for pkg in hook.packages:
+                    self._handle_package_install(instance, pkg)
+
+    def _handle_package_install(self, instance: Any, pkg: Package) -> None:
         dispatch = {"charmlib": lambda x: self._env.add(json.loads(x))}
 
         dump_data = pkg._dump()
@@ -128,5 +100,84 @@ def __handle_package_install(self, instance: Any, pkg: Package) -> None:
         instance.execute(["chmod", "+x", "/root/install"])
         result = instance.execute(["/root/install"])
 
-        if (tmp := pkg.__class__.__name__.lower()) in dispatch:
-            dispatch[tmp](result.stdout)
+        if pkg.__class__.__name__.lower() in dispatch:
+            dispatch[pkg.__class__.__name__.lower()](result.stdout)
+
+    def _process(self, result: Any) -> Result:
+        return Result(exit_code=result.exit_code, stdout=result.stdout, stderr=result.stderr)
+
+    def _check_exists(self, instance: Instance) -> Instance:
+        if self._client.instances.exists(instance.name):
+            instance.exists = True
+
+        return instance
+
+    def _construct_instance_metaclasses(self) -> List[Instance]:
+        return [Instance(name=f"{self._name}-{i}", image=i) for i in self._image]
+
+
+class Serial(Entrypoint, LXDHandler):
+    def __init__(self, attr: Dict[str, Any], func: Callable) -> None:
+        [setattr(self, k, v) for k, v in attr.items()]
+        self._func = inspect.getsource(func)
+        self._func_name = func.__name__
+
+    def run(self) -> Dict[str, Result]:
+        results = {}
+        self._set_client()
+        for i in self._construct_instance_metaclasses():
+            self._build(self._check_exists(i))
+            self._handle_start_env_hooks(i)
+            result = self._execute(
+                self._construct_testlet(
+                    self._func, self._func_name, [re.compile(r"^@lxd\(([^)]+)\)")]
+                ),
+                i,
+            )
+            if self._preserve is False:
+                self._teardown(i)
+            results.update({i.name: self._process(result)})
+
+        return results
+
+
+class Parallel(Entrypoint, LXDHandler):
+    def __init__(self, attr: Dict[str, Any], func: Callable) -> None:
+        [setattr(self, k, v) for k, v in attr.items()]
+        print(dir(func), file=sys.stderr)
+        self._func = inspect.getsource(func)
+        self._func_name = func.__name__
+
+    def run(self) -> Dict[str, Result]:
+        results = {}
+        with ProcessPoolExecutor(
+            max_workers=self._num_threads,
+        ) as pool:
+            pool_results = pool.map(self._target, self._construct_instance_metaclasses())
+            for res in pool_results:
+                [results.update({key: value}) for key, value in res.items()]
+
+        return results
+
+    def _target(self, i: Instance) -> Dict[str, Result]:
+        self._set_client()
+        self._build(self._check_exists(i))
+        self._handle_start_env_hooks(i)
+        result = self._execute(
+            self._construct_testlet(self._func, self._func_name, [re.compile(r"^@lxd\(([^)]+)\)")]),
+            i,
+        )
+        if self._preserve is False:
+            self._teardown(i)
+
+        return {i.name: self._process(result)}
+
+
+class LXDProvider:
+    @staticmethod
+    def serial(lxd, func: Callable) -> "Serial":
+        return Serial(lxd.__dict__, func)
+
+    @staticmethod
+    def parallel(lxd, func: Callable) -> "Parallel":
+        return Parallel(lxd.__dict__, func)
diff --git a/src/cleantest/provider/lxd.py b/src/cleantest/provider/lxd.py
index e262740..6b84154 100644
--- a/src/cleantest/provider/lxd.py
+++ b/src/cleantest/provider/lxd.py
@@ -1,28 +1,28 @@
 #!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
+# Copyright 2022 Jason C. Nucciarone, Canonical Ltd.
 # See LICENSE file for licensing details.
 
 """LXD test environment provider functions and utilities."""
 
 from __future__ import annotations
 
-from types import NoneType
+import os
+from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Tuple
 
-from pydantic import BaseModel
-from pylxd import Client
-
 from cleantest.control.configurator import Configure
-from cleantest.provider._handler import LXDHandler
 from cleantest.provider.data import EnvDataStore, LXDDataStore
 
+from ._handler import LXDProvider, Result
+
 
-class LXDClientConfig(BaseModel):
-    endpoint: Any | None = None
+@dataclass
+class LXDClientConfig:
+    endpoint: Any = None
     version: str = "1.0"
-    cert: Tuple[str, str] | None = None
+    cert: Tuple[str, str] = None
     verify: bool | str = True
-    timeout: float | Tuple[float, float] | None = None
+    timeout: float | Tuple[float, float] = None
     project: str = "default"
 
 
@@ -34,13 +34,18 @@ def __init__(
         preserve: bool = True,
         env: EnvDataStore = EnvDataStore(),
         data: LXDDataStore = LXDDataStore(),
-        image_config: Dict[str, Any] | List[Dict[str, Any]] | None = None,
-        client_config: LXDClientConfig | None = None,
+        image_config: Dict[str, Any] | List[Dict[str, Any]] = None,
+        client_config: LXDClientConfig = None,
+        parallel: bool = False,
+        num_threads: int = None,
     ) -> None:
         self._name = name
         self._preserve = preserve
         self._env = env
         self._data = data
+        self._parallel = parallel
+        self._client_config = client_config
+        self._clean_config = Configure()
 
         if type(image) == str:
             self._image = [image]
@@ -53,23 +58,21 @@ def __init__(
             for c in image_config:
                 self._data.add_config(c)
 
-        if type(client_config) == NoneType:
-            self._client = Client(project="default")
-        else:
-            self._client = Client(
-                endpoint=client_config.endpoint,
-                version=client_config.version,
-                cert=client_config.cert,
-                verify=client_config.verify,
-                timeout=client_config.timeout,
-                project=client_config.project,
+        if (type(num_threads) != int or num_threads < 1) and self._parallel is True:
+            env_var = os.getenv("CLEANTEST_NUM_THREADS")
+            self._num_threads = (
+                env_var if env_var is not None and type(env_var) == int else os.cpu_count()
             )
-
-        self._cleanconfig = Configure()
+        elif type(num_threads) == int and self._parallel is True:
+            self._num_threads = num_threads
 
     def __call__(self, func: Callable) -> Callable:
-        def wrapper(*args, **kwargs) -> None:
-            handler = LXDHandler.serial(self, func)
+        def wrapper(*args, **kwargs) -> Dict[str, Result]:
+            handler = (
+                LXDProvider.parallel(self, func)
+                if self._parallel is True
+                else LXDProvider.serial(self, func)
+            )
             return handler.run()
 
         return wrapper
diff --git a/src/cleantest/utils/__init__.py b/src/cleantest/utils/__init__.py
index 3dbedc6..eec62b0 100644
--- a/src/cleantest/utils/__init__.py
+++ b/src/cleantest/utils/__init__.py
@@ -2,4 +2,4 @@
 # Copyright 2022 Canonical Ltd.
 # See LICENSE file for licensing details.
 
-from .detect_os import detect_os_variant
+from cleantest.utils.detect_os import detect_os_variant
diff --git a/tests/lxd/local/test_lxd_local.py b/tests/lxd/local/test_lxd_local.py
index 93d83f9..a131b90 100644
--- a/tests/lxd/local/test_lxd_local.py
+++ b/tests/lxd/local/test_lxd_local.py
@@ -26,6 +26,7 @@
 )
 cleanconfig.register_hook(startenvhook)
 
+
 # Define the testlets.
 @lxd(image="jammy-amd64", preserve=False)
 def install_snapd():
@@ -65,5 +66,6 @@ def install_snapd():
 # Run the tests through pytest.
 class TestLocalLXD:
     def test_local_lxd(self) -> None:
-        result = install_snapd()
-        assert result.exit_code == 0
+        results = install_snapd()
+        for name, result in results.items():
+            assert result.exit_code == 0
diff --git a/tests/lxd/parallel/test_lxd_parallel.py b/tests/lxd/parallel/test_lxd_parallel.py
index bf98b47..56accec 100644
--- a/tests/lxd/parallel/test_lxd_parallel.py
+++ b/tests/lxd/parallel/test_lxd_parallel.py
@@ -1,3 +1,41 @@
 #!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
+# Copyright 2022 Jason C. Nucciarone, Canonical Ltd.
 # See LICENSE file for licensing details.
+
+"""Test parallel testing capabilites of LXD provider using local LXD cluster."""
+
+from cleantest import Configure
+from cleantest.hooks import StartEnvHook
+from cleantest.pkg import Pip
+from cleantest.provider import lxd
+
+cleantest_config = Configure()
+start_hook = StartEnvHook(name="pip_injection", packages=[Pip(packages="tabulate")])
+cleantest_config.register_hook(start_hook)
+
+
+@lxd(
+    image=["jammy-amd64", "focal-amd64", "bionic-amd64"],
+    preserve=True,
+    parallel=True,
+    num_threads=2,
+)
+def install_tabulate():
+    import sys
+
+    try:
+        from tabulate import tabulate
+
+        print("tabulate is installed.", file=sys.stdout)
+    except ImportError:
+        print("Failed to import tabulate package.", file=sys.stderr)
+        sys.exit(1)
+
+    sys.exit(0)
+
+
+class TestParallelLXD:
+    def test_parallel_lxd(self) -> None:
+        results = install_tabulate()
+        for name, result in results.items():
+            assert result.exit_code == 0
diff --git a/tests/packages/snap/test_snap.py b/tests/packages/snap/test_snap.py
index 1f4619f..48eb778 100644
--- a/tests/packages/snap/test_snap.py
+++ b/tests/packages/snap/test_snap.py
@@ -43,5 +43,6 @@ def functional_snaps():
 
 class TestLocalLXD:
     def test_snap_package(self) -> None:
-        result = functional_snaps()
-        assert result.exit_code == 0
+        results = functional_snaps()
+        for name, result in results.items():
+            assert result.exit_code == 0