Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

test_win32file.py 40KB

1 year ago
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085
  1. import datetime
  2. import os
  3. import random
  4. import shutil
  5. import socket
  6. import tempfile
  7. import threading
  8. import time
  9. import unittest
  10. import ntsecuritycon
  11. import pywintypes
  12. import win32api
  13. import win32con
  14. import win32event
  15. import win32file
  16. import win32pipe
  17. import win32timezone
  18. import winerror
  19. from pywin32_testutil import TestSkipped, str2bytes, testmain
  20. try:
  21. set
  22. except NameError:
  23. from sets import Set as set
  24. class TestReadBuffer(unittest.TestCase):
  25. def testLen(self):
  26. buffer = win32file.AllocateReadBuffer(1)
  27. self.assertEqual(len(buffer), 1)
  28. def testSimpleIndex(self):
  29. buffer = win32file.AllocateReadBuffer(1)
  30. buffer[0] = 0xFF
  31. self.assertEqual(buffer[0], 0xFF)
  32. def testSimpleSlice(self):
  33. buffer = win32file.AllocateReadBuffer(2)
  34. val = str2bytes("\0\0")
  35. buffer[:2] = val
  36. self.assertEqual(buffer[0:2], val)
  37. class TestSimpleOps(unittest.TestCase):
  38. def testSimpleFiles(self):
  39. fd, filename = tempfile.mkstemp()
  40. os.close(fd)
  41. os.unlink(filename)
  42. handle = win32file.CreateFile(
  43. filename, win32file.GENERIC_WRITE, 0, None, win32con.CREATE_NEW, 0, None
  44. )
  45. test_data = str2bytes("Hello\0there")
  46. try:
  47. win32file.WriteFile(handle, test_data)
  48. handle.Close()
  49. # Try and open for read
  50. handle = win32file.CreateFile(
  51. filename,
  52. win32file.GENERIC_READ,
  53. 0,
  54. None,
  55. win32con.OPEN_EXISTING,
  56. 0,
  57. None,
  58. )
  59. rc, data = win32file.ReadFile(handle, 1024)
  60. self.assertEqual(data, test_data)
  61. finally:
  62. handle.Close()
  63. try:
  64. os.unlink(filename)
  65. except os.error:
  66. pass
  67. # A simple test using normal read/write operations.
  68. def testMoreFiles(self):
  69. # Create a file in the %TEMP% directory.
  70. testName = os.path.join(win32api.GetTempPath(), "win32filetest.dat")
  71. desiredAccess = win32file.GENERIC_READ | win32file.GENERIC_WRITE
  72. # Set a flag to delete the file automatically when it is closed.
  73. fileFlags = win32file.FILE_FLAG_DELETE_ON_CLOSE
  74. h = win32file.CreateFile(
  75. testName,
  76. desiredAccess,
  77. win32file.FILE_SHARE_READ,
  78. None,
  79. win32file.CREATE_ALWAYS,
  80. fileFlags,
  81. 0,
  82. )
  83. # Write a known number of bytes to the file.
  84. data = str2bytes("z") * 1025
  85. win32file.WriteFile(h, data)
  86. self.assertTrue(
  87. win32file.GetFileSize(h) == len(data),
  88. "WARNING: Written file does not have the same size as the length of the data in it!",
  89. )
  90. # Ensure we can read the data back.
  91. win32file.SetFilePointer(h, 0, win32file.FILE_BEGIN)
  92. hr, read_data = win32file.ReadFile(
  93. h, len(data) + 10
  94. ) # + 10 to get anything extra
  95. self.assertTrue(hr == 0, "Readfile returned %d" % hr)
  96. self.assertTrue(read_data == data, "Read data is not what we wrote!")
  97. # Now truncate the file at 1/2 its existing size.
  98. newSize = len(data) // 2
  99. win32file.SetFilePointer(h, newSize, win32file.FILE_BEGIN)
  100. win32file.SetEndOfFile(h)
  101. self.assertEqual(win32file.GetFileSize(h), newSize)
  102. # GetFileAttributesEx/GetFileAttributesExW tests.
  103. self.assertEqual(
  104. win32file.GetFileAttributesEx(testName),
  105. win32file.GetFileAttributesExW(testName),
  106. )
  107. attr, ct, at, wt, size = win32file.GetFileAttributesEx(testName)
  108. self.assertTrue(
  109. size == newSize,
  110. "Expected GetFileAttributesEx to return the same size as GetFileSize()",
  111. )
  112. self.assertTrue(
  113. attr == win32file.GetFileAttributes(testName),
  114. "Expected GetFileAttributesEx to return the same attributes as GetFileAttributes",
  115. )
  116. h = None # Close the file by removing the last reference to the handle!
  117. self.assertTrue(
  118. not os.path.isfile(testName), "After closing the file, it still exists!"
  119. )
  120. def testFilePointer(self):
  121. # via [ 979270 ] SetFilePointer fails with negative offset
  122. # Create a file in the %TEMP% directory.
  123. filename = os.path.join(win32api.GetTempPath(), "win32filetest.dat")
  124. f = win32file.CreateFile(
  125. filename,
  126. win32file.GENERIC_READ | win32file.GENERIC_WRITE,
  127. 0,
  128. None,
  129. win32file.CREATE_ALWAYS,
  130. win32file.FILE_ATTRIBUTE_NORMAL,
  131. 0,
  132. )
  133. try:
  134. # Write some data
  135. data = str2bytes("Some data")
  136. (res, written) = win32file.WriteFile(f, data)
  137. self.assertFalse(res)
  138. self.assertEqual(written, len(data))
  139. # Move at the beginning and read the data
  140. win32file.SetFilePointer(f, 0, win32file.FILE_BEGIN)
  141. (res, s) = win32file.ReadFile(f, len(data))
  142. self.assertFalse(res)
  143. self.assertEqual(s, data)
  144. # Move at the end and read the data
  145. win32file.SetFilePointer(f, -len(data), win32file.FILE_END)
  146. (res, s) = win32file.ReadFile(f, len(data))
  147. self.assertFalse(res)
  148. self.assertEqual(s, data)
  149. finally:
  150. f.Close()
  151. os.unlink(filename)
  152. def testFileTimesTimezones(self):
  153. filename = tempfile.mktemp("-testFileTimes")
  154. # now() is always returning a timestamp with microseconds but the
  155. # file APIs all have zero microseconds, so some comparisons fail.
  156. now_utc = win32timezone.utcnow().replace(microsecond=0)
  157. now_local = now_utc.astimezone(win32timezone.TimeZoneInfo.local())
  158. h = win32file.CreateFile(
  159. filename,
  160. win32file.GENERIC_READ | win32file.GENERIC_WRITE,
  161. 0,
  162. None,
  163. win32file.CREATE_ALWAYS,
  164. 0,
  165. 0,
  166. )
  167. try:
  168. win32file.SetFileTime(h, now_utc, now_utc, now_utc)
  169. ct, at, wt = win32file.GetFileTime(h)
  170. self.assertEqual(now_local, ct)
  171. self.assertEqual(now_local, at)
  172. self.assertEqual(now_local, wt)
  173. # and the reverse - set local, check against utc
  174. win32file.SetFileTime(h, now_local, now_local, now_local)
  175. ct, at, wt = win32file.GetFileTime(h)
  176. self.assertEqual(now_utc, ct)
  177. self.assertEqual(now_utc, at)
  178. self.assertEqual(now_utc, wt)
  179. finally:
  180. h.close()
  181. os.unlink(filename)
  182. def testFileTimes(self):
  183. from win32timezone import TimeZoneInfo
  184. # now() is always returning a timestamp with microseconds but the
  185. # file APIs all have zero microseconds, so some comparisons fail.
  186. now = datetime.datetime.now(tz=TimeZoneInfo.utc()).replace(microsecond=0)
  187. nowish = now + datetime.timedelta(seconds=1)
  188. later = now + datetime.timedelta(seconds=120)
  189. filename = tempfile.mktemp("-testFileTimes")
  190. # Windows docs the 'last time' isn't valid until the last write
  191. # handle is closed - so create the file, then re-open it to check.
  192. open(filename, "w").close()
  193. f = win32file.CreateFile(
  194. filename,
  195. win32file.GENERIC_READ | win32file.GENERIC_WRITE,
  196. 0,
  197. None,
  198. win32con.OPEN_EXISTING,
  199. 0,
  200. None,
  201. )
  202. try:
  203. ct, at, wt = win32file.GetFileTime(f)
  204. self.assertTrue(
  205. ct >= now,
  206. "File was created in the past - now=%s, created=%s" % (now, ct),
  207. )
  208. self.assertTrue(now <= ct <= nowish, (now, ct))
  209. self.assertTrue(
  210. wt >= now,
  211. "File was written-to in the past now=%s, written=%s" % (now, wt),
  212. )
  213. self.assertTrue(now <= wt <= nowish, (now, wt))
  214. # Now set the times.
  215. win32file.SetFileTime(f, later, later, later, UTCTimes=True)
  216. # Get them back.
  217. ct, at, wt = win32file.GetFileTime(f)
  218. # XXX - the builtin PyTime type appears to be out by a dst offset.
  219. # just ignore that type here...
  220. self.assertEqual(ct, later)
  221. self.assertEqual(at, later)
  222. self.assertEqual(wt, later)
  223. finally:
  224. f.Close()
  225. os.unlink(filename)
  226. class TestGetFileInfoByHandleEx(unittest.TestCase):
  227. __handle = __filename = None
  228. def setUp(self):
  229. fd, self.__filename = tempfile.mkstemp()
  230. os.close(fd)
  231. def tearDown(self):
  232. if self.__handle is not None:
  233. self.__handle.Close()
  234. if self.__filename is not None:
  235. try:
  236. os.unlink(self.__filename)
  237. except OSError:
  238. pass
  239. self.__handle = self.__filename = None
  240. def testFileBasicInfo(self):
  241. attr = win32file.GetFileAttributes(self.__filename)
  242. f = win32file.CreateFile(
  243. self.__filename,
  244. win32file.GENERIC_READ,
  245. 0,
  246. None,
  247. win32con.OPEN_EXISTING,
  248. 0,
  249. None,
  250. )
  251. self.__handle = f
  252. ct, at, wt = win32file.GetFileTime(f)
  253. # bug #752: this throws ERROR_BAD_LENGTH (24) in x86 binaries of build 221
  254. basic_info = win32file.GetFileInformationByHandleEx(f, win32file.FileBasicInfo)
  255. self.assertEqual(ct, basic_info["CreationTime"])
  256. self.assertEqual(at, basic_info["LastAccessTime"])
  257. self.assertEqual(wt, basic_info["LastWriteTime"])
  258. self.assertEqual(attr, basic_info["FileAttributes"])
  259. class TestOverlapped(unittest.TestCase):
  260. def testSimpleOverlapped(self):
  261. # Create a file in the %TEMP% directory.
  262. import win32event
  263. testName = os.path.join(win32api.GetTempPath(), "win32filetest.dat")
  264. desiredAccess = win32file.GENERIC_WRITE
  265. overlapped = pywintypes.OVERLAPPED()
  266. evt = win32event.CreateEvent(None, 0, 0, None)
  267. overlapped.hEvent = evt
  268. # Create the file and write shit-loads of data to it.
  269. h = win32file.CreateFile(
  270. testName, desiredAccess, 0, None, win32file.CREATE_ALWAYS, 0, 0
  271. )
  272. chunk_data = str2bytes("z") * 0x8000
  273. num_loops = 512
  274. expected_size = num_loops * len(chunk_data)
  275. for i in range(num_loops):
  276. win32file.WriteFile(h, chunk_data, overlapped)
  277. win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
  278. overlapped.Offset = overlapped.Offset + len(chunk_data)
  279. h.Close()
  280. # Now read the data back overlapped
  281. overlapped = pywintypes.OVERLAPPED()
  282. evt = win32event.CreateEvent(None, 0, 0, None)
  283. overlapped.hEvent = evt
  284. desiredAccess = win32file.GENERIC_READ
  285. h = win32file.CreateFile(
  286. testName, desiredAccess, 0, None, win32file.OPEN_EXISTING, 0, 0
  287. )
  288. buffer = win32file.AllocateReadBuffer(0xFFFF)
  289. while 1:
  290. try:
  291. hr, data = win32file.ReadFile(h, buffer, overlapped)
  292. win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
  293. overlapped.Offset = overlapped.Offset + len(data)
  294. if not data is buffer:
  295. self.fail(
  296. "Unexpected result from ReadFile - should be the same buffer we passed it"
  297. )
  298. except win32api.error:
  299. break
  300. h.Close()
  301. def testCompletionPortsMultiple(self):
  302. # Mainly checking that we can "associate" an existing handle. This
  303. # failed in build 203.
  304. ioport = win32file.CreateIoCompletionPort(
  305. win32file.INVALID_HANDLE_VALUE, 0, 0, 0
  306. )
  307. socks = []
  308. for PORT in range(9123, 9125):
  309. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  310. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  311. sock.bind(("", PORT))
  312. sock.listen(1)
  313. socks.append(sock)
  314. new = win32file.CreateIoCompletionPort(sock.fileno(), ioport, PORT, 0)
  315. assert new is ioport
  316. for s in socks:
  317. s.close()
  318. hv = int(ioport)
  319. ioport = new = None
  320. # The handle itself should be closed now (unless we leak references!)
  321. # Check that.
  322. try:
  323. win32file.CloseHandle(hv)
  324. raise RuntimeError("Expected close to fail!")
  325. except win32file.error as details:
  326. self.assertEqual(details.winerror, winerror.ERROR_INVALID_HANDLE)
  327. def testCompletionPortsQueued(self):
  328. class Foo:
  329. pass
  330. io_req_port = win32file.CreateIoCompletionPort(-1, None, 0, 0)
  331. overlapped = pywintypes.OVERLAPPED()
  332. overlapped.object = Foo()
  333. win32file.PostQueuedCompletionStatus(io_req_port, 0, 99, overlapped)
  334. errCode, bytes, key, overlapped = win32file.GetQueuedCompletionStatus(
  335. io_req_port, win32event.INFINITE
  336. )
  337. self.assertEqual(errCode, 0)
  338. self.assertTrue(isinstance(overlapped.object, Foo))
  339. def _IOCPServerThread(self, handle, port, drop_overlapped_reference):
  340. overlapped = pywintypes.OVERLAPPED()
  341. win32pipe.ConnectNamedPipe(handle, overlapped)
  342. if drop_overlapped_reference:
  343. # Be naughty - the overlapped object is now dead, but
  344. # GetQueuedCompletionStatus will still find it. Our check of
  345. # reference counting should catch that error.
  346. overlapped = None
  347. # even if we fail, be sure to close the handle; prevents hangs
  348. # on Vista 64...
  349. try:
  350. self.assertRaises(
  351. RuntimeError, win32file.GetQueuedCompletionStatus, port, -1
  352. )
  353. finally:
  354. handle.Close()
  355. return
  356. result = win32file.GetQueuedCompletionStatus(port, -1)
  357. ol2 = result[-1]
  358. self.assertTrue(ol2 is overlapped)
  359. data = win32file.ReadFile(handle, 512)[1]
  360. win32file.WriteFile(handle, data)
  361. def testCompletionPortsNonQueued(self, test_overlapped_death=0):
  362. # In 204 we had a reference count bug when OVERLAPPED objects were
  363. # associated with a completion port other than via
  364. # PostQueuedCompletionStatus. This test is based on the reproduction
  365. # reported with that bug.
  366. # Create the pipe.
  367. BUFSIZE = 512
  368. pipe_name = r"\\.\pipe\pywin32_test_pipe"
  369. handle = win32pipe.CreateNamedPipe(
  370. pipe_name,
  371. win32pipe.PIPE_ACCESS_DUPLEX | win32file.FILE_FLAG_OVERLAPPED,
  372. win32pipe.PIPE_TYPE_MESSAGE
  373. | win32pipe.PIPE_READMODE_MESSAGE
  374. | win32pipe.PIPE_WAIT,
  375. 1,
  376. BUFSIZE,
  377. BUFSIZE,
  378. win32pipe.NMPWAIT_WAIT_FOREVER,
  379. None,
  380. )
  381. # Create an IOCP and associate it with the handle.
  382. port = win32file.CreateIoCompletionPort(-1, 0, 0, 0)
  383. win32file.CreateIoCompletionPort(handle, port, 1, 0)
  384. t = threading.Thread(
  385. target=self._IOCPServerThread, args=(handle, port, test_overlapped_death)
  386. )
  387. t.setDaemon(True) # avoid hanging entire test suite on failure.
  388. t.start()
  389. try:
  390. time.sleep(0.1) # let thread do its thing.
  391. try:
  392. win32pipe.CallNamedPipe(
  393. r"\\.\pipe\pywin32_test_pipe", str2bytes("Hello there"), BUFSIZE, 0
  394. )
  395. except win32pipe.error:
  396. # Testing for overlapped death causes this
  397. if not test_overlapped_death:
  398. raise
  399. finally:
  400. if not test_overlapped_death:
  401. handle.Close()
  402. t.join(3)
  403. self.assertFalse(t.is_alive(), "thread didn't finish")
  404. def testCompletionPortsNonQueuedBadReference(self):
  405. self.testCompletionPortsNonQueued(True)
  406. def testHashable(self):
  407. overlapped = pywintypes.OVERLAPPED()
  408. d = {}
  409. d[overlapped] = "hello"
  410. self.assertEqual(d[overlapped], "hello")
  411. def testComparable(self):
  412. overlapped = pywintypes.OVERLAPPED()
  413. self.assertEqual(overlapped, overlapped)
  414. # ensure we explicitly test the operators.
  415. self.assertTrue(overlapped == overlapped)
  416. self.assertFalse(overlapped != overlapped)
  417. def testComparable2(self):
  418. # 2 overlapped objects compare equal if their contents are the same.
  419. overlapped1 = pywintypes.OVERLAPPED()
  420. overlapped2 = pywintypes.OVERLAPPED()
  421. self.assertEqual(overlapped1, overlapped2)
  422. # ensure we explicitly test the operators.
  423. self.assertTrue(overlapped1 == overlapped2)
  424. self.assertFalse(overlapped1 != overlapped2)
  425. # now change something in one of them - should no longer be equal.
  426. overlapped1.hEvent = 1
  427. self.assertNotEqual(overlapped1, overlapped2)
  428. # ensure we explicitly test the operators.
  429. self.assertFalse(overlapped1 == overlapped2)
  430. self.assertTrue(overlapped1 != overlapped2)
  431. class TestSocketExtensions(unittest.TestCase):
  432. def acceptWorker(self, port, running_event, stopped_event):
  433. listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  434. listener.bind(("", port))
  435. listener.listen(200)
  436. # create accept socket
  437. accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  438. # An overlapped
  439. overlapped = pywintypes.OVERLAPPED()
  440. overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
  441. # accept the connection.
  442. # We used to allow strings etc to be passed here, and they would be
  443. # modified! Obviously this is evil :)
  444. buffer = " " * 1024 # EVIL - SHOULD NOT BE ALLOWED.
  445. self.assertRaises(
  446. TypeError, win32file.AcceptEx, listener, accepter, buffer, overlapped
  447. )
  448. # This is the correct way to allocate the buffer...
  449. buffer = win32file.AllocateReadBuffer(1024)
  450. rc = win32file.AcceptEx(listener, accepter, buffer, overlapped)
  451. self.assertEqual(rc, winerror.ERROR_IO_PENDING)
  452. # Set the event to say we are all ready
  453. running_event.set()
  454. # and wait for the connection.
  455. rc = win32event.WaitForSingleObject(overlapped.hEvent, 2000)
  456. if rc == win32event.WAIT_TIMEOUT:
  457. self.fail("timed out waiting for a connection")
  458. nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
  459. # fam, loc, rem = win32file.GetAcceptExSockaddrs(accepter, buffer)
  460. accepter.send(buffer[:nbytes])
  461. # NOT set in a finally - this means *successfully* stopped!
  462. stopped_event.set()
  463. def testAcceptEx(self):
  464. port = 4680
  465. running = threading.Event()
  466. stopped = threading.Event()
  467. t = threading.Thread(target=self.acceptWorker, args=(port, running, stopped))
  468. t.start()
  469. running.wait(2)
  470. if not running.isSet():
  471. self.fail("AcceptEx Worker thread failed to start")
  472. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  473. s.connect(("127.0.0.1", port))
  474. win32file.WSASend(s, str2bytes("hello"), None)
  475. overlapped = pywintypes.OVERLAPPED()
  476. overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
  477. # Like above - WSARecv used to allow strings as the receive buffer!!
  478. buffer = " " * 10
  479. self.assertRaises(TypeError, win32file.WSARecv, s, buffer, overlapped)
  480. # This one should work :)
  481. buffer = win32file.AllocateReadBuffer(10)
  482. win32file.WSARecv(s, buffer, overlapped)
  483. nbytes = win32file.GetOverlappedResult(s.fileno(), overlapped, True)
  484. got = buffer[:nbytes]
  485. self.assertEqual(got, str2bytes("hello"))
  486. # thread should have stopped
  487. stopped.wait(2)
  488. if not stopped.isSet():
  489. self.fail("AcceptEx Worker thread failed to successfully stop")
  490. class TestFindFiles(unittest.TestCase):
  491. def testIter(self):
  492. dir = os.path.join(os.getcwd(), "*")
  493. files = win32file.FindFilesW(dir)
  494. set1 = set()
  495. set1.update(files)
  496. set2 = set()
  497. for file in win32file.FindFilesIterator(dir):
  498. set2.add(file)
  499. assert len(set2) > 5, "This directory has less than 5 files!?"
  500. self.assertEqual(set1, set2)
  501. def testBadDir(self):
  502. dir = os.path.join(os.getcwd(), "a dir that doesnt exist", "*")
  503. self.assertRaises(win32file.error, win32file.FindFilesIterator, dir)
  504. def testEmptySpec(self):
  505. spec = os.path.join(os.getcwd(), "*.foo_bar")
  506. num = 0
  507. for i in win32file.FindFilesIterator(spec):
  508. num += 1
  509. self.assertEqual(0, num)
  510. def testEmptyDir(self):
  511. test_path = os.path.join(win32api.GetTempPath(), "win32file_test_directory")
  512. try:
  513. # Note: previously used shutil.rmtree, but when looking for
  514. # reference count leaks, that function showed leaks! os.rmdir
  515. # doesn't have that problem.
  516. os.rmdir(test_path)
  517. except os.error:
  518. pass
  519. os.mkdir(test_path)
  520. try:
  521. num = 0
  522. for i in win32file.FindFilesIterator(os.path.join(test_path, "*")):
  523. num += 1
  524. # Expecting "." and ".." only
  525. self.assertEqual(2, num)
  526. finally:
  527. os.rmdir(test_path)
  528. class TestDirectoryChanges(unittest.TestCase):
  529. num_test_dirs = 1
  530. def setUp(self):
  531. self.watcher_threads = []
  532. self.watcher_thread_changes = []
  533. self.dir_names = []
  534. self.dir_handles = []
  535. for i in range(self.num_test_dirs):
  536. td = tempfile.mktemp("-test-directory-changes-%d" % i)
  537. os.mkdir(td)
  538. self.dir_names.append(td)
  539. hdir = win32file.CreateFile(
  540. td,
  541. ntsecuritycon.FILE_LIST_DIRECTORY,
  542. win32con.FILE_SHARE_READ,
  543. None, # security desc
  544. win32con.OPEN_EXISTING,
  545. win32con.FILE_FLAG_BACKUP_SEMANTICS | win32con.FILE_FLAG_OVERLAPPED,
  546. None,
  547. )
  548. self.dir_handles.append(hdir)
  549. changes = []
  550. t = threading.Thread(
  551. target=self._watcherThreadOverlapped, args=(td, hdir, changes)
  552. )
  553. t.start()
  554. self.watcher_threads.append(t)
  555. self.watcher_thread_changes.append(changes)
  556. def _watcherThread(self, dn, dh, changes):
  557. # A synchronous version:
  558. # XXX - not used - I was having a whole lot of problems trying to
  559. # get this to work. Specifically:
  560. # * ReadDirectoryChangesW without an OVERLAPPED blocks infinitely.
  561. # * If another thread attempts to close the handle while
  562. # ReadDirectoryChangesW is waiting on it, the ::CloseHandle() method
  563. # blocks (which has nothing to do with the GIL - it is correctly
  564. # managed)
  565. # Which ends up with no way to kill the thread!
  566. flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME
  567. while 1:
  568. try:
  569. print("waiting", dh)
  570. changes = win32file.ReadDirectoryChangesW(
  571. dh, 8192, False, flags # sub-tree
  572. )
  573. print("got", changes)
  574. except:
  575. raise
  576. changes.extend(changes)
  577. def _watcherThreadOverlapped(self, dn, dh, changes):
  578. flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME
  579. buf = win32file.AllocateReadBuffer(8192)
  580. overlapped = pywintypes.OVERLAPPED()
  581. overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
  582. while 1:
  583. win32file.ReadDirectoryChangesW(
  584. dh, buf, False, flags, overlapped # sub-tree
  585. )
  586. # Wait for our event, or for 5 seconds.
  587. rc = win32event.WaitForSingleObject(overlapped.hEvent, 5000)
  588. if rc == win32event.WAIT_OBJECT_0:
  589. # got some data! Must use GetOverlappedResult to find out
  590. # how much is valid! 0 generally means the handle has
  591. # been closed. Blocking is OK here, as the event has
  592. # already been set.
  593. nbytes = win32file.GetOverlappedResult(dh, overlapped, True)
  594. if nbytes:
  595. bits = win32file.FILE_NOTIFY_INFORMATION(buf, nbytes)
  596. changes.extend(bits)
  597. else:
  598. # This is "normal" exit - our 'tearDown' closes the
  599. # handle.
  600. # print "looks like dir handle was closed!"
  601. return
  602. else:
  603. print("ERROR: Watcher thread timed-out!")
  604. return # kill the thread!
  605. def tearDown(self):
  606. # be careful about raising errors at teardown!
  607. for h in self.dir_handles:
  608. # See comments in _watcherThread above - this appears to
  609. # deadlock if a synchronous ReadDirectoryChangesW is waiting...
  610. # (No such problems with an asynch ReadDirectoryChangesW)
  611. h.Close()
  612. for dn in self.dir_names:
  613. try:
  614. shutil.rmtree(dn)
  615. except OSError:
  616. print("FAILED to remove directory", dn)
  617. for t in self.watcher_threads:
  618. # closing dir handle should have killed threads!
  619. t.join(5)
  620. if t.is_alive():
  621. print("FAILED to wait for thread termination")
  622. def stablize(self):
  623. time.sleep(0.5)
  624. def testSimple(self):
  625. self.stablize()
  626. for dn in self.dir_names:
  627. fn = os.path.join(dn, "test_file")
  628. open(fn, "w").close()
  629. self.stablize()
  630. changes = self.watcher_thread_changes[0]
  631. self.assertEqual(changes, [(1, "test_file")])
  632. def testSmall(self):
  633. self.stablize()
  634. for dn in self.dir_names:
  635. fn = os.path.join(dn, "x")
  636. open(fn, "w").close()
  637. self.stablize()
  638. changes = self.watcher_thread_changes[0]
  639. self.assertEqual(changes, [(1, "x")])
  640. class TestEncrypt(unittest.TestCase):
  641. def testEncrypt(self):
  642. fname = tempfile.mktemp("win32file_test")
  643. f = open(fname, "wb")
  644. f.write(str2bytes("hello"))
  645. f.close()
  646. f = None
  647. try:
  648. try:
  649. win32file.EncryptFile(fname)
  650. except win32file.error as details:
  651. if details.winerror != winerror.ERROR_ACCESS_DENIED:
  652. raise
  653. print("It appears this is not NTFS - cant encrypt/decrypt")
  654. win32file.DecryptFile(fname)
  655. finally:
  656. if f is not None:
  657. f.close()
  658. os.unlink(fname)
  659. class TestConnect(unittest.TestCase):
  660. def connect_thread_runner(self, expect_payload, giveup_event):
  661. # As Windows 2000 doesn't do ConnectEx, we need to use a non-blocking
  662. # accept, as our test connection may never come. May as well use
  663. # AcceptEx for this...
  664. listener = socket.socket()
  665. self.addr = ("localhost", random.randint(10000, 64000))
  666. listener.bind(self.addr)
  667. listener.listen(1)
  668. # create accept socket
  669. accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  670. # An overlapped
  671. overlapped = pywintypes.OVERLAPPED()
  672. overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
  673. # accept the connection.
  674. if expect_payload:
  675. buf_size = 1024
  676. else:
  677. # when we don't expect data we must be careful to only pass the
  678. # exact number of bytes for the endpoint data...
  679. buf_size = win32file.CalculateSocketEndPointSize(listener)
  680. buffer = win32file.AllocateReadBuffer(buf_size)
  681. win32file.AcceptEx(listener, accepter, buffer, overlapped)
  682. # wait for the connection or our test to fail.
  683. events = giveup_event, overlapped.hEvent
  684. rc = win32event.WaitForMultipleObjects(events, False, 2000)
  685. if rc == win32event.WAIT_TIMEOUT:
  686. self.fail("timed out waiting for a connection")
  687. if rc == win32event.WAIT_OBJECT_0:
  688. # Our main thread running the test failed and will never connect.
  689. return
  690. # must be a connection.
  691. nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
  692. if expect_payload:
  693. self.request = buffer[:nbytes]
  694. accepter.send(str2bytes("some expected response"))
  695. def test_connect_with_payload(self):
  696. giveup_event = win32event.CreateEvent(None, 0, 0, None)
  697. t = threading.Thread(
  698. target=self.connect_thread_runner, args=(True, giveup_event)
  699. )
  700. t.start()
  701. time.sleep(0.1)
  702. s2 = socket.socket()
  703. ol = pywintypes.OVERLAPPED()
  704. s2.bind(("0.0.0.0", 0)) # connectex requires the socket be bound beforehand
  705. try:
  706. win32file.ConnectEx(s2, self.addr, ol, str2bytes("some expected request"))
  707. except win32file.error as exc:
  708. win32event.SetEvent(giveup_event)
  709. if exc.winerror == 10022: # WSAEINVAL
  710. raise TestSkipped("ConnectEx is not available on this platform")
  711. raise # some error error we don't expect.
  712. # We occasionally see ERROR_CONNECTION_REFUSED in automation
  713. try:
  714. win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  715. except win32file.error as exc:
  716. win32event.SetEvent(giveup_event)
  717. if exc.winerror == winerror.ERROR_CONNECTION_REFUSED:
  718. raise TestSkipped("Assuming ERROR_CONNECTION_REFUSED is transient")
  719. raise
  720. ol = pywintypes.OVERLAPPED()
  721. buff = win32file.AllocateReadBuffer(1024)
  722. win32file.WSARecv(s2, buff, ol, 0)
  723. length = win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  724. self.response = buff[:length]
  725. self.assertEqual(self.response, str2bytes("some expected response"))
  726. self.assertEqual(self.request, str2bytes("some expected request"))
  727. t.join(5)
  728. self.assertFalse(t.is_alive(), "worker thread didn't terminate")
  729. def test_connect_without_payload(self):
  730. giveup_event = win32event.CreateEvent(None, 0, 0, None)
  731. t = threading.Thread(
  732. target=self.connect_thread_runner, args=(False, giveup_event)
  733. )
  734. t.start()
  735. time.sleep(0.1)
  736. s2 = socket.socket()
  737. ol = pywintypes.OVERLAPPED()
  738. s2.bind(("0.0.0.0", 0)) # connectex requires the socket be bound beforehand
  739. try:
  740. win32file.ConnectEx(s2, self.addr, ol)
  741. except win32file.error as exc:
  742. win32event.SetEvent(giveup_event)
  743. if exc.winerror == 10022: # WSAEINVAL
  744. raise TestSkipped("ConnectEx is not available on this platform")
  745. raise # some error error we don't expect.
  746. # We occasionally see ERROR_CONNECTION_REFUSED in automation
  747. try:
  748. win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  749. except win32file.error as exc:
  750. win32event.SetEvent(giveup_event)
  751. if exc.winerror == winerror.ERROR_CONNECTION_REFUSED:
  752. raise TestSkipped("Assuming ERROR_CONNECTION_REFUSED is transient")
  753. raise
  754. ol = pywintypes.OVERLAPPED()
  755. buff = win32file.AllocateReadBuffer(1024)
  756. win32file.WSARecv(s2, buff, ol, 0)
  757. length = win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  758. self.response = buff[:length]
  759. self.assertEqual(self.response, str2bytes("some expected response"))
  760. t.join(5)
  761. self.assertFalse(t.is_alive(), "worker thread didn't terminate")
  762. class TestTransmit(unittest.TestCase):
  763. def test_transmit(self):
  764. import binascii
  765. bytes = os.urandom(1024 * 1024)
  766. val = binascii.hexlify(bytes)
  767. val_length = len(val)
  768. f = tempfile.TemporaryFile()
  769. f.write(val)
  770. def runner():
  771. s1 = socket.socket()
  772. # binding fails occasionally on github CI with:
  773. # OSError: [WinError 10013] An attempt was made to access a socket in a way forbidden by its access permissions
  774. # which probably just means the random port is already in use, so
  775. # let that happen a few times.
  776. for i in range(5):
  777. self.addr = ("localhost", random.randint(10000, 64000))
  778. try:
  779. s1.bind(self.addr)
  780. break
  781. except os.error as exc:
  782. if exc.winerror != 10013:
  783. raise
  784. print("Failed to use port", self.addr, "trying another random one")
  785. else:
  786. raise RuntimeError("Failed to find an available port to bind to.")
  787. s1.listen(1)
  788. cli, addr = s1.accept()
  789. buf = 1
  790. self.request = []
  791. while buf:
  792. buf = cli.recv(1024 * 100)
  793. self.request.append(buf)
  794. th = threading.Thread(target=runner)
  795. th.start()
  796. time.sleep(0.5)
  797. s2 = socket.socket()
  798. s2.connect(self.addr)
  799. length = 0
  800. aaa = str2bytes("[AAA]")
  801. bbb = str2bytes("[BBB]")
  802. ccc = str2bytes("[CCC]")
  803. ddd = str2bytes("[DDD]")
  804. empty = str2bytes("")
  805. ol = pywintypes.OVERLAPPED()
  806. f.seek(0)
  807. win32file.TransmitFile(
  808. s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0
  809. )
  810. length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  811. ol = pywintypes.OVERLAPPED()
  812. f.seek(0)
  813. win32file.TransmitFile(
  814. s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, aaa, bbb
  815. )
  816. length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  817. ol = pywintypes.OVERLAPPED()
  818. f.seek(0)
  819. win32file.TransmitFile(
  820. s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, empty, empty
  821. )
  822. length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  823. ol = pywintypes.OVERLAPPED()
  824. f.seek(0)
  825. win32file.TransmitFile(
  826. s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, None, ccc
  827. )
  828. length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  829. ol = pywintypes.OVERLAPPED()
  830. f.seek(0)
  831. win32file.TransmitFile(
  832. s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, ddd
  833. )
  834. length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
  835. s2.close()
  836. th.join()
  837. buf = str2bytes("").join(self.request)
  838. self.assertEqual(length, len(buf))
  839. expected = val + aaa + val + bbb + val + val + ccc + ddd + val
  840. self.assertEqual(type(expected), type(buf))
  841. self.assertEqual(expected, buf)
  842. class TestWSAEnumNetworkEvents(unittest.TestCase):
  843. def test_basics(self):
  844. s = socket.socket()
  845. e = win32event.CreateEvent(None, 1, 0, None)
  846. win32file.WSAEventSelect(s, e, 0)
  847. self.assertEqual(win32file.WSAEnumNetworkEvents(s), {})
  848. self.assertEqual(win32file.WSAEnumNetworkEvents(s, e), {})
  849. self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, e, 3)
  850. self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, "spam")
  851. self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam", e)
  852. self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam")
  853. f = open("NUL")
  854. h = win32file._get_osfhandle(f.fileno())
  855. self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, h)
  856. self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, s, h)
  857. try:
  858. win32file.WSAEnumNetworkEvents(h)
  859. except win32file.error as e:
  860. self.assertEqual(e.winerror, win32file.WSAENOTSOCK)
  861. try:
  862. win32file.WSAEnumNetworkEvents(s, h)
  863. except win32file.error as e:
  864. # According to the docs it would seem reasonable that
  865. # this would fail with WSAEINVAL, but it doesn't.
  866. self.assertEqual(e.winerror, win32file.WSAENOTSOCK)
  867. def test_functional(self):
  868. # This is not really a unit test, but it does exercise the code
  869. # quite well and can serve as an example of WSAEventSelect and
  870. # WSAEnumNetworkEvents usage.
  871. port = socket.socket()
  872. port.setblocking(0)
  873. port_event = win32event.CreateEvent(None, 0, 0, None)
  874. win32file.WSAEventSelect(
  875. port, port_event, win32file.FD_ACCEPT | win32file.FD_CLOSE
  876. )
  877. port.bind(("127.0.0.1", 0))
  878. port.listen(10)
  879. client = socket.socket()
  880. client.setblocking(0)
  881. client_event = win32event.CreateEvent(None, 0, 0, None)
  882. win32file.WSAEventSelect(
  883. client,
  884. client_event,
  885. win32file.FD_CONNECT
  886. | win32file.FD_READ
  887. | win32file.FD_WRITE
  888. | win32file.FD_CLOSE,
  889. )
  890. err = client.connect_ex(port.getsockname())
  891. self.assertEqual(err, win32file.WSAEWOULDBLOCK)
  892. res = win32event.WaitForSingleObject(port_event, 1000)
  893. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  894. events = win32file.WSAEnumNetworkEvents(port, port_event)
  895. self.assertEqual(events, {win32file.FD_ACCEPT: 0})
  896. server, addr = port.accept()
  897. server.setblocking(0)
  898. server_event = win32event.CreateEvent(None, 1, 0, None)
  899. win32file.WSAEventSelect(
  900. server,
  901. server_event,
  902. win32file.FD_READ | win32file.FD_WRITE | win32file.FD_CLOSE,
  903. )
  904. res = win32event.WaitForSingleObject(server_event, 1000)
  905. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  906. events = win32file.WSAEnumNetworkEvents(server, server_event)
  907. self.assertEqual(events, {win32file.FD_WRITE: 0})
  908. res = win32event.WaitForSingleObject(client_event, 1000)
  909. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  910. events = win32file.WSAEnumNetworkEvents(client, client_event)
  911. self.assertEqual(events, {win32file.FD_CONNECT: 0, win32file.FD_WRITE: 0})
  912. sent = 0
  913. data = str2bytes("x") * 16 * 1024
  914. while sent < 16 * 1024 * 1024:
  915. try:
  916. sent += client.send(data)
  917. except socket.error as e:
  918. if e.args[0] == win32file.WSAEINTR:
  919. continue
  920. elif e.args[0] in (win32file.WSAEWOULDBLOCK, win32file.WSAENOBUFS):
  921. break
  922. else:
  923. raise
  924. else:
  925. self.fail("could not find socket buffer limit")
  926. events = win32file.WSAEnumNetworkEvents(client)
  927. self.assertEqual(events, {})
  928. res = win32event.WaitForSingleObject(server_event, 1000)
  929. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  930. events = win32file.WSAEnumNetworkEvents(server, server_event)
  931. self.assertEqual(events, {win32file.FD_READ: 0})
  932. received = 0
  933. while received < sent:
  934. try:
  935. received += len(server.recv(16 * 1024))
  936. except socket.error as e:
  937. if e.args[0] in [win32file.WSAEINTR, win32file.WSAEWOULDBLOCK]:
  938. continue
  939. else:
  940. raise
  941. self.assertEqual(received, sent)
  942. events = win32file.WSAEnumNetworkEvents(server)
  943. self.assertEqual(events, {})
  944. res = win32event.WaitForSingleObject(client_event, 1000)
  945. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  946. events = win32file.WSAEnumNetworkEvents(client, client_event)
  947. self.assertEqual(events, {win32file.FD_WRITE: 0})
  948. client.shutdown(socket.SHUT_WR)
  949. res = win32event.WaitForSingleObject(server_event, 1000)
  950. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  951. # strange timing issues...
  952. for i in range(5):
  953. events = win32file.WSAEnumNetworkEvents(server, server_event)
  954. if events:
  955. break
  956. win32api.Sleep(100)
  957. else:
  958. raise AssertionError("failed to get events")
  959. self.assertEqual(events, {win32file.FD_CLOSE: 0})
  960. events = win32file.WSAEnumNetworkEvents(client)
  961. self.assertEqual(events, {})
  962. server.close()
  963. res = win32event.WaitForSingleObject(client_event, 1000)
  964. self.assertEqual(res, win32event.WAIT_OBJECT_0)
  965. events = win32file.WSAEnumNetworkEvents(client, client_event)
  966. self.assertEqual(events, {win32file.FD_CLOSE: 0})
  967. client.close()
  968. events = win32file.WSAEnumNetworkEvents(port)
  969. self.assertEqual(events, {})
  970. if __name__ == "__main__":
  971. testmain()