Make methods DRY and add type annotations to server/connectionpoints.py. (enthought#826)

* Add type hints.

* Split methods.

* Update `test.test_comserver.TestEvents.test`.

Author: junkmd

comtypes/server/connectionpoints.py

@@ -1,14 +1,23 @@
+import functools
 import logging
 from _ctypes import COMError
-from ctypes import pointer
-from typing import TYPE_CHECKING, ClassVar, Dict, List, Tuple, Type
+from ctypes import c_void_p, pointer
+from ctypes.wintypes import DWORD
+from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Tuple, Type
+from typing import Union as _UnionT
 
-from comtypes import COMObject, IUnknown
+from comtypes import GUID, COMObject, IUnknown
 from comtypes.automation import IDispatch
 from comtypes.connectionpoints import IConnectionPoint
 from comtypes.hresult import *
 from comtypes.typeinfo import ITypeInfo, LoadRegTypeLib
 
+if TYPE_CHECKING:
+    from ctypes import _Pointer
+    from typing import ClassVar
+
+    from comtypes import hints  # type: ignore
+
 logger = logging.getLogger(__name__)
 
 __all__ = ["ConnectableObjectMixin"]
@@ -31,7 +40,9 @@ def __init__(
     # per MSDN, all interface methods *must* be implemented, E_NOTIMPL
     # is no allowed return value
 
-    def IConnectionPoint_Advise(self, this, pUnk, pdwCookie):
+    def IConnectionPoint_Advise(
+        self, this: Any, pUnk: IUnknown, pdwCookie: "_Pointer[DWORD]"
+    ) -> "hints.Hresult":
         if not pUnk or not pdwCookie:
             return E_POINTER
         logger.debug("Advise")
@@ -43,21 +54,25 @@ def IConnectionPoint_Advise(self, this, pUnk, pdwCookie):
         self._connections[self._cookie] = ptr
         return S_OK
 
-    def IConnectionPoint_Unadvise(self, this, dwCookie):
+    def IConnectionPoint_Unadvise(self, this: Any, dwCookie: int) -> "hints.Hresult":
         logger.debug("Unadvise %s", dwCookie)
         try:
             del self._connections[dwCookie]
         except KeyError:
             return CONNECT_E_NOCONNECTION
         return S_OK
 
-    def IConnectionPoint_GetConnectionPointContainer(self, this, ppCPC):
+    def IConnectionPoint_GetConnectionPointContainer(
+        self, this: Any, ppCPC: c_void_p
+    ) -> "hints.Hresult":
         return E_NOTIMPL
 
-    def IConnectionPoint_GetConnectionInterface(self, this, pIID):
+    def IConnectionPoint_GetConnectionInterface(
+        self, this: Any, pIID: "_Pointer[GUID]"
+    ) -> "hints.Hresult":
         return E_NOTIMPL
 
-    def _call_sinks(self, name, *args, **kw):
+    def _call_sinks(self, name: str, *args: Any, **kw: Any) -> List[Any]:
         results = []
         logger.debug("_call_sinks(%s, %s, *%s, **%s)", self, name, args, kw)
         # Is it an IDispatch derived interface?  Then, events have to be delivered
@@ -66,61 +81,33 @@ def _call_sinks(self, name, *args, **kw):
             # for better performance, we could cache the dispids.
             dispid = self._typeinfo.GetIDsOfNames(name)[0]
             for key, p in self._connections.items():
-                try:
-                    result = p.Invoke(dispid, *args, **kw)
-                except COMError as details:
-                    if details.hresult == RPC_S_SERVER_UNAVAILABLE:
-                        logger.warning(
-                            "_call_sinks(%s, %s, *%s, **%s) failed; removing connection",
-                            self,
-                            name,
-                            args,
-                            kw,
-                            exc_info=True,
-                        )
-                        try:
-                            del self._connections[key]
-                        except KeyError:
-                            pass  # connection already gone
-                    else:
-                        logger.warning(
-                            "_call_sinks(%s, %s, *%s, **%s)",
-                            self,
-                            name,
-                            args,
-                            kw,
-                            exc_info=True,
-                        )
-                else:
-                    results.append(result)
+                mth = functools.partial(p.Invoke, dispid)  # type: ignore
+                results.extend(self._call_sink(name, key, mth, *args, **kw))
         else:
-            for p in self._connections.values():
-                try:
-                    result = getattr(p, name)(*args, **kw)
-                except COMError as details:
-                    if details.hresult == RPC_S_SERVER_UNAVAILABLE:
-                        logger.warning(
-                            "_call_sinks(%s, %s, *%s, **%s) failed; removing connection",
-                            self,
-                            name,
-                            args,
-                            kw,
-                            exc_info=True,
-                        )
-                        del self._connections[key]
-                    else:
-                        logger.warning(
-                            "_call_sinks(%s, %s, *%s, **%s)",
-                            self,
-                            name,
-                            args,
-                            kw,
-                            exc_info=True,
-                        )
-                else:
-                    results.append(result)
+            for key, p in self._connections.items():
+                mth = getattr(p, name)
+                results.extend(self._call_sink(name, key, mth, *args, **kw))
         return results
 
+    def _call_sink(
+        self, name: str, key: int, mth: Callable[..., Any], *args: Any, **kw: Any
+    ) -> Iterator[Any]:
+        try:
+            result = mth(*args, **kw)
+        except COMError as details:
+            if details.hresult == RPC_S_SERVER_UNAVAILABLE:
+                warn_msg = "_call_sinks(%s, %s, *%s, **%s) failed; removing connection"
+                logger.warning(warn_msg, self, name, args, kw, exc_info=True)
+                try:
+                    del self._connections[key]
+                except KeyError:
+                    pass  # connection already gone
+            else:
+                warn_msg = "_call_sinks(%s, %s, *%s, **%s)"
+                logger.warning(warn_msg, self, name, args, kw, exc_info=True)
+        else:
+            yield result
+
 
 class ConnectableObjectMixin(object):
     """Mixin which implements IConnectionPointContainer.
@@ -143,13 +130,17 @@ def __init__(self) -> None:
             typeinfo = tlib.GetTypeInfoOfGuid(itf._iid_)
             self.__connections[itf] = ConnectionPointImpl(itf, typeinfo)
 
-    def IConnectionPointContainer_EnumConnectionPoints(self, this, ppEnum):
+    def IConnectionPointContainer_EnumConnectionPoints(
+        self, this: Any, ppEnum: c_void_p
+    ) -> "hints.Hresult":
         # according to MSDN, E_NOTIMPL is specificially disallowed
         # because, without typeinfo, there's no way for the caller to
         # find out.
         return E_NOTIMPL
 
-    def IConnectionPointContainer_FindConnectionPoint(self, this, refiid, ppcp):
+    def IConnectionPointContainer_FindConnectionPoint(
+        self, this: Any, refiid: "_Pointer[GUID]", ppcp: c_void_p
+    ) -> "hints.Hresult":
         iid = refiid[0]
         logger.debug("FindConnectionPoint %s", iid)
         if not ppcp:
@@ -169,7 +160,9 @@ def IConnectionPointContainer_FindConnectionPoint(self, this, refiid, ppcp):
         logger.debug("No connectionpoint found")
         return CONNECT_E_NOCONNECTION
 
-    def Fire_Event(self, itf, name, *args, **kw):
+    def Fire_Event(
+        self, itf: _UnionT[int, Type[IDispatch]], name: str, *args: Any, **kw: Any
+    ) -> Any:
         # Fire event 'name' with arguments *args and **kw.
         # Accepts either an interface index or an interface as first argument.
         # Returns a list of results.

comtypes/test/test_comserver.py

@@ -297,7 +297,10 @@ def test(self):
         import comtypes.test.test_comserver
 
         doctest.testmod(
-            comtypes.test.test_comserver, verbose=False, optionflags=doctest.ELLIPSIS
+            comtypes.test.test_comserver,
+            verbose=False,
+            optionflags=doctest.ELLIPSIS,
+            raise_on_error=True,
         )