Add type hints to target

grizzly/common/runner.py

@@ -95,7 +95,9 @@ def __init__(
         if idle_threshold > 0:
             assert idle_delay > 0
             LOG.debug("using idle check, th %d, delay %ds", idle_threshold, idle_delay)
-            self._idle = _IdleChecker(target.is_idle, idle_threshold, idle_delay)
+            self._idle = _IdleChecker(
+                target.monitor.is_idle, idle_threshold, idle_delay
+            )
         else:
             self._idle = None
         assert close_delay > 0
@@ -332,13 +334,13 @@ def run(
                 LOG.debug("relaunch/shutdown limit hit")
                 # ideally all browser tabs should be closed at this point
                 # and the browser should exit on its own
-                # NOTE: this will take the full duration if target.is_idle()
-                # is not implemented
+                # NOTE: this will take the full duration if target.monitor.is_idle()
+                # is unable to detect if the target is idle
                 for close_delay in range(max(int(self._close_delay / 0.5), 1)):
                     if not self._target.monitor.is_healthy():
                         break
                     # wait 3 seconds (6 passes) before attempting idle exit
-                    if close_delay > 5 and self._target.is_idle(10):
+                    if close_delay > 5 and self._target.monitor.is_idle(10):
                         # NOTE: this will always trigger on systems where the
                         # browser does not exit when the last window is closed
                         LOG.debug("target idle")

grizzly/common/test_runner.py

@@ -75,15 +75,15 @@ def test_runner_02(mocker):
         required=serv_files,
     )
     # single run/iteration relaunch (not idle exit)
-    target.is_idle.return_value = False
+    target.monitor.is_idle.return_value = False
     runner = Runner(server, target, relaunch=1)
     assert runner._relaunch == 1
     smap = ServerMap()
     result = runner.run([], smap, testcase)
     assert runner.initial
     assert result.attempted
     assert target.close.call_count == 1
-    assert target.is_idle.call_count > 0
+    assert target.monitor.is_idle.call_count > 0
     assert target.monitor.is_healthy.call_count > 0
     assert result.status == Result.NONE
     assert result.served == serv_files
@@ -93,7 +93,7 @@ def test_runner_02(mocker):
     target.reset_mock()
     testcase.reset_mock()
     # single run/iteration relaunch (idle exit)
-    target.is_idle.return_value = True
+    target.monitor.is_idle.return_value = True
     runner = Runner(server, target, relaunch=1)
     assert runner._relaunch == 1
     result = runner.run([], ServerMap(), testcase)
@@ -122,7 +122,7 @@ def test_runner_02(mocker):
     assert not runner.initial
     assert result.attempted
     assert target.close.call_count == 1
-    assert target.is_idle.call_count == 0
+    assert target.monitor.is_idle.call_count == 0
     assert target.monitor.is_healthy.call_count == 1
     assert result.status == Result.NONE
     assert result.served == serv_files

grizzly/target/assets.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from shutil import copyfile, copytree, move, rmtree
 from tempfile import mkdtemp
+from typing import Any, Dict, List, Optional
 
 __all__ = ("AssetError", "AssetManager")
 __author__ = "Tyson Smith"
@@ -20,30 +21,28 @@ class AssetError(Exception):
 class AssetManager:
     __slots__ = ("assets", "path")
 
-    def __init__(self, base_path=None):
-        self.assets = {}
+    def __init__(self, base_path: Optional[Path] = None) -> None:
+        self.assets: Dict[str, str] = {}
         self.path = Path(mkdtemp(prefix="assets_", dir=base_path))
 
-    def __enter__(self):
+    def __enter__(self) -> "AssetManager":
         return self
 
-    def __exit__(self, *exc):
+    def __exit__(self, *exc: Any) -> None:
         self.cleanup()
 
-    def add(self, asset, path, copy=True):
+    def add(self, asset: str, path: Path, copy: bool = True) -> Path:
         """Add asset to the AssetManager.
 
         Args:
-            asset (str): Name of asset.
-            path (Path): Location on disk.
-            copy (bool): Copy or move the content.
+            asset: Name of asset.
+            path: Location on disk.
+            copy: Copy or move the content.
 
         Returns:
-            str: Path to the asset on the filesystem.
+            Path to the asset on the filesystem.
         """
-        assert isinstance(asset, str)
-        assert isinstance(path, Path)
-        assert self.path, "cleanup() was called"
+        assert asset
         if not path.exists():
             raise OSError(f"'{path}' does not exist")
         dst = self.path / path.name
@@ -66,19 +65,19 @@ def add(self, asset, path, copy=True):
         LOG.debug("%s asset %r to '%s'", "copied" if copy else "moved", asset, dst)
         return dst
 
-    def add_batch(self, assets):
+    def add_batch(self, assets: List[List[str]]) -> None:
         """Add collection of assets to the AssetManager.
 
         Args:
-            assets (list(list(str, str))): List of list that contain asset, path pairs.
+            assets: List of list that contain asset, path pairs.
 
         Returns:
             None
         """
         for asset, path in assets:
             self.add(asset, Path(path))
 
-    def cleanup(self):
+    def cleanup(self) -> None:
         """Remove asset files from filesystem.
 
         Args:
@@ -90,54 +89,54 @@ def cleanup(self):
         if self.path:
             rmtree(self.path, ignore_errors=True)
             self.assets.clear()
-            self.path = None
 
-    def get(self, asset):
+    def get(self, asset: str) -> Optional[Path]:
         """Get path to content on filesystem for given asset.
 
         Args:
-            asset (str): Asset to lookup.
+            asset: Asset to lookup.
 
         Returns:
-            Path: Path to asset content or None if asset does not exist.
+            Path to asset content or None if asset does not exist.
         """
         item = self.assets.get(asset, None)
         return self.path / item if item else None
 
-    def is_empty(self):
+    def is_empty(self) -> bool:
         """Check if AssetManager contains entries.
 
         Args:
             None
 
         Returns:
-            bool: True if AssetManager contains entries else False.
+            True if AssetManager contains entries else False.
         """
         return not self.assets
 
     @classmethod
-    def load(cls, assets, src_path, base_path=None):
+    def load(
+        cls, assets: Dict[str, str], src_path: Path, base_path: Optional[Path] = None
+    ) -> "AssetManager":
         """Load assets from filesystem.
 
         Args:
-            asset (dict): Asset paths on filesystem relative to src_path, keyed on
-                          asset name.
-            src_path (Path): Path to scan for assets.
-            base_path (str): Base path to use to create local storage.
+            asset: Asset paths on filesystem relative to src_path, keyed on asset name.
+            src_path: Path to scan for assets.
+            base_path: Base path to use to create local storage.
 
         Returns:
-            AssetManager: Populated with contents provided by assets argument.
+            AssetManager populated with contents provided by assets argument.
         """
         obj = cls(base_path=base_path)
         for asset, src_name in assets.items():
             obj.add(asset, src_path / src_name)
         return obj
 
-    def remove(self, asset):
+    def remove(self, asset: str) -> None:
         """Remove asset from AssetManager if asset exists.
 
         Args:
-            asset (str): Asset to remove.
+            asset: Asset to remove.
 
         Returns:
             None