fix(notifications): simplify notification usage

notification system now takes a filter to determine if the element should be returned. by default
elements from the callback is returned.

atomacos/AXCallbacks.py

@@ -17,43 +17,19 @@
 import fnmatch
 
 
-def elemDisappearedCallback(retelem, obj, **kwargs):
-    """Callback for checking if a UI element is no longer onscreen.
-
-    kwargs should contains some unique set of identifier (e.g. title/value, role)
-    Returns:  Boolean
-    """
-    return not obj.findFirstR(**kwargs)
-
-
-def returnElemCallback(retelem):
-    """Callback for when a sheet appears.
-
-    Returns: element returned by observer callback
-    """
-    return retelem
-
-
-def match(obj, **kwargs):
-    """Method which indicates if the object matches specified criteria.
-
-    Match accepts criteria as kwargs and looks them up on attributes.
-    Actual matching is performed with fnmatch, so shell-like wildcards
-    work within match strings. Examples:
-
-    match(obj, AXTitle='Terminal*')
-    match(obj, AXRole='TextField', AXRoleDescription='search text field')
-    """
-    for k in kwargs.keys():
-        try:
-            val = getattr(obj, k)
-        except AttributeError:
-            return False
-        # Not all values may be strings (e.g. size, position)
-        if isinstance(val, str):
-            if not fnmatch.fnmatch(val, kwargs[k]):
-                return False
-        else:
-            if val != kwargs[k]:
+def match_filter(**kwargs):
+    def _match(obj):
+        for k in kwargs.keys():
+            try:
+                val = getattr(obj, k)
+            except AttributeError:
                 return False
-    return True
+            if isinstance(val, str):
+                if not fnmatch.fnmatch(val, kwargs[k]):
+                    return False
+            else:
+                if val != kwargs[k]:
+                    return False
+        return True
+
+    return _match

atomacos/mixin/_search.py

@@ -18,9 +18,10 @@ def _generateChildren(self, target=None, recursive=False):
 
     def _findAll(self, recursive=False, **kwargs):
         """Return a list of all children that match the specified criteria."""
-        for needle in self._generateChildren(recursive=recursive):
-            if AXCallbacks.match(needle, **kwargs):
-                yield needle
+        return filter(
+            AXCallbacks.match_filter(**kwargs),
+            self._generateChildren(recursive=recursive),
+        )
 
     def _findFirst(self, recursive=False, **kwargs):
         """Return the first object that matches the criteria."""

atomacos/mixin/_wait.py

@@ -3,57 +3,6 @@
 
 
 class WaitForMixin(object):
-    def _waitFor(self, timeout, notification, **kwargs):
-        """Wait for a particular UI event to occur; this can be built
-        upon in NativeUIElement for specific convenience methods.
-        """
-        callback = AXCallbacks.match
-        retelem = None
-        callbackArgs = None
-        callbackKwargs = None
-
-        # Allow customization of the callback, though by default use the basic
-        # _match() method
-        if "callback" in kwargs:
-            callback = kwargs["callback"]
-            del kwargs["callback"]
-
-            # Deal with these only if callback is provided:
-            if "args" in kwargs:
-                if not isinstance(kwargs["args"], tuple):
-                    errStr = "Notification callback args not given as a tuple"
-                    raise TypeError(errStr)
-
-                # If args are given, notification will pass back the returned
-                # element in the first positional arg
-                callbackArgs = kwargs["args"]
-                del kwargs["args"]
-
-            if "kwargs" in kwargs:
-                if not isinstance(kwargs["kwargs"], dict):
-                    errStr = "Notification callback kwargs not given as a dict"
-                    raise TypeError(errStr)
-
-                callbackKwargs = kwargs["kwargs"]
-                del kwargs["kwargs"]
-            # If kwargs are not given as a dictionary but individually listed
-            # need to update the callbackKwargs dict with the remaining items in
-            # kwargs
-            if kwargs:
-                if callbackKwargs:
-                    callbackKwargs.update(kwargs)
-                else:
-                    callbackKwargs = kwargs
-        else:
-            if retelem:
-                callbackArgs = (retelem,)
-            # Pass the kwargs to the default callback
-            callbackKwargs = kwargs
-
-        return Observer(self).set_notification(
-            timeout, notification, callback, callbackArgs, callbackKwargs
-        )
-
     def waitFor(self, timeout, notification, **kwargs):
         """Generic wait for a UI event that matches the specified
         criteria to occur.
@@ -66,18 +15,18 @@ def waitFor(self, timeout, notification, **kwargs):
         destroyed, callback should not use it, otherwise the function will
         hang.
         """
-        return self._waitFor(timeout, notification, **kwargs)
+        return Observer(self).wait_for(
+            filter_=AXCallbacks.match_filter(**kwargs),
+            notification=notification,
+            timeout=timeout,
+        )
 
     def waitForCreation(self, timeout=10, notification="AXCreated"):
         """Convenience method to wait for creation of some UI element.
 
         Returns: The element created
         """
-        callback = AXCallbacks.returnElemCallback
-        retelem = None
-        args = (retelem,)
-
-        return self.waitFor(timeout, notification, callback=callback, args=args)
+        return self.waitFor(timeout, notification)
 
     def waitForWindowToAppear(self, winName, timeout=10):
         """Convenience method to wait for a window with the given name to
@@ -93,20 +42,11 @@ def waitForWindowToDisappear(self, winName, timeout=10):
 
         Returns: Boolean
         """
-        callback = AXCallbacks.elemDisappearedCallback
-        retelem = None
-        args = (retelem, self)
-
         # For some reason for the AXUIElementDestroyed notification to fire,
         # we need to have a reference to it first
         win = self.findFirst(AXRole="AXWindow", AXTitle=winName)  # noqa: F841
         return self.waitFor(
-            timeout,
-            "AXUIElementDestroyed",
-            callback=callback,
-            args=args,
-            AXRole="AXWindow",
-            AXTitle=winName,
+            timeout, "AXUIElementDestroyed", AXRole="AXWindow", AXTitle=winName
         )
 
     def waitForSheetToAppear(self, timeout=10):
@@ -130,11 +70,7 @@ def waitForValueToChange(self, timeout=10):
         # object's.  Unique identifiers considered include role and position
         # This seems to work best if you set the notification at the application
         # level
-        callback = AXCallbacks.returnElemCallback
-        retelem = None
-        return self.waitFor(
-            timeout, "AXValueChanged", callback=callback, args=(retelem,)
-        )
+        return self.waitFor(timeout, "AXValueChanged")
 
     def waitForFocusToChange(self, newFocusedElem, timeout=10):
         """Convenience method to wait for focused element to change (to new
@@ -163,15 +99,4 @@ def waitForFocusToMatchCriteria(self, timeout=10, **kwargs):
         Returns: Element or None
 
         """
-
-        def _matchFocused(retelem, **kwargs):
-            return retelem if retelem._match(**kwargs) else None
-
-        retelem = None
-        return self._waitFor(
-            timeout,
-            "AXFocusedUIElementChanged",
-            callback=_matchFocused,
-            args=(retelem,),
-            **kwargs
-        )
+        return self.waitFor(timeout, "AXFocusedUIElementChanged", **kwargs)

atomacos/notification.py

@@ -36,47 +36,23 @@ def __init__(self, uielement=None):
         self.callback = None
         self.callback_result = None
 
-    def set_notification(
-        self,
-        timeout=0,
-        notification_name=None,
-        callbackFn=None,
-        callbackArgs=None,
-        callbackKwargs=None,
-    ):
-        if callable(callbackFn):
-            self.callbackFn = callbackFn
-
-        if isinstance(callbackArgs, tuple):
-            self.callbackArgs = callbackArgs
-        else:
-            self.callbackArgs = tuple()
-
-        if isinstance(callbackKwargs, dict):
-            self.callbackKwargs = callbackKwargs
-        else:
-            self.callbackKwargs = dict()
-
+    def wait_for(self, notification=None, filter_=None, timeout=5):
         self.callback_result = None
 
         @PAXObserverCallback
         def _callback(observer, element, notification, refcon):
-            if self.callbackFn is not None:
-                ret_element = self.ref.__class__(element)
-                if ret_element is None:
-                    raise RuntimeError("Could not create new AX UI Element.")
-                callback_args = (ret_element,) + self.callbackArgs
-                callback_result = self.callbackFn(*callback_args, **self.callbackKwargs)
-                if callback_result is None:
-                    raise RuntimeError("Python callback failed.")
-                if callback_result in (-1, 1):
-                    self.callback_result = callback_result
+            logger.debug("CALLBACK")
+            logger.debug("%s, %s, %s, %s" % (observer, element, notification, refcon))
+            ret_element = self.ref.__class__(element)
+            if filter_(ret_element):
+                self.callback_result = ret_element
 
         observer = PAXObserverCreate(self.ref.pid, _callback)
 
         PAXObserverAddNotification(
-            observer, self.ref.ref, notification_name, id(self.ref.ref)
+            observer, self.ref.ref, notification, id(self.ref.ref)
         )
+
         # Add observer source to run loop
         CFRunLoopAddSource(
             CFRunLoopGetCurrent(),
@@ -100,6 +76,6 @@ def event_stopper():
         AppHelper.runConsoleEventLoop()
         MachSignals.signal(signal.SIGINT, oldSigIntHandler)
 
-        PAXObserverRemoveNotification(observer, self.ref.ref, notification_name)
+        PAXObserverRemoveNotification(observer, self.ref.ref, notification)
 
         return self.callback_result

tests/test_atomacos.py

@@ -136,7 +136,7 @@ def test_observer_init(self, front_title_ui):
         notification.Observer(front_title_ui)
 
     @pytest.mark.slow
-    def test_observer_set_notification(self, monkeypatch, finder_app):
+    def test_observer_wait_for(self, monkeypatch, finder_app):
         import threading
         from ApplicationServices import kAXWindowCreatedNotification
 
@@ -148,10 +148,10 @@ def open_new_window():
         new_window.start()
 
         observer = notification.Observer(finder_app)
-        result = observer.set_notification(
+        result = observer.wait_for(
             timeout=10,
-            notification_name=kAXWindowCreatedNotification,
-            callbackFn=lambda *_, **__: -1,
+            notification=kAXWindowCreatedNotification,
+            filter_=lambda _: True,
         )
 
-        assert result == -1
+        assert isinstance(result, NativeUIElement)

tests/test_wait_for_methods.py

@@ -18,7 +18,7 @@ def open_new_window():
 def test_waitfor_notification_names(monkeypatch):
     sut = atomacos.NativeUIElement()
     monkeypatch.setattr(
-        sut, "_waitFor", lambda timeout, notification, **kwargs: notification
+        sut, "waitFor", lambda timeout, notification, **kwargs: notification
     )
 
     assert sut.waitForCreation() == "AXCreated"