Implement static import for ISequentialStream (#474)

comtypes/client/_generate.py

@@ -272,6 +272,7 @@ def _get_known_namespaces() -> Tuple[
         "comtypes.persist",
         "comtypes.typeinfo",
         "comtypes.automation",
+        "comtypes.objidl",
         "comtypes",
         "ctypes.wintypes",
         "ctypes",

comtypes/objidl.py

@@ -0,0 +1,59 @@
+from ctypes import Array, c_ubyte, c_ulong, HRESULT, POINTER, pointer
+from typing import Tuple, TYPE_CHECKING
+
+from comtypes import COMMETHOD, GUID, IUnknown
+
+
+class ISequentialStream(IUnknown):
+    """Defines methods for the stream objects in sequence."""
+
+    _iid_ = GUID("{0C733A30-2A1C-11CE-ADE5-00AA0044773D}")
+    _idlflags_ = []
+
+    _methods_ = [
+        COMMETHOD(
+            [],
+            HRESULT,
+            "RemoteRead",
+            # This call only works if `pv` is pre-allocated with `cb` bytes,
+            # which cannot be done by the high level function generated by metaclasses.
+            # Therefore, we override the high level function to implement this behaviour
+            # and then delegate the call the raw COM method.
+            (["out"], POINTER(c_ubyte), "pv"),
+            (["in"], c_ulong, "cb"),
+            (["out"], POINTER(c_ulong), "pcbRead"),
+        ),
+        COMMETHOD(
+            [],
+            HRESULT,
+            "RemoteWrite",
+            (["in"], POINTER(c_ubyte), "pv"),
+            (["in"], c_ulong, "cb"),
+            (["out"], POINTER(c_ulong), "pcbWritten"),
+        ),
+    ]
+
+    def RemoteRead(self, cb: int) -> Tuple["Array[c_ubyte]", int]:
+        """Reads a specified number of bytes from the stream object into memory
+        starting at the current seek pointer.
+        """
+        # Behaves as if `pv` is pre-allocated with `cb` bytes by the high level func.
+        pv = (c_ubyte * cb)()
+        pcb_read = pointer(c_ulong(0))
+        self.__com_RemoteRead(pv, c_ulong(cb), pcb_read)  # type: ignore
+        return pv, pcb_read.contents.value
+
+    if TYPE_CHECKING:
+
+        def RemoteWrite(self, pv: "Array[c_ubyte]", cb: int) -> int:
+            """Writes a specified number of bytes into the stream object starting at
+            the current seek pointer.
+            """
+            ...
+
+
+# fmt: off
+__known_symbols__ = [
+    'ISequentialStream',
+]
+# fmt: on

comtypes/test/test_client.py

@@ -71,6 +71,12 @@ def test_mscorlib(self):
         # the `_Pointer` interface, rather than importing `_Pointer` from `ctypes`.
         self.assertTrue(issubclass(mod._Pointer, comtypes.IUnknown))
 
+    def test_portabledeviceapi(self):
+        mod = comtypes.client.GetModule("portabledeviceapi.dll")
+        from comtypes.objidl import ISequentialStream
+
+        self.assertTrue(issubclass(mod.IStream, ISequentialStream))
+
     def test_no_replacing_Patch_namespace(self):
         # NOTE: An object named `Patch` is defined in some dll.
         # Depending on how the namespace is defined in the static module,
@@ -117,6 +123,11 @@ def test_symbols_in_comtypes(self):
 
         self._doit(comtypes)
 
+    def test_symbols_in_comtypes_objidl(self):
+        import comtypes.objidl
+
+        self._doit(comtypes.objidl)
+
     def test_symbols_in_comtypes_automation(self):
         import comtypes.automation
 

comtypes/test/test_istream.py

@@ -0,0 +1,53 @@
+import unittest as ut
+
+from ctypes import POINTER, byref, c_bool, c_ubyte
+import comtypes
+import comtypes.client
+
+comtypes.client.GetModule("portabledeviceapi.dll")
+from comtypes.gen.PortableDeviceApiLib import IStream
+
+
+def _create_stream() -> IStream:
+    # Create an IStream
+    stream = POINTER(IStream)()  # type: ignore
+    comtypes._ole32.CreateStreamOnHGlobal(None, c_bool(True), byref(stream))
+    return stream  # type: ignore
+
+
+class Test_RemoteWrite(ut.TestCase):
+    def test_RemoteWrite(self):
+        stream = _create_stream()
+        test_data = "Some data".encode("utf-8")
+        pv = (c_ubyte * len(test_data)).from_buffer(bytearray(test_data))
+
+        written = stream.RemoteWrite(pv, len(test_data))
+
+        # Verification
+        self.assertEqual(written, len(test_data))
+
+
+class Test_RemoteRead(ut.TestCase):
+    def test_RemoteRead(self):
+        stream = _create_stream()
+        test_data = "Some data".encode("utf-8")
+        pv = (c_ubyte * len(test_data)).from_buffer(bytearray(test_data))
+        stream.RemoteWrite(pv, len(test_data))
+
+        # Make sure the data actually gets written before trying to read back
+        stream.Commit(0)
+        # Move the stream back to the beginning
+        STREAM_SEEK_SET = 0
+        stream.RemoteSeek(0, STREAM_SEEK_SET)
+
+        buffer_size = 1024
+
+        read_buffer, data_read = stream.RemoteRead(buffer_size)
+
+        # Verification
+        self.assertEqual(data_read, len(test_data))
+        self.assertEqual(bytearray(read_buffer)[0:data_read], test_data)
+
+
+if __name__ == "__main__":
+    ut.main()