Skip to content

Commit 5932fe9

Browse files
junkmdjunkmd
authored
Enhance IStorage and IStream testing and refactoring. (#896)
* test: Add `LockRegion`/`UnlockRegion` tests for `IStream`. Implement new tests for `IStream.LockRegion` and `IStream.UnlockRegion` in `test_stream.py`. These tests verify the behavior of these methods for both memory-backed and file-backed streams. * docs: Clarify `IStream::Revert` behavior in `test_stream.py`. Add comments to `comtypes/test/test_stream.py` explaining the expected behavior and limitations of the `IStream::Revert` method. * refactor: Rename `_create_stream` to `_create_stream_on_hglobal`. Renamed the helper function `_create_stream` in `comtypes/test/test_stream.py` to `_create_stream_on_hglobal` to better reflect its purpose of creating a stream backed by a global memory handle. All usages have been updated. * test: Use `portabledeviceapi.dll` for `IStorage` tests Replaced `msvidctl.dll` with `portabledeviceapi.dll` in `test/test_storage.py` for `IStorage` related tests. This change allows for the removal of the `contextlib.redirect_stdout` call, as `portabledeviceapi.dll` does not generate warning messages during module generation, unlike `msvidctl.dll`. * refactor: Remove unused `OPEN_STM_FLAG` in `test_storage.py`. The `OPEN_STM_FLAG` constant in `comtypes/test/test_storage.py` was identified as unused and has been removed to clean up the codebase. * docs: Explain skipped `IStorage` tests in `test_storage.py`. Adds comments to `comtypes/test/test_storage.py` to clarify why `OpenStream` and `EnumElements` are currently skipped due to difficulties in calling remote-side auto-generated methods. * refactor: Standardize `IStorage` and `IStream` creation flags Refactored `Test_IStorage` in `test_storage.py` to introduce new constants. These constants standardize the creation and access modes for `IStorage` and `IStream` objects within the tests, improving readability and reducing redundancy. * feat: Add `test_SetClass` for `IStorage`. Introduced `test_SetClass` in `test_storage.py` to verify the `IStorage.SetClass` method's functionality, including setting and resetting CLSIDs. * refactor: Use `cm` for `assertRaises` context managers in `test_storage.py` Standardize the variable name for `assertRaises` context managers from `ctx` to `cm` within `comtypes/test/test_storage.py` for consistency. * test: Add `IStorage.Stat` tests. Introduced `test_Stat` in `test_storage.py` to thoroughly verify the behavior of the `IStorage.Stat` method, including error handling for invalid flags and correct retrieval of `tagSTATSTG` information. The `_create_docfile` helper method was updated to allow specifying a file path, enabling more precise control over test document creation and allowing for the validation of temporary file creation and existence.
1 parent 9c742c8 commit 5932fe9

File tree

2 files changed

+224
-70
lines changed

2 files changed

+224
-70
lines changed

‎comtypes/test/test_storage.py‎

Lines changed: 111 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,18 @@
1-
import contextlib
1+
import tempfile
22
import unittest
33
from _ctypes import COMError
44
from ctypes import HRESULT, POINTER, OleDLL, byref, c_ubyte
55
from ctypes.wintypes import DWORD, PWCHAR
66
from pathlib import Path
7+
from typing import Optional
78

89
import comtypes
910
import comtypes.client
1011

11-
with contextlib.redirect_stdout(None): # supress warnings
12-
mod = comtypes.client.GetModule("msvidctl.dll")
12+
comtypes.client.GetModule("portabledeviceapi.dll")
13+
from comtypes.gen.PortableDeviceApiLib import IStorage, tagSTATSTG
1314

14-
from comtypes.gen.MSVidCtlLib import IStorage
15+
STGTY_STORAGE = 1
1516

1617
STATFLAG_DEFAULT = 0
1718
STGC_DEFAULT = 0
@@ -26,7 +27,7 @@
2627
STREAM_SEEK_SET = 0
2728

2829
STG_E_PATHNOTFOUND = -2147287038
29-
30+
STG_E_INVALIDFLAG = -2147286785
3031

3132
_ole32 = OleDLL("ole32")
3233

@@ -36,28 +37,25 @@
3637

3738

3839
class Test_IStorage(unittest.TestCase):
39-
CREATE_DOC_FLAG = (
40-
STGM_DIRECT
41-
| STGM_READWRITE
42-
| STGM_CREATE
43-
| STGM_SHARE_EXCLUSIVE
44-
| STGM_DELETEONRELEASE
45-
)
46-
CREATE_STM_FLAG = STGM_CREATE | STGM_READWRITE | STGM_SHARE_EXCLUSIVE
47-
OPEN_STM_FLAG = STGM_READ | STGM_SHARE_EXCLUSIVE
48-
CREATE_STG_FLAG = STGM_TRANSACTED | STGM_READWRITE | STGM_SHARE_EXCLUSIVE
49-
OPEN_STG_FLAG = STGM_TRANSACTED | STGM_READWRITE | STGM_SHARE_EXCLUSIVE
50-
51-
def _create_docfile(self) -> IStorage:
40+
RW_EXCLUSIVE = STGM_READWRITE | STGM_SHARE_EXCLUSIVE
41+
RW_EXCLUSIVE_TX = RW_EXCLUSIVE | STGM_TRANSACTED
42+
RW_EXCLUSIVE_CREATE = RW_EXCLUSIVE | STGM_CREATE
43+
CREATE_TESTDOC = STGM_DIRECT | STGM_CREATE | RW_EXCLUSIVE
44+
CREATE_TEMP_TESTDOC = CREATE_TESTDOC | STGM_DELETEONRELEASE
45+
46+
def _create_docfile(self, mode: int, name: Optional[str] = None) -> IStorage:
5247
stg = POINTER(IStorage)()
53-
_StgCreateDocfile(None, self.CREATE_DOC_FLAG, 0, byref(stg))
48+
_StgCreateDocfile(name, mode, 0, byref(stg))
5449
return stg # type: ignore
5550

5651
def test_CreateStream(self):
57-
storage = self._create_docfile()
52+
storage = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
53+
# When created with `StgCreateDocfile(NULL, ...)`, `pwcsName` is a
54+
# temporary filename. The file really exists on disk because Windows
55+
# creates an actual temporary file for the compound storage.
5856
filepath = Path(storage.Stat(STATFLAG_DEFAULT).pwcsName)
5957
self.assertTrue(filepath.exists())
60-
stream = storage.CreateStream("example", self.CREATE_STM_FLAG, 0, 0)
58+
stream = storage.CreateStream("example", self.RW_EXCLUSIVE_CREATE, 0, 0)
6159
test_data = b"Some data"
6260
pv = (c_ubyte * len(test_data)).from_buffer(bytearray(test_data))
6361
stream.RemoteWrite(pv, len(test_data))
@@ -70,64 +68,120 @@ def test_CreateStream(self):
7068
del storage
7169
self.assertFalse(filepath.exists())
7270

71+
# TODO: Auto-generated methods based on type info are remote-side and hard
72+
# to call from the client.
73+
# If a proper invocation method or workaround is found, testing
74+
# becomes possible.
75+
# See: https://github.com/enthought/comtypes/issues/607
76+
# def test_RemoteOpenStream(self):
77+
# pass
78+
7379
def test_CreateStorage(self):
74-
parent = self._create_docfile()
75-
child = parent.CreateStorage("child", self.CREATE_STG_FLAG, 0, 0)
80+
parent = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
81+
child = parent.CreateStorage("child", self.RW_EXCLUSIVE_TX, 0, 0)
7682
self.assertEqual("child", child.Stat(STATFLAG_DEFAULT).pwcsName)
7783

7884
def test_OpenStorage(self):
79-
parent = self._create_docfile()
80-
created_child = parent.CreateStorage("child", self.CREATE_STG_FLAG, 0, 0)
81-
del created_child
82-
opened_child = parent.OpenStorage("child", None, self.OPEN_STG_FLAG, None, 0)
83-
self.assertEqual("child", opened_child.Stat(STATFLAG_DEFAULT).pwcsName)
85+
parent = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
86+
with self.assertRaises(COMError) as cm:
87+
parent.OpenStorage("child", None, self.RW_EXCLUSIVE_TX, None, 0)
88+
self.assertEqual(cm.exception.hresult, STG_E_PATHNOTFOUND)
89+
parent.CreateStorage("child", self.RW_EXCLUSIVE_TX, 0, 0)
90+
child = parent.OpenStorage("child", None, self.RW_EXCLUSIVE_TX, None, 0)
91+
self.assertEqual("child", child.Stat(STATFLAG_DEFAULT).pwcsName)
8492

8593
def test_RemoteCopyTo(self):
86-
src_stg = self._create_docfile()
87-
src_stg.CreateStorage("child", self.CREATE_STG_FLAG, 0, 0)
88-
dst_stg = self._create_docfile()
94+
src_stg = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
95+
src_stg.CreateStorage("child", self.RW_EXCLUSIVE_TX, 0, 0)
96+
dst_stg = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
8997
src_stg.RemoteCopyTo(0, None, None, dst_stg)
9098
src_stg.Commit(STGC_DEFAULT)
9199
del src_stg
92-
opened_stg = dst_stg.OpenStorage("child", None, self.OPEN_STG_FLAG, None, 0)
100+
opened_stg = dst_stg.OpenStorage("child", None, self.RW_EXCLUSIVE_TX, None, 0)
93101
self.assertEqual("child", opened_stg.Stat(STATFLAG_DEFAULT).pwcsName)
94102

95103
def test_MoveElementTo(self):
96-
src_stg = self._create_docfile()
97-
src_stg.CreateStorage("foo", self.CREATE_STG_FLAG, 0, 0)
98-
dst_stg = self._create_docfile()
104+
src_stg = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
105+
src_stg.CreateStorage("foo", self.RW_EXCLUSIVE_TX, 0, 0)
106+
dst_stg = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
99107
src_stg.MoveElementTo("foo", dst_stg, "bar", STGMOVE_MOVE)
100-
opened_stg = dst_stg.OpenStorage("bar", None, self.OPEN_STG_FLAG, None, 0)
108+
opened_stg = dst_stg.OpenStorage("bar", None, self.RW_EXCLUSIVE_TX, None, 0)
101109
self.assertEqual("bar", opened_stg.Stat(STATFLAG_DEFAULT).pwcsName)
102-
with self.assertRaises(COMError) as ctx:
103-
src_stg.OpenStorage("foo", None, self.OPEN_STG_FLAG, None, 0)
104-
self.assertEqual(ctx.exception.hresult, STG_E_PATHNOTFOUND)
110+
with self.assertRaises(COMError) as cm:
111+
src_stg.OpenStorage("foo", None, self.RW_EXCLUSIVE_TX, None, 0)
112+
self.assertEqual(cm.exception.hresult, STG_E_PATHNOTFOUND)
105113

106114
def test_Revert(self):
107-
storage = self._create_docfile()
108-
foo = storage.CreateStorage("foo", self.CREATE_STG_FLAG, 0, 0)
109-
foo.CreateStorage("bar", self.CREATE_STG_FLAG, 0, 0)
110-
bar = foo.OpenStorage("bar", None, self.OPEN_STG_FLAG, None, 0)
115+
storage = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
116+
foo = storage.CreateStorage("foo", self.RW_EXCLUSIVE_TX, 0, 0)
117+
foo.CreateStorage("bar", self.RW_EXCLUSIVE_TX, 0, 0)
118+
bar = foo.OpenStorage("bar", None, self.RW_EXCLUSIVE_TX, None, 0)
111119
self.assertEqual("bar", bar.Stat(STATFLAG_DEFAULT).pwcsName)
112120
foo.Revert()
113-
with self.assertRaises(COMError) as ctx:
114-
foo.OpenStorage("bar", None, self.OPEN_STG_FLAG, None, 0)
115-
self.assertEqual(ctx.exception.hresult, STG_E_PATHNOTFOUND)
121+
with self.assertRaises(COMError) as cm:
122+
foo.OpenStorage("bar", None, self.RW_EXCLUSIVE_TX, None, 0)
123+
self.assertEqual(cm.exception.hresult, STG_E_PATHNOTFOUND)
124+
125+
# TODO: Auto-generated methods based on type info are remote-side and hard
126+
# to call from the client.
127+
# If a proper invocation method or workaround is found, testing
128+
# becomes possible.
129+
# See: https://github.com/enthought/comtypes/issues/607
130+
# def test_RemoteEnumElements(self):
131+
# pass
116132

117133
def test_DestroyElement(self):
118-
storage = self._create_docfile()
119-
storage.CreateStorage("example", self.CREATE_STG_FLAG, 0, 0)
134+
storage = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
135+
storage.CreateStorage("example", self.RW_EXCLUSIVE_TX, 0, 0)
120136
storage.DestroyElement("example")
121-
with self.assertRaises(COMError) as ctx:
122-
storage.OpenStorage("example", None, self.OPEN_STG_FLAG, None, 0)
123-
self.assertEqual(ctx.exception.hresult, STG_E_PATHNOTFOUND)
137+
with self.assertRaises(COMError) as cm:
138+
storage.OpenStorage("example", None, self.RW_EXCLUSIVE_TX, None, 0)
139+
self.assertEqual(cm.exception.hresult, STG_E_PATHNOTFOUND)
124140

125141
def test_RenameElement(self):
126-
storage = self._create_docfile()
127-
storage.CreateStorage("example", self.CREATE_STG_FLAG, 0, 0)
142+
storage = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
143+
storage.CreateStorage("example", self.RW_EXCLUSIVE_TX, 0, 0)
128144
storage.RenameElement("example", "sample")
129-
sample = storage.OpenStorage("sample", None, self.OPEN_STG_FLAG, None, 0)
145+
sample = storage.OpenStorage("sample", None, self.RW_EXCLUSIVE_TX, None, 0)
130146
self.assertEqual("sample", sample.Stat(STATFLAG_DEFAULT).pwcsName)
131-
with self.assertRaises(COMError) as ctx:
132-
storage.OpenStorage("example", None, self.OPEN_STG_FLAG, None, 0)
133-
self.assertEqual(ctx.exception.hresult, STG_E_PATHNOTFOUND)
147+
with self.assertRaises(COMError) as cm:
148+
storage.OpenStorage("example", None, self.RW_EXCLUSIVE_TX, None, 0)
149+
self.assertEqual(cm.exception.hresult, STG_E_PATHNOTFOUND)
150+
151+
def test_SetClass(self):
152+
storage = self._create_docfile(mode=self.CREATE_TEMP_TESTDOC)
153+
# Initial value is CLSID_NULL.
154+
self.assertEqual(storage.Stat(STATFLAG_DEFAULT).clsid, comtypes.GUID())
155+
new_clsid = comtypes.GUID.create_new()
156+
storage.SetClass(new_clsid)
157+
self.assertEqual(storage.Stat(STATFLAG_DEFAULT).clsid, new_clsid)
158+
# Re-set CLSID to CLSID_NULL and verify it is correctly set.
159+
storage.SetClass(comtypes.GUID())
160+
self.assertEqual(storage.Stat(STATFLAG_DEFAULT).clsid, comtypes.GUID())
161+
162+
def test_Stat(self):
163+
with tempfile.TemporaryDirectory() as t:
164+
tmpdir = Path(t)
165+
tmpfile = tmpdir / "test_docfile.cfs"
166+
self.assertFalse(tmpfile.exists())
167+
# When created with `StgCreateDocfile(filepath_string, ...)`, the
168+
# compound file is created at that location.
169+
storage = self._create_docfile(
170+
name=str(tmpfile), mode=self.CREATE_TEMP_TESTDOC
171+
)
172+
self.assertTrue(tmpfile.exists())
173+
with self.assertRaises(COMError) as cm:
174+
storage.Stat(0xFFFFFFFF) # Invalid flag
175+
self.assertEqual(cm.exception.hresult, STG_E_INVALIDFLAG)
176+
stat = storage.Stat(STATFLAG_DEFAULT)
177+
self.assertIsInstance(stat, tagSTATSTG)
178+
del storage # Release the storage to prevent 'cannot access the file ...'
179+
self.assertEqual(stat.type, STGTY_STORAGE)
180+
# Due to header overhead and file system allocation, the size may be
181+
# greater than 0 bytes.
182+
self.assertGreaterEqual(stat.cbSize, 0)
183+
# `grfMode` should reflect the access mode flags from creation.
184+
self.assertEqual(stat.grfMode, self.RW_EXCLUSIVE | STGM_DIRECT)
185+
self.assertEqual(stat.grfLocksSupported, 0)
186+
self.assertEqual(stat.clsid, comtypes.GUID()) # CLSID_NULL for new creation.
187+
self.assertEqual(stat.grfStateBits, 0)

0 commit comments

Comments
 (0)