Static Type Checking: Enums

doc/source/hacking/using_the_testsuite.rst

@@ -190,13 +190,22 @@ consists of running the ``pylint`` tool, run the following::
 
 .. _contributing_formatting_code:
 
+Running Static Type Checkers
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Static Type Checking is performed separately from testing. In order to run the static type checking step which
+consists of running the ``mypy`` tool, run the following::
+
+  tox -e mypy
+
 Formatting code
 ~~~~~~~~~~~~~~~
 Similar to linting, code formatting is also done via a ``tox`` environment. To
 format the code using the ``black`` tool, run the following::
 
    tox -e format
 
+In CI `tox -e format-check` is used to ensure formatting has been run.
+
 Observing coverage
 ~~~~~~~~~~~~~~~~~~
 Once you have run the tests using `tox` (or `detox`), some coverage reports will

src/buildstream/_elementproxy.py

@@ -95,7 +95,7 @@ def stage_artifact(
         sandbox: "Sandbox",
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True
@@ -120,7 +120,7 @@ def stage_dependency_artifacts(
         selection: Optional[Sequence["Element"]] = None,
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True
@@ -168,7 +168,7 @@ def _stage_artifact(
         sandbox: "Sandbox",
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True,

src/buildstream/_overlapcollector.py

@@ -66,7 +66,7 @@ def __init__(self, element: "Element"):
     #    location (str): The Sandbox relative location this session was created for
     #
     @contextmanager
-    def session(self, action: str, location: Optional[str]):
+    def session(self, action: OverlapAction, location: Optional[str]):
         assert self._session is None, "Stage session already started"
 
         if location is None:
@@ -108,13 +108,13 @@ def collect_stage_result(self, element: "Element", result: FileListResult):
 #    location (str): The Sandbox relative location this session was created for
 #
 class OverlapCollectorSession:
-    def __init__(self, element: "Element", action: str, location: str):
+    def __init__(self, element: "Element", action: OverlapAction, location: str):
 
         # The Element we are staging for, on which we'll issue warnings
         self._element = element  # type: Element
 
         # The OverlapAction for this session
-        self._action = action  # type: str
+        self._action = action  # type: OverlapAction
 
         # The Sandbox relative directory this session was created for
         self._location = location  # type: str

src/buildstream/_pipeline.py

@@ -44,7 +44,7 @@
 # Yields:
 #    Elements in the scope of the specified target elements
 #
-def dependencies(targets: List[Element], scope: int, *, recurse: bool = True) -> Iterator[Element]:
+def dependencies(targets: List[Element], scope: _Scope, *, recurse: bool = True) -> Iterator[Element]:
     # Keep track of 'visited' in this scope, so that all targets
     # share the same context.
     visited = (BitMap(), BitMap())
@@ -73,7 +73,12 @@ def dependencies(targets: List[Element], scope: int, *, recurse: bool = True) ->
 #    A list of Elements appropriate for the specified selection mode
 #
 def get_selection(
-    context: Context, targets: List[Element], mode: str, *, silent: bool = True, depth_sort: bool = False
+    context: Context,
+    targets: List[Element],
+    mode: _PipelineSelection,
+    *,
+    silent: bool = True,
+    depth_sort: bool = False
 ) -> List[Element]:
     def redirect_and_log() -> List[Element]:
         # Redirect and log if permitted

src/buildstream/_stream.py

@@ -153,7 +153,7 @@ def load_selection(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         except_targets: Iterable[str] = (),
         load_artifacts: bool = False,
         connect_artifact_cache: bool = False,
@@ -259,7 +259,7 @@ def query_cache(self, elements, *, sources_of_cached_elements=False, only_source
     def shell(
         self,
         target: str,
-        scope: int,
+        scope: _Scope,
         prompt: Callable[[Element], str],
         *,
         unique_id: Optional[str] = None,
@@ -382,7 +382,7 @@ def build(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         ignore_junction_targets: bool = False,
         artifact_remotes: Iterable[RemoteSpec] = (),
         source_remotes: Iterable[RemoteSpec] = (),
@@ -456,7 +456,7 @@ def fetch(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         except_targets: Iterable[str] = (),
         source_remotes: Iterable[RemoteSpec] = (),
         ignore_project_source_remotes: bool = False,
@@ -579,7 +579,7 @@ def pull(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         ignore_junction_targets: bool = False,
         artifact_remotes: Iterable[RemoteSpec] = (),
         ignore_project_artifact_remotes: bool = False,
@@ -633,7 +633,7 @@ def push(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         ignore_junction_targets: bool = False,
         artifact_remotes: Iterable[RemoteSpec] = (),
         ignore_project_artifact_remotes: bool = False,
@@ -688,7 +688,7 @@ def checkout(
         *,
         location: Optional[str] = None,
         force: bool = False,
-        selection: str = _PipelineSelection.RUN,
+        selection: _PipelineSelection = _PipelineSelection.RUN,
         integrate: bool = True,
         hardlinks: bool = False,
         compression: str = "",
@@ -1668,7 +1668,7 @@ def _load(
         self,
         targets: Iterable[str],
         *,
-        selection: str = _PipelineSelection.NONE,
+        selection: _PipelineSelection = _PipelineSelection.NONE,
         except_targets: Iterable[str] = (),
         ignore_junction_targets: bool = False,
         dynamic_plan: bool = False,

src/buildstream/element.py

@@ -90,7 +90,7 @@
 from .sandbox import _SandboxFlags, SandboxCommandError
 from .sandbox._config import SandboxConfig
 from .sandbox._sandboxremote import SandboxRemote
-from .types import _Scope, _CacheBuildTrees, _KeyStrength, OverlapAction, _DisplayKey
+from .types import _HostMount, _Scope, _CacheBuildTrees, _KeyStrength, OverlapAction, _DisplayKey
 from ._artifact import Artifact
 from ._elementproxy import ElementProxy
 from ._elementsources import ElementSources
@@ -604,7 +604,7 @@ def stage_artifact(
         sandbox: "Sandbox",
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True,
@@ -664,7 +664,7 @@ def stage_dependency_artifacts(
         selection: Optional[Sequence["Element"]] = None,
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True,
@@ -864,7 +864,7 @@ def subsandbox(self, sandbox: "Sandbox") -> Iterator["Sandbox"]:
     # Yields:
     #    (Element): The dependencies in `scope`, in deterministic staging order
     #
-    def _dependencies(self, scope, *, recurse=True, visited=None):
+    def _dependencies(self, scope: _Scope, *, recurse=True, visited=None):
 
         # The format of visited is (BitMap(), BitMap()), with the first BitMap
         # containing element that have been visited for the `_Scope.BUILD` case
@@ -971,7 +971,7 @@ def _stage_artifact(
         sandbox: "Sandbox",
         *,
         path: Optional[str] = None,
-        action: str = OverlapAction.WARNING,
+        action: OverlapAction = OverlapAction.WARNING,
         include: Optional[List[str]] = None,
         exclude: Optional[List[str]] = None,
         orphans: bool = True,
@@ -2060,7 +2060,16 @@ def _push(self):
     #    usebuildtree (bool): Use the buildtree as its source
     #
     # Returns: Exit code
-    def _shell(self, scope=None, *, mounts=None, isolate=False, prompt=None, command=None, usebuildtree=False):
+    def _shell(
+        self,
+        scope: _Scope | None = None,
+        *,
+        mounts: List[_HostMount] | None = None,
+        isolate: bool = False,
+        prompt: str | None = None,
+        command: List[str] | None = None,
+        usebuildtree: bool = False,
+    ):
 
         with self._prepare_sandbox(scope, shell=True, usebuildtree=usebuildtree) as sandbox:
             environment = sandbox._get_configured_environment() or self.get_environment()

src/buildstream/types.py

@@ -35,6 +35,9 @@ class FastEnum(metaclass=MetaFastEnum):
 
     :class:`enum.Enum` attributes accesses can be really slow, and slow down the execution noticeably.
     This reimplementation doesn't suffer the same problems, but also does not reimplement everything.
+
+    For mypy Enum static type checking support, all FastEnum should be stubbed as inheriting from enum.Enum.
+    Use `stubgen src/buildstream/types.py --include-docstrings` to generate the stubs, add `from enum import Enum` and replace all `(FastEnum)` with `(Enum)`
     """
 
     name = None
@@ -61,7 +64,7 @@ def __new__(cls, value):
         try:
             return cls._value_to_entry[value]
         except KeyError:
-            if type(value) is cls:  # pylint: disable=unidiomatic-typecheck
+            if isinstance(value, cls):  # pylint: disable=unidiomatic-typecheck
                 return value
             raise ValueError("Unknown enum value: {}".format(value))
 
@@ -169,7 +172,6 @@ class OverlapAction(FastEnum):
 # Defines the scope of dependencies to include for a given element
 # when iterating over the dependency graph in APIs like
 # Element._dependencies().
-#
 class _Scope(FastEnum):
 
     # All elements which the given element depends on, following
@@ -384,7 +386,7 @@ def new_from_node(cls, node: MappingNode) -> "_SourceMirror":
         alias_node: MappingNode = node.get_mapping("aliases")
 
         for alias, uris in alias_node.items():
-            assert type(uris) is SequenceNode  # pylint: disable=unidiomatic-typecheck
+            assert isinstance(uris, SequenceNode)  # pylint: disable=unidiomatic-typecheck
             aliases[alias] = uris.as_str_list()
 
         return cls(name, aliases)