Merge lp:~jelmer/brz/dirty-tracker into lp:brz/3.2

[Jelmer Vernooij]

Import new version of dirty_tracker from lintian-brush.

Update dirty_tracker.

[Jelmer Vernooij (jelmer)]

review: Approve

=== modified file 'breezy/dirty_tracker.py'
--- breezy/dirty_tracker.py	2020-07-27 22:52:21 +0000
+++ breezy/dirty_tracker.py	2022-01-17 02:05:14 +0000
@@ -15,45 +15,41 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
-"""Track whether a directory structure was touched since last revision.
-"""
-
-from __future__ import absolute_import
-
-# TODO(jelmer): Add support for ignore files
+"""Track whether a particular directory structure is dirty."""
 
 import os
-try:
-    from pyinotify import (
-        WatchManager,
-        IN_CREATE,
-        IN_CLOSE_WRITE,
-        IN_Q_OVERFLOW,
-        IN_DELETE,
-        IN_MOVED_TO,
-        IN_MOVED_FROM,
-        IN_ATTRIB,
-        ProcessEvent,
-        Notifier,
-        Event,
-        )
-except ImportError as e:
-    from .errors import DependencyNotPresent
-    raise DependencyNotPresent(library='pyinotify', error=e)
+from pyinotify import (
+    WatchManager,
+    IN_CREATE,
+    IN_CLOSE_WRITE,
+    IN_Q_OVERFLOW,
+    IN_DELETE,
+    IN_MOVED_TO,
+    IN_MOVED_FROM,
+    IN_ATTRIB,
+    ProcessEvent,
+    Notifier,
+    Event,
+)
+from typing import Set
+from .workingtree import WorkingTree
 
 
 MASK = (
-    IN_CLOSE_WRITE | IN_DELETE | IN_Q_OVERFLOW | IN_MOVED_TO | IN_MOVED_FROM |
-    IN_ATTRIB)
-
-
-class _Process(ProcessEvent):
-
-    def my_init(self):
+    IN_CLOSE_WRITE | IN_DELETE | IN_Q_OVERFLOW | IN_MOVED_TO | IN_MOVED_FROM | IN_ATTRIB
+)
+
+
+class _Process(ProcessEvent):  # type: ignore
+
+    paths: Set[str]
+    created: Set[str]
+
+    def my_init(self) -> None:
         self.paths = set()
         self.created = set()
 
-    def process_default(self, event):
+    def process_default(self, event: Event) -> None:
         path = os.path.join(event.path, event.name)
         if event.mask & IN_CREATE:
             self.created.add(path)
@@ -66,43 +62,45 @@
 class DirtyTracker(object):
     """Track the changes to (part of) a working tree."""
 
-    def __init__(self, tree, subpath='.'):
+    def __init__(self, tree: WorkingTree, subpath: str = ".") -> None:
         self._tree = tree
         self._wm = WatchManager()
         self._process = _Process()
         self._notifier = Notifier(self._wm, self._process)
         self._notifier.coalesce_events(True)
 
-        def check_excluded(p):
-            return tree.is_control_filename(tree.relpath(p))
+        def check_excluded(p: str) -> bool:
+            return tree.is_control_filename(tree.relpath(p))  # type: ignore
+
         self._wdd = self._wm.add_watch(
-            tree.abspath(subpath), MASK, rec=True, auto_add=True,
-            exclude_filter=check_excluded)
+            tree.abspath(subpath),
+            MASK,
+            rec=True,
+            auto_add=True,
+            exclude_filter=check_excluded,
+        )
 
-    def _process_pending(self):
+    def _process_pending(self) -> None:
         if self._notifier.check_events(timeout=0):
             self._notifier.read_events()
         self._notifier.process_events()
 
-    def __del__(self):
-        self._notifier.stop()
-
-    def mark_clean(self):
+    def mark_clean(self) -> None:
         """Mark the subtree as not having any changes."""
         self._process_pending()
         self._process.paths.clear()
         self._process.created.clear()
 
-    def is_dirty(self):
+    def is_dirty(self) -> bool:
         """Check whether there are any changes."""
         self._process_pending()
         return bool(self._process.paths)
 
-    def paths(self):
+    def paths(self) -> Set[str]:
         """Return the paths that have changed."""
         self._process_pending()
         return self._process.paths
 
-    def relpaths(self):
+    def relpaths(self) -> Set[str]:
         """Return the paths relative to the tree root that changed."""
         return set(self._tree.relpath(p) for p in self.paths())