Uploaded Test files

This commit is contained in:
Batuhan Berk Başoğlu 2020-11-12 11:05:57 -05:00
parent f584ad9d97
commit 2e81cb7d99
16627 changed files with 2065359 additions and 102444 deletions

View file

@ -0,0 +1,153 @@
import sys
import unittest
import pywintypes
import win32api
from pywin32_testutil import int2long
# A class that will never die vie refcounting, but will die via GC.
class Cycle:
def __init__(self, handle):
self.cycle = self
self.handle = handle
class PyHandleTestCase(unittest.TestCase):
def testCleanup1(self):
# We used to clobber all outstanding exceptions.
def f1(invalidate):
import win32event
h = win32event.CreateEvent(None, 0, 0, None)
if invalidate:
win32api.CloseHandle(int(h))
1/0
# If we invalidated, then the object destruction code will attempt
# to close an invalid handle. We don't wan't an exception in
# this case
def f2(invalidate):
""" This function should throw an IOError. """
try:
f1(invalidate)
except ZeroDivisionError as exc:
raise IOError("raise 2")
self.assertRaises(IOError, f2, False)
# Now do it again, but so the auto object destruction
# actually fails.
self.assertRaises(IOError, f2, True)
def testCleanup2(self):
# Cause an exception during object destruction.
# The worst this does is cause an ".XXX undetected error (why=3)"
# So avoiding that is the goal
import win32event
h = win32event.CreateEvent(None, 0, 0, None)
# Close the handle underneath the object.
win32api.CloseHandle(int(h))
# Object destructor runs with the implicit close failing
h = None
def testCleanup3(self):
# And again with a class - no __del__
import win32event
class Test:
def __init__(self):
self.h = win32event.CreateEvent(None, 0, 0, None)
win32api.CloseHandle(int(self.h))
t=Test()
t = None
def testCleanupGood(self):
# And check that normal error semantics *do* work.
import win32event
h = win32event.CreateEvent(None, 0, 0, None)
win32api.CloseHandle(int(h))
self.assertRaises(win32api.error, h.Close)
# A following Close is documented as working
h.Close()
def testInvalid(self):
h=pywintypes.HANDLE(-2)
self.assertRaises(win32api.error, h.Close)
def testOtherHandle(self):
h=pywintypes.HANDLE(1)
h2=pywintypes.HANDLE(h)
self.failUnlessEqual(h, h2)
# but the above doesn't really test everything - we want a way to
# pass the handle directly into PyWinLong_AsVoidPtr. One way to
# to that is to abuse win32api.GetProcAddress() - the 2nd param
# is passed to PyWinLong_AsVoidPtr() if its not a string.
# passing a handle value of '1' should work - there is something
# at that ordinal
win32api.GetProcAddress(sys.dllhandle, h)
def testHandleInDict(self):
h=pywintypes.HANDLE(1)
d = dict(foo=h)
self.failUnlessEqual(d['foo'], h)
def testHandleInDictThenInt(self):
h=pywintypes.HANDLE(1)
d = dict(foo=h)
self.failUnlessEqual(d['foo'], 1)
def testHandleCompareNone(self):
h=pywintypes.HANDLE(1)
self.failIfEqual(h, None)
self.failIfEqual(None, h)
# ensure we use both __eq__ and __ne__ ops
self.failIf(h==None)
self.failUnless(h!=None)
def testHandleCompareInt(self):
h=pywintypes.HANDLE(1)
self.failIfEqual(h, 0)
self.failUnlessEqual(h, 1)
# ensure we use both __eq__ and __ne__ ops
self.failUnless(h==1)
self.failUnless(1==h)
self.failIf(h!=1)
self.failIf(1!=h)
self.failIf(h==0)
self.failIf(0==h)
self.failUnless(h!=0)
self.failUnless(0!=h)
def testHandleNonZero(self):
h=pywintypes.HANDLE(0)
self.failIf(h)
h=pywintypes.HANDLE(1)
self.failUnless(h)
def testLong(self):
# sys.maxint+1 should always be a 'valid' handle, treated as an
# unsigned int, even though it is a long. Although pywin32 should not
# directly create such longs, using struct.unpack() with a P format
# may well return them. eg:
# >>> struct.unpack("P", struct.pack("P", -1))
# (4294967295L,)
try:
big = sys.maxsize
except AttributeError:
big = sys.maxint
pywintypes.HANDLE(big+1)
def testGC(self):
# This used to provoke:
# Fatal Python error: unexpected exception during garbage collection
def make():
h=pywintypes.HANDLE(-2)
c = Cycle(h)
import gc
make()
gc.collect()
def testTypes(self):
self.assertRaises(TypeError, pywintypes.HANDLE, "foo")
self.assertRaises(TypeError, pywintypes.HANDLE, ())
# should be able to get a long!
pywintypes.HANDLE(int2long(0))
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,112 @@
# General test module for win32api - please add some :)
import sys, os
import unittest
from win32clipboard import *
import win32gui, win32con
import pywintypes
import array
from pywin32_testutil import str2bytes
custom_format_name = "PythonClipboardTestFormat"
class CrashingTestCase(unittest.TestCase):
def test_722082(self):
class crasher(object):
pass
obj = crasher()
OpenClipboard()
try:
EmptyClipboard()
# This used to crash - now correctly raises type error.
self.assertRaises(TypeError, SetClipboardData, 0, obj )
finally:
CloseClipboard()
class TestBitmap(unittest.TestCase):
def setUp(self):
self.bmp_handle = None
try:
this_file = __file__
except NameError:
this_file = sys.argv[0]
this_dir = os.path.dirname(this_file)
self.bmp_name = os.path.join(os.path.abspath(this_dir),
"..", "Demos", "images", "smiley.bmp")
self.failUnless(os.path.isfile(self.bmp_name), self.bmp_name)
flags = win32con.LR_DEFAULTSIZE | win32con.LR_LOADFROMFILE
self.bmp_handle = win32gui.LoadImage(0, self.bmp_name,
win32con.IMAGE_BITMAP,
0, 0, flags)
self.failUnless(self.bmp_handle, "Failed to get a bitmap handle")
def tearDown(self):
if self.bmp_handle:
win32gui.DeleteObject(self.bmp_handle)
def test_bitmap_roundtrip(self):
OpenClipboard()
try:
SetClipboardData(win32con.CF_BITMAP, self.bmp_handle)
got_handle = GetClipboardDataHandle(win32con.CF_BITMAP)
self.failUnlessEqual(got_handle, self.bmp_handle)
finally:
CloseClipboard()
class TestStrings(unittest.TestCase):
def setUp(self):
OpenClipboard()
def tearDown(self):
CloseClipboard()
def test_unicode(self):
val = "test-\a9har"
SetClipboardData(win32con.CF_UNICODETEXT, val)
self.failUnlessEqual(GetClipboardData(win32con.CF_UNICODETEXT), val)
def test_unicode_text(self):
val = "test-val"
SetClipboardText(val)
# GetClipboardData doesn't to auto string conversions - so on py3k,
# CF_TEXT returns bytes.
expected = str2bytes(val)
self.failUnlessEqual(GetClipboardData(win32con.CF_TEXT), expected)
SetClipboardText(val, win32con.CF_UNICODETEXT)
self.failUnlessEqual(GetClipboardData(win32con.CF_UNICODETEXT), val)
def test_string(self):
val = str2bytes("test")
SetClipboardData(win32con.CF_TEXT, val)
self.failUnlessEqual(GetClipboardData(win32con.CF_TEXT), val)
class TestGlobalMemory(unittest.TestCase):
def setUp(self):
OpenClipboard()
def tearDown(self):
CloseClipboard()
def test_mem(self):
val = str2bytes("test")
expected = str2bytes("test\0")
SetClipboardData(win32con.CF_TEXT, val)
# Get the raw data - this will include the '\0'
raw_data = GetGlobalMemory(GetClipboardDataHandle(win32con.CF_TEXT))
self.failUnlessEqual(expected, raw_data)
def test_bad_mem(self):
self.failUnlessRaises(pywintypes.error, GetGlobalMemory, 0)
self.failUnlessRaises(pywintypes.error, GetGlobalMemory, -1)
if sys.getwindowsversion()[0] <= 5:
# For some reason, the value '1' dies from a 64bit process, but
# "works" (ie, gives the correct exception) from a 32bit process.
# just silently skip this value on Vista.
self.failUnlessRaises(pywintypes.error, GetGlobalMemory, 1)
def test_custom_mem(self):
test_data = str2bytes("hello\x00\xff")
test_buffer = array.array("b", test_data)
cf = RegisterClipboardFormat(custom_format_name)
self.failUnlessEqual(custom_format_name, GetClipboardFormatName(cf))
SetClipboardData(cf, test_buffer)
hglobal = GetClipboardDataHandle(cf)
data = GetGlobalMemory(hglobal)
self.failUnlessEqual(data, test_data)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,214 @@
"""Test pywin32's error semantics"""
import sys
import unittest
import win32api, win32file, pywintypes
import pythoncom
import winerror
class TestBase(unittest.TestCase):
def _testExceptionIndex(self, exc, index, expected):
# check the exception itself can be indexed if not py3k
if sys.version_info < (3,):
self.failUnlessEqual(exc[index], expected)
# and that exception.args can is the same.
self.failUnlessEqual(exc.args[index], expected)
class TestAPISimple(TestBase):
def _getInvalidHandleException(self):
try:
win32api.CloseHandle(1)
except win32api.error as exc:
return exc
self.fail("Didn't get invalid-handle exception.")
def testSimple(self):
self.assertRaises(pywintypes.error, win32api.CloseHandle, 1)
def testErrnoIndex(self):
exc = self._getInvalidHandleException()
self._testExceptionIndex(exc, 0, winerror.ERROR_INVALID_HANDLE)
def testFuncIndex(self):
exc = self._getInvalidHandleException()
self._testExceptionIndex(exc, 1, "CloseHandle")
def testMessageIndex(self):
exc = self._getInvalidHandleException()
expected = win32api.FormatMessage(winerror.ERROR_INVALID_HANDLE).rstrip()
self._testExceptionIndex(exc, 2, expected)
def testUnpack(self):
try:
win32api.CloseHandle(1)
self.fail("expected exception!")
except win32api.error as exc:
self.failUnlessEqual(exc.winerror, winerror.ERROR_INVALID_HANDLE)
self.failUnlessEqual(exc.funcname, "CloseHandle")
expected_msg = win32api.FormatMessage(winerror.ERROR_INVALID_HANDLE).rstrip()
self.failUnlessEqual(exc.strerror, expected_msg)
def testAsStr(self):
exc = self._getInvalidHandleException()
err_msg = win32api.FormatMessage(winerror.ERROR_INVALID_HANDLE).rstrip()
# early on the result actually *was* a tuple - it must always look like one
err_tuple = (winerror.ERROR_INVALID_HANDLE, 'CloseHandle', err_msg)
self.failUnlessEqual(str(exc), str(err_tuple))
def testAsTuple(self):
exc = self._getInvalidHandleException()
err_msg = win32api.FormatMessage(winerror.ERROR_INVALID_HANDLE).rstrip()
# early on the result actually *was* a tuple - it must be able to be one
err_tuple = (winerror.ERROR_INVALID_HANDLE, 'CloseHandle', err_msg)
if sys.version_info < (3,):
self.failUnlessEqual(tuple(exc), err_tuple)
else:
self.failUnlessEqual(exc.args, err_tuple)
def testClassName(self):
exc = self._getInvalidHandleException()
# The error class has always been named 'error'. That's not ideal :(
self.failUnlessEqual(exc.__class__.__name__, "error")
def testIdentity(self):
exc = self._getInvalidHandleException()
self.failUnless(exc.__class__ is pywintypes.error)
def testBaseClass(self):
self.failUnlessEqual(pywintypes.error.__bases__, (Exception,))
def testAttributes(self):
exc = self._getInvalidHandleException()
err_msg = win32api.FormatMessage(winerror.ERROR_INVALID_HANDLE).rstrip()
self.failUnlessEqual(exc.winerror, winerror.ERROR_INVALID_HANDLE)
self.failUnlessEqual(exc.strerror, err_msg)
self.failUnlessEqual(exc.funcname, 'CloseHandle')
# some tests for 'insane' args.
def testStrangeArgsNone(self):
try:
raise pywintypes.error()
self.fail("Expected exception")
except pywintypes.error as exc:
self.failUnlessEqual(exc.args, ())
self.failUnlessEqual(exc.winerror, None)
self.failUnlessEqual(exc.funcname, None)
self.failUnlessEqual(exc.strerror, None)
def testStrangeArgsNotEnough(self):
try:
raise pywintypes.error("foo")
self.fail("Expected exception")
except pywintypes.error as exc:
assert exc.args[0] == "foo"
# 'winerror' always args[0]
self.failUnlessEqual(exc.winerror, "foo")
self.failUnlessEqual(exc.funcname, None)
self.failUnlessEqual(exc.strerror, None)
def testStrangeArgsTooMany(self):
try:
raise pywintypes.error("foo", "bar", "you", "never", "kn", 0)
self.fail("Expected exception")
except pywintypes.error as exc:
self.failUnlessEqual(exc.args[0], "foo")
self.failUnlessEqual(exc.args[-1], 0)
self.failUnlessEqual(exc.winerror, "foo")
self.failUnlessEqual(exc.funcname, "bar")
self.failUnlessEqual(exc.strerror, "you")
class TestCOMSimple(TestBase):
def _getException(self):
try:
pythoncom.StgOpenStorage("foo", None, 0)
except pythoncom.com_error as exc:
return exc
self.fail("Didn't get storage exception.")
def testIs(self):
self.failUnless(pythoncom.com_error is pywintypes.com_error)
def testSimple(self):
self.assertRaises(pythoncom.com_error, pythoncom.StgOpenStorage, "foo", None, 0)
def testErrnoIndex(self):
exc = self._getException()
self._testExceptionIndex(exc, 0, winerror.STG_E_INVALIDFLAG)
def testMessageIndex(self):
exc = self._getException()
expected = win32api.FormatMessage(winerror.STG_E_INVALIDFLAG).rstrip()
self._testExceptionIndex(exc, 1, expected)
def testAsStr(self):
exc = self._getException()
err_msg = win32api.FormatMessage(winerror.STG_E_INVALIDFLAG).rstrip()
# early on the result actually *was* a tuple - it must always look like one
err_tuple = (winerror.STG_E_INVALIDFLAG, err_msg, None, None)
self.failUnlessEqual(str(exc), str(err_tuple))
def testAsTuple(self):
exc = self._getException()
err_msg = win32api.FormatMessage(winerror.STG_E_INVALIDFLAG).rstrip()
# early on the result actually *was* a tuple - it must be able to be one
err_tuple = (winerror.STG_E_INVALIDFLAG, err_msg, None, None)
if sys.version_info < (3,):
self.failUnlessEqual(tuple(exc), err_tuple)
else:
self.failUnlessEqual(exc.args, err_tuple)
def testClassName(self):
exc = self._getException()
self.failUnlessEqual(exc.__class__.__name__, "com_error")
def testIdentity(self):
exc = self._getException()
self.failUnless(exc.__class__ is pywintypes.com_error)
def testBaseClass(self):
exc = self._getException()
self.failUnlessEqual(pywintypes.com_error.__bases__, (Exception,))
def testAttributes(self):
exc = self._getException()
err_msg = win32api.FormatMessage(winerror.STG_E_INVALIDFLAG).rstrip()
self.failUnlessEqual(exc.hresult, winerror.STG_E_INVALIDFLAG)
self.failUnlessEqual(exc.strerror, err_msg)
self.failUnlessEqual(exc.argerror, None)
self.failUnlessEqual(exc.excepinfo, None)
def testStrangeArgsNone(self):
try:
raise pywintypes.com_error()
self.fail("Expected exception")
except pywintypes.com_error as exc:
self.failUnlessEqual(exc.args, ())
self.failUnlessEqual(exc.hresult, None)
self.failUnlessEqual(exc.strerror, None)
self.failUnlessEqual(exc.argerror, None)
self.failUnlessEqual(exc.excepinfo, None)
def testStrangeArgsNotEnough(self):
try:
raise pywintypes.com_error("foo")
self.fail("Expected exception")
except pywintypes.com_error as exc:
self.failUnlessEqual(exc.args[0], "foo")
self.failUnlessEqual(exc.hresult, "foo")
self.failUnlessEqual(exc.strerror, None)
self.failUnlessEqual(exc.excepinfo, None)
self.failUnlessEqual(exc.argerror, None)
def testStrangeArgsTooMany(self):
try:
raise pywintypes.com_error("foo", "bar", "you", "never", "kn", 0)
self.fail("Expected exception")
except pywintypes.com_error as exc:
self.failUnlessEqual(exc.args[0], "foo")
self.failUnlessEqual(exc.args[-1], 0)
self.failUnlessEqual(exc.hresult, "foo")
self.failUnlessEqual(exc.strerror, "bar")
self.failUnlessEqual(exc.excepinfo, "you")
self.failUnlessEqual(exc.argerror, "never")
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,192 @@
# odbc test suite kindly contributed by Frank Millman.
import sys
import os
import unittest
import odbc
import tempfile
from pywin32_testutil import str2bytes, str2memory, TestSkipped
# We use the DAO ODBC driver
from win32com.client.gencache import EnsureDispatch
from win32com.client import constants
import pythoncom
class TestStuff(unittest.TestCase):
def setUp(self):
self.tablename = "pywin32test_users"
self.db_filename = None
self.conn = self.cur = None
try:
# Test any database if a connection string is supplied...
conn_str = os.environ['TEST_ODBC_CONNECTION_STRING']
except KeyError:
# Create a local MSAccess DB for testing.
self.db_filename = tempfile.NamedTemporaryFile().name + '.mdb'
# Create a brand-new database - what is the story with these?
for suffix in (".36", ".35", ".30"):
try:
dbe = EnsureDispatch("DAO.DBEngine" + suffix)
break
except pythoncom.com_error:
pass
else:
raise TestSkipped("Can't find a DB engine")
workspace = dbe.Workspaces(0)
newdb = workspace.CreateDatabase(self.db_filename,
constants.dbLangGeneral,
constants.dbEncrypt)
newdb.Close()
conn_str = "Driver={Microsoft Access Driver (*.mdb)};dbq=%s;Uid=;Pwd=;" \
% (self.db_filename,)
## print 'Connection string:', conn_str
self.conn = odbc.odbc(conn_str)
# And we expect a 'users' table for these tests.
self.cur = self.conn.cursor()
## self.cur.setoutputsize(1000)
try:
self.cur.execute("""drop table %s""" %self.tablename)
except (odbc.error, odbc.progError):
pass
## This needs to be adjusted for sql server syntax for unicode fields
## - memo -> TEXT
## - varchar -> nvarchar
self.assertEqual(self.cur.execute(
"""create table %s (
userid varchar(25),
username varchar(25),
bitfield bit,
intfield integer,
floatfield float,
datefield datetime,
rawfield varbinary(100),
longtextfield memo,
longbinaryfield image
)""" %self.tablename),-1)
def tearDown(self):
if self.cur is not None:
try:
self.cur.execute("""drop table %s""" %self.tablename)
except (odbc.error, odbc.progError) as why:
print("Failed to delete test table %s" %self.tablename, why)
self.cur.close()
self.cur = None
if self.conn is not None:
self.conn.close()
self.conn = None
if self.db_filename is not None:
try:
os.unlink(self.db_filename)
except OSError:
pass
def test_insert_select(self, userid='Frank', username='Frank Millman'):
self.assertEqual(self.cur.execute("insert into %s (userid, username) \
values (?,?)" %self.tablename, [userid, username]),1)
self.assertEqual(self.cur.execute("select * from %s \
where userid = ?" %self.tablename, [userid.lower()]),0)
self.assertEqual(self.cur.execute("select * from %s \
where username = ?" %self.tablename, [username.lower()]),0)
def test_insert_select_unicode(self, userid='Frank', username="Frank Millman"):
self.assertEqual(self.cur.execute("insert into %s (userid, username)\
values (?,?)" %self.tablename, [userid, username]),1)
self.assertEqual(self.cur.execute("select * from %s \
where userid = ?" %self.tablename, [userid.lower()]),0)
self.assertEqual(self.cur.execute("select * from %s \
where username = ?" %self.tablename, [username.lower()]),0)
def test_insert_select_unicode_ext(self):
userid = "t-\xe0\xf2"
username = "test-\xe0\xf2 name"
self.test_insert_select_unicode(userid, username)
def _test_val(self, fieldName, value):
for x in range(100):
self.cur.execute("delete from %s where userid='Frank'" %self.tablename)
self.assertEqual(self.cur.execute(
"insert into %s (userid, %s) values (?,?)" % (self.tablename, fieldName),
["Frank", value]), 1)
self.cur.execute("select %s from %s where userid = ?" % (fieldName, self.tablename),
["Frank"])
rows = self.cur.fetchmany()
self.failUnlessEqual(1, len(rows))
row = rows[0]
self.failUnlessEqual(row[0], value)
def testBit(self):
self._test_val('bitfield', 1)
self._test_val('bitfield', 0)
def testInt(self):
self._test_val('intfield', 1)
self._test_val('intfield', 0)
try:
big = sys.maxsize
except AttributeError:
big = sys.maxint
self._test_val('intfield', big)
def testFloat(self):
self._test_val('floatfield', 1.01)
self._test_val('floatfield', 0)
def testVarchar(self, ):
self._test_val('username', 'foo')
def testLongVarchar(self):
""" Test a long text field in excess of internal cursor data size (65536)"""
self._test_val('longtextfield', 'abc' * 70000)
def testLongBinary(self):
""" Test a long raw field in excess of internal cursor data size (65536)"""
self._test_val('longbinaryfield', str2memory('\0\1\2' * 70000))
def testRaw(self):
## Test binary data
self._test_val('rawfield', str2memory('\1\2\3\4\0\5\6\7\8'))
def test_widechar(self):
"""Test a unicode character that would be mangled if bound as plain character.
For example, previously the below was returned as ascii 'a'
"""
self._test_val('username', '\u0101')
def testDates(self):
import datetime
for v in (
(1900, 12, 25, 23, 39, 59),
):
d = datetime.datetime(*v)
self._test_val('datefield', d)
def test_set_nonzero_length(self):
self.assertEqual(self.cur.execute("insert into %s (userid,username) "
"values (?,?)" %self.tablename, ['Frank', 'Frank Millman']),1)
self.assertEqual(self.cur.execute("update %s set username = ?" %self.tablename,
['Frank']),1)
self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
self.assertEqual(len(self.cur.fetchone()[1]),5)
def test_set_zero_length(self):
self.assertEqual(self.cur.execute("insert into %s (userid,username) "
"values (?,?)" %self.tablename, [str2bytes('Frank'), '']),1)
self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
self.assertEqual(len(self.cur.fetchone()[1]),0)
def test_set_zero_length_unicode(self):
self.assertEqual(self.cur.execute("insert into %s (userid,username) "
"values (?,?)" %self.tablename, ['Frank', '']),1)
self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
self.assertEqual(len(self.cur.fetchone()[1]),0)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,113 @@
import sys
import unittest
import pywintypes
import time
from pywin32_testutil import str2bytes, ob2memory
import datetime
import operator
class TestCase(unittest.TestCase):
def testPyTimeFormat(self):
struct_current = time.localtime()
pytime_current = pywintypes.Time(struct_current)
# try and test all the standard parts of the format
# Note we used to include '%Z' testing, but that was pretty useless as
# it always returned the local timezone.
format_strings = "%a %A %b %B %c %d %H %I %j %m %M %p %S %U %w %W %x %X %y %Y"
for fmt in format_strings.split():
v1 = pytime_current.Format(fmt)
v2 = time.strftime(fmt, struct_current)
self.assertEquals(v1, v2, "format %s failed - %r != %r" % (fmt, v1, v2))
def testPyTimePrint(self):
# This used to crash with an invalid, or too early time.
# We don't really want to check that it does cause a ValueError
# (as hopefully this wont be true forever). So either working, or
# ValueError is OK.
try:
t = pywintypes.Time(-2)
t.Format()
except ValueError:
return
def testTimeInDict(self):
d = {}
d['t1'] = pywintypes.Time(1)
self.failUnlessEqual(d['t1'], pywintypes.Time(1))
def testPyTimeCompare(self):
t1 = pywintypes.Time(100)
t1_2 = pywintypes.Time(100)
t2 = pywintypes.Time(101)
self.failUnlessEqual(t1, t1_2)
self.failUnless(t1 <= t1_2)
self.failUnless(t1_2 >= t1)
self.failIfEqual(t1, t2)
self.failUnless(t1 < t2)
self.failUnless(t2 > t1 )
def testPyTimeCompareOther(self):
t1 = pywintypes.Time(100)
t2 = None
self.failIfEqual(t1, t2)
def testTimeTuple(self):
now = datetime.datetime.now() # has usec...
# timetuple() lost usec - pt must be <=...
pt = pywintypes.Time(now.timetuple())
# *sob* - only if we have a datetime object can we compare like this.
if isinstance(pt, datetime.datetime):
self.failUnless(pt <= now)
def testTimeTuplems(self):
now = datetime.datetime.now() # has usec...
tt = now.timetuple() + (now.microsecond // 1000,)
pt = pywintypes.Time(tt)
# we can't compare if using the old type, as it loses all sub-second res.
if isinstance(pt, datetime.datetime):
self.failUnlessEqual(now, pt)
def testPyTimeFromTime(self):
t1 = pywintypes.Time(time.time())
self.failUnless(pywintypes.Time(t1) is t1)
def testPyTimeTooLarge(self):
# We only do this special thing for python 3.x
if not issubclass(pywintypes.TimeType, datetime.datetime):
return
MAX_TIMESTAMP = 0x7FFFFFFFFFFFFFFF # used by some API function to mean "never"
ts = pywintypes.TimeStamp(MAX_TIMESTAMP)
self.failUnlessEqual(ts, datetime.datetime.max)
def testGUID(self):
s = "{00020400-0000-0000-C000-000000000046}"
iid = pywintypes.IID(s)
iid2 = pywintypes.IID(ob2memory(iid), True)
self.assertEquals(iid, iid2)
self.assertRaises(ValueError, pywintypes.IID, str2bytes('00'), True) # too short
self.assertRaises(TypeError, pywintypes.IID, 0, True) # no buffer
def testGUIDRichCmp(self):
s = "{00020400-0000-0000-C000-000000000046}"
iid = pywintypes.IID(s)
self.failIf(s==None)
self.failIf(None==s)
self.failUnless(s!=None)
self.failUnless(None!=s)
if sys.version_info > (3,0):
self.assertRaises(TypeError, operator.gt, None, s)
self.assertRaises(TypeError, operator.gt, s, None)
self.assertRaises(TypeError, operator.lt, None, s)
self.assertRaises(TypeError, operator.lt, s, None)
def testGUIDInDict(self):
s = "{00020400-0000-0000-C000-000000000046}"
iid = pywintypes.IID(s)
d = dict(item=iid)
self.failUnlessEqual(d['item'], iid)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,127 @@
# Tests for the win32security module.
import sys, os
import unittest
import winerror
from pywin32_testutil import testmain, TestSkipped, ob2memory
import win32api, win32con, win32security, ntsecuritycon
class SecurityTests(unittest.TestCase):
def setUp(self):
self.pwr_sid=win32security.LookupAccountName('','Power Users')[0]
self.admin_sid=win32security.LookupAccountName('','Administrator')[0]
def tearDown(self):
pass
def testEqual(self):
self.failUnlessEqual(win32security.LookupAccountName('','Administrator')[0],
win32security.LookupAccountName('','Administrator')[0])
def testNESID(self):
self.failUnless(self.pwr_sid==self.pwr_sid)
self.failUnless(self.pwr_sid!=self.admin_sid)
def testNEOther(self):
self.failUnless(self.pwr_sid!=None)
self.failUnless(None!=self.pwr_sid)
self.failIf(self.pwr_sid==None)
self.failIf(None==self.pwr_sid)
self.failIfEqual(None, self.pwr_sid)
def testSIDInDict(self):
d = dict(foo=self.pwr_sid)
self.failUnlessEqual(d['foo'], self.pwr_sid)
def testBuffer(self):
self.failUnlessEqual(ob2memory(win32security.LookupAccountName('','Administrator')[0]),
ob2memory(win32security.LookupAccountName('','Administrator')[0]))
def testMemory(self):
pwr_sid = self.pwr_sid
admin_sid = self.admin_sid
sd1=win32security.SECURITY_DESCRIPTOR()
sd2=win32security.SECURITY_DESCRIPTOR()
sd3=win32security.SECURITY_DESCRIPTOR()
dacl=win32security.ACL()
dacl.AddAccessAllowedAce(win32security.ACL_REVISION,win32con.GENERIC_READ,pwr_sid)
dacl.AddAccessAllowedAce(win32security.ACL_REVISION,win32con.GENERIC_ALL,admin_sid)
sd4=win32security.SECURITY_DESCRIPTOR()
sacl=win32security.ACL()
sacl.AddAuditAccessAce(win32security.ACL_REVISION,win32con.DELETE,admin_sid,1,1)
sacl.AddAuditAccessAce(win32security.ACL_REVISION,win32con.GENERIC_ALL,pwr_sid,1,1)
for x in range(0,200000):
sd1.SetSecurityDescriptorOwner(admin_sid,0)
sd2.SetSecurityDescriptorGroup(pwr_sid,0)
sd3.SetSecurityDescriptorDacl(1,dacl,0)
sd4.SetSecurityDescriptorSacl(1,sacl,0)
class DomainTests(unittest.TestCase):
def setUp(self):
self.ds_handle = None
try:
# saving the handle means the other test itself should bind faster.
self.ds_handle = win32security.DsBind()
except win32security.error as exc:
if exc.winerror != winerror.ERROR_NO_SUCH_DOMAIN:
raise
raise TestSkipped(exc)
def tearDown(self):
if self.ds_handle is not None:
self.ds_handle.close()
class TestDS(DomainTests):
def testDsGetDcName(self):
# Not sure what we can actually test here! At least calling it
# does something :)
win32security.DsGetDcName()
def testDsListServerInfo(self):
# again, not checking much, just exercising the code.
h=win32security.DsBind()
for (status, ignore, site) in win32security.DsListSites(h):
for (status, ignore, server) in win32security.DsListServersInSite(h, site):
info = win32security.DsListInfoForServer(h, server)
for (status, ignore, domain) in win32security.DsListDomainsInSite(h, site):
pass
def testDsCrackNames(self):
h = win32security.DsBind()
fmt_offered = ntsecuritycon.DS_FQDN_1779_NAME
name = win32api.GetUserNameEx(fmt_offered)
result = win32security.DsCrackNames(h, 0, fmt_offered, fmt_offered, (name,))
self.failUnlessEqual(name, result[0][2])
def testDsCrackNamesSyntax(self):
# Do a syntax check only - that allows us to avoid binding.
# But must use DS_CANONICAL_NAME (or _EX)
expected = win32api.GetUserNameEx(win32api.NameCanonical)
fmt_offered = ntsecuritycon.DS_FQDN_1779_NAME
name = win32api.GetUserNameEx(fmt_offered)
result = win32security.DsCrackNames(None, ntsecuritycon.DS_NAME_FLAG_SYNTACTICAL_ONLY,
fmt_offered, ntsecuritycon.DS_CANONICAL_NAME,
(name,))
self.failUnlessEqual(expected, result[0][2])
class TestTranslate(DomainTests):
def _testTranslate(self, fmt_from, fmt_to):
name = win32api.GetUserNameEx(fmt_from)
expected = win32api.GetUserNameEx(fmt_to)
got = win32security.TranslateName(name, fmt_from, fmt_to)
self.failUnlessEqual(got, expected)
def testTranslate1(self):
self._testTranslate(win32api.NameFullyQualifiedDN, win32api.NameSamCompatible)
def testTranslate2(self):
self._testTranslate(win32api.NameSamCompatible, win32api.NameFullyQualifiedDN)
def testTranslate3(self):
self._testTranslate(win32api.NameFullyQualifiedDN, win32api.NameUniqueId)
def testTranslate4(self):
self._testTranslate(win32api.NameUniqueId, win32api.NameFullyQualifiedDN)
if __name__=='__main__':
testmain()

View file

@ -0,0 +1,192 @@
# Some tests of the win32security sspi functions.
# Stolen from Roger's original test_sspi.c, a version of which is in "Demos"
# See also the other SSPI demos.
import re
import win32security, sspi, sspicon, win32api
from pywin32_testutil import TestSkipped, testmain, str2bytes
import unittest
# It is quite likely that the Kerberos tests will fail due to not being
# installed. The NTLM tests do *not* get the same behaviour as they should
# always be there.
def applyHandlingSkips(func, *args):
try:
return func(*args)
except win32api.error as exc:
if exc.winerror == sspicon.SEC_E_NO_CREDENTIALS:
raise TestSkipped(exc)
raise
class TestSSPI(unittest.TestCase):
def assertRaisesHRESULT(self, hr, func, *args):
try:
return func(*args)
raise RuntimeError("expecting %s failure" % (hr,))
except win32security.error as exc:
self.failUnlessEqual(exc.winerror, hr)
def _doAuth(self, pkg_name):
sspiclient=sspi.ClientAuth(pkg_name,targetspn=win32api.GetUserName())
sspiserver=sspi.ServerAuth(pkg_name)
sec_buffer=None
err = 1
while err != 0:
err, sec_buffer = sspiclient.authorize(sec_buffer)
err, sec_buffer = sspiserver.authorize(sec_buffer)
return sspiclient, sspiserver
def _doTestImpersonate(self, pkg_name):
# Just for the sake of code exercising!
sspiclient, sspiserver = self._doAuth(pkg_name)
sspiserver.ctxt.ImpersonateSecurityContext()
sspiserver.ctxt.RevertSecurityContext()
def testImpersonateKerberos(self):
applyHandlingSkips(self._doTestImpersonate, "Kerberos")
def testImpersonateNTLM(self):
self._doTestImpersonate("NTLM")
def _doTestEncrypt(self, pkg_name):
sspiclient, sspiserver = self._doAuth(pkg_name)
pkg_size_info=sspiclient.ctxt.QueryContextAttributes(sspicon.SECPKG_ATTR_SIZES)
msg=str2bytes('some data to be encrypted ......')
trailersize=pkg_size_info['SecurityTrailer']
encbuf=win32security.PySecBufferDescType()
encbuf.append(win32security.PySecBufferType(len(msg), sspicon.SECBUFFER_DATA))
encbuf.append(win32security.PySecBufferType(trailersize, sspicon.SECBUFFER_TOKEN))
encbuf[0].Buffer=msg
sspiclient.ctxt.EncryptMessage(0,encbuf,1)
sspiserver.ctxt.DecryptMessage(encbuf,1)
self.failUnlessEqual(msg, encbuf[0].Buffer)
# and test the higher-level functions
data_in = str2bytes("hello")
data, sig = sspiclient.encrypt(data_in)
self.assertEqual(sspiserver.decrypt(data, sig), data_in)
data, sig = sspiserver.encrypt(data_in)
self.assertEqual(sspiclient.decrypt(data, sig), data_in)
def _doTestEncryptStream(self, pkg_name):
# Test out the SSPI/GSSAPI interop wrapping examples at
# https://docs.microsoft.com/en-us/windows/win32/secauthn/sspi-kerberos-interoperability-with-gssapi
sspiclient, sspiserver = self._doAuth(pkg_name)
pkg_size_info=sspiclient.ctxt.QueryContextAttributes(sspicon.SECPKG_ATTR_SIZES)
msg=str2bytes('some data to be encrypted ......')
trailersize=pkg_size_info['SecurityTrailer']
blocksize=pkg_size_info['BlockSize']
encbuf=win32security.PySecBufferDescType()
encbuf.append(win32security.PySecBufferType(trailersize, sspicon.SECBUFFER_TOKEN))
encbuf.append(win32security.PySecBufferType(len(msg), sspicon.SECBUFFER_DATA))
encbuf.append(win32security.PySecBufferType(blocksize, sspicon.SECBUFFER_PADDING))
encbuf[1].Buffer=msg
sspiclient.ctxt.EncryptMessage(0,encbuf,1)
encmsg = encbuf[0].Buffer + encbuf[1].Buffer + encbuf[2].Buffer
decbuf=win32security.PySecBufferDescType()
decbuf.append(win32security.PySecBufferType(len(encmsg), sspicon.SECBUFFER_STREAM))
decbuf.append(win32security.PySecBufferType(0, sspicon.SECBUFFER_DATA))
decbuf[0].Buffer = encmsg
sspiserver.ctxt.DecryptMessage(decbuf,1)
self.failUnlessEqual(msg, decbuf[1].Buffer)
def testEncryptNTLM(self):
self._doTestEncrypt("NTLM")
def testEncryptStreamNTLM(self):
self._doTestEncryptStream("NTLM")
def testEncryptKerberos(self):
applyHandlingSkips(self._doTestEncrypt, "Kerberos")
def testEncryptStreamKerberos(self):
applyHandlingSkips(self._doTestEncryptStream, "Kerberos")
def _doTestSign(self, pkg_name):
sspiclient, sspiserver = self._doAuth(pkg_name)
pkg_size_info=sspiclient.ctxt.QueryContextAttributes(sspicon.SECPKG_ATTR_SIZES)
msg=str2bytes('some data to be encrypted ......')
sigsize=pkg_size_info['MaxSignature']
sigbuf=win32security.PySecBufferDescType()
sigbuf.append(win32security.PySecBufferType(len(msg), sspicon.SECBUFFER_DATA))
sigbuf.append(win32security.PySecBufferType(sigsize, sspicon.SECBUFFER_TOKEN))
sigbuf[0].Buffer=msg
sspiclient.ctxt.MakeSignature(0,sigbuf,0)
sspiserver.ctxt.VerifySignature(sigbuf,0)
# and test the higher-level functions
sspiclient.next_seq_num = 1
sspiserver.next_seq_num = 1
data = str2bytes("hello")
key = sspiclient.sign(data)
sspiserver.verify(data, key)
key = sspiclient.sign(data)
self.assertRaisesHRESULT(sspicon.SEC_E_MESSAGE_ALTERED,
sspiserver.verify, data + data, key)
# and the other way
key = sspiserver.sign(data)
sspiclient.verify(data, key)
key = sspiserver.sign(data)
self.assertRaisesHRESULT(sspicon.SEC_E_MESSAGE_ALTERED,
sspiclient.verify, data + data, key)
def testSignNTLM(self):
self._doTestSign("NTLM")
def testSignKerberos(self):
applyHandlingSkips(self._doTestSign, "Kerberos")
def _testSequenceSign(self):
# Only Kerberos supports sequence detection.
sspiclient, sspiserver = self._doAuth("Kerberos")
key = sspiclient.sign("hello")
sspiclient.sign("hello")
self.assertRaisesHRESULT(sspicon.SEC_E_OUT_OF_SEQUENCE,
sspiserver.verify, 'hello', key)
def testSequenceSign(self):
applyHandlingSkips(self._testSequenceSign)
def _testSequenceEncrypt(self):
# Only Kerberos supports sequence detection.
sspiclient, sspiserver = self._doAuth("Kerberos")
blob, key = sspiclient.encrypt("hello",)
blob, key = sspiclient.encrypt("hello")
self.assertRaisesHRESULT(sspicon.SEC_E_OUT_OF_SEQUENCE,
sspiserver.decrypt, blob, key)
def testSequenceEncrypt(self):
applyHandlingSkips(self._testSequenceEncrypt)
def testSecBufferRepr(self):
desc = win32security.PySecBufferDescType()
assert re.match('PySecBufferDesc\(ulVersion: 0 \| cBuffers: 0 \| pBuffers: 0x[\da-fA-F]{8,16}\)', repr(desc))
buffer1 = win32security.PySecBufferType(0, sspicon.SECBUFFER_TOKEN)
assert re.match('PySecBuffer\(cbBuffer: 0 \| BufferType: 2 \| pvBuffer: 0x[\da-fA-F]{8,16}\)', repr(buffer1))
'PySecBuffer(cbBuffer: 0 | BufferType: 2 | pvBuffer: 0x000001B8CC6D8020)'
desc.append(buffer1)
assert re.match('PySecBufferDesc\(ulVersion: 0 \| cBuffers: 1 \| pBuffers: 0x[\da-fA-F]{8,16}\)', repr(desc))
buffer2 = win32security.PySecBufferType(4, sspicon.SECBUFFER_DATA)
assert re.match('PySecBuffer\(cbBuffer: 4 \| BufferType: 1 \| pvBuffer: 0x[\da-fA-F]{8,16}\)', repr(buffer2))
desc.append(buffer2)
assert re.match('PySecBufferDesc\(ulVersion: 0 \| cBuffers: 2 \| pBuffers: 0x[\da-fA-F]{8,16}\)', repr(desc))
if __name__=='__main__':
testmain()

View file

@ -0,0 +1,203 @@
# General test module for win32api - please add some :)
import unittest
from pywin32_testutil import str2bytes
import win32api, win32con, win32event, winerror
import sys, os
import tempfile
import datetime
class CurrentUserTestCase(unittest.TestCase):
def testGetCurrentUser(self):
name = "%s\\%s" % (win32api.GetDomainName(), win32api.GetUserName())
self.failUnless(name == win32api.GetUserNameEx(win32api.NameSamCompatible))
class TestTime(unittest.TestCase):
def testTimezone(self):
# GetTimeZoneInformation
rc, tzinfo = win32api.GetTimeZoneInformation()
if rc == win32con.TIME_ZONE_ID_DAYLIGHT:
tz_str = tzinfo[4]
tz_time = tzinfo[5]
else:
tz_str = tzinfo[1]
tz_time = tzinfo[2]
# for the sake of code exercise but don't output
tz_str.encode()
if not isinstance(tz_time, datetime.datetime):
tz_time.Format()
def TestDateFormat(self):
DATE_LONGDATE = 2
date_flags = DATE_LONGDATE
win32api.GetDateFormat(0, date_flags, None)
win32api.GetDateFormat(0, date_flags, 0)
win32api.GetDateFormat(0, date_flags, datetime.datetime.now())
win32api.GetDateFormat(0, date_flags, time.time())
def TestTimeFormat(self):
win32api.GetTimeFormat(0, 0, None)
win32api.GetTimeFormat(0, 0, 0)
win32api.GetTimeFormat(0, 0, datetime.datetime.now())
win32api.GetTimeFormat(0, 0, time.time())
class Registry(unittest.TestCase):
key_name = r'PythonTestHarness\Whatever'
def test1(self):
# This used to leave a stale exception behind.
def reg_operation():
hkey = win32api.RegCreateKey(win32con.HKEY_CURRENT_USER, self.key_name)
x = 3/0 # or a statement like: raise 'error'
# do the test
try:
try:
try:
reg_operation()
except:
1/0 # Force exception
finally:
win32api.RegDeleteKey(win32con.HKEY_CURRENT_USER, self.key_name)
except ZeroDivisionError:
pass
def testValues(self):
key_name = r'PythonTestHarness\win32api'
## tuples containing value name, value type, data
values=(
(None, win32con.REG_SZ, 'This is default unnamed value'),
('REG_SZ', win32con.REG_SZ,'REG_SZ text data'),
('REG_EXPAND_SZ', win32con.REG_EXPAND_SZ, '%systemdir%'),
## REG_MULTI_SZ value needs to be a list since strings are returned as a list
('REG_MULTI_SZ', win32con.REG_MULTI_SZ, ['string 1','string 2','string 3','string 4']),
('REG_MULTI_SZ_empty', win32con.REG_MULTI_SZ, []),
('REG_DWORD', win32con.REG_DWORD, 666),
('REG_QWORD_INT', win32con.REG_QWORD, 99),
('REG_QWORD', win32con.REG_QWORD, 2**33),
('REG_BINARY', win32con.REG_BINARY, str2bytes('\x00\x01\x02\x03\x04\x05\x06\x07\x08\x01\x00')),
)
hkey = win32api.RegCreateKey(win32con.HKEY_CURRENT_USER, key_name)
for value_name, reg_type, data in values:
win32api.RegSetValueEx(hkey, value_name, None, reg_type, data)
for value_name, orig_type, orig_data in values:
data, typ=win32api.RegQueryValueEx(hkey, value_name)
self.assertEqual(typ, orig_type)
self.assertEqual(data, orig_data)
def testNotifyChange(self):
def change():
hkey = win32api.RegCreateKey(win32con.HKEY_CURRENT_USER, self.key_name)
try:
win32api.RegSetValue(hkey, None, win32con.REG_SZ, "foo")
finally:
win32api.RegDeleteKey(win32con.HKEY_CURRENT_USER, self.key_name)
evt = win32event.CreateEvent(None,0,0,None)
## REG_NOTIFY_CHANGE_LAST_SET - values
## REG_CHANGE_NOTIFY_NAME - keys
## REG_NOTIFY_CHANGE_SECURITY - security descriptor
## REG_NOTIFY_CHANGE_ATTRIBUTES
win32api.RegNotifyChangeKeyValue(win32con.HKEY_CURRENT_USER,1,win32api.REG_NOTIFY_CHANGE_LAST_SET,evt,True)
ret_code=win32event.WaitForSingleObject(evt,0)
# Should be no change.
self.failUnless(ret_code==win32con.WAIT_TIMEOUT)
change()
# Our event should now be in a signalled state.
ret_code=win32event.WaitForSingleObject(evt,0)
self.failUnless(ret_code==win32con.WAIT_OBJECT_0)
class FileNames(unittest.TestCase):
def testShortLongPathNames(self):
try:
me = __file__
except NameError:
me = sys.argv[0]
fname = os.path.abspath(me).lower()
short_name = win32api.GetShortPathName(fname).lower()
long_name = win32api.GetLongPathName(short_name).lower()
self.failUnless(long_name==fname, \
"Expected long name ('%s') to be original name ('%s')" % (long_name, fname))
self.failUnlessEqual(long_name, win32api.GetLongPathNameW(short_name).lower())
long_name = win32api.GetLongPathNameW(short_name).lower()
self.failUnless(type(long_name)==str, "GetLongPathNameW returned type '%s'" % (type(long_name),))
self.failUnless(long_name==fname, \
"Expected long name ('%s') to be original name ('%s')" % (long_name, fname))
def testShortUnicodeNames(self):
try:
me = __file__
except NameError:
me = sys.argv[0]
fname = os.path.abspath(me).lower()
# passing unicode should cause GetShortPathNameW to be called.
short_name = win32api.GetShortPathName(str(fname)).lower()
self.failUnless(isinstance(short_name, str))
long_name = win32api.GetLongPathName(short_name).lower()
self.failUnless(long_name==fname, \
"Expected long name ('%s') to be original name ('%s')" % (long_name, fname))
self.failUnlessEqual(long_name, win32api.GetLongPathNameW(short_name).lower())
long_name = win32api.GetLongPathNameW(short_name).lower()
self.failUnless(type(long_name)==str, "GetLongPathNameW returned type '%s'" % (type(long_name),))
self.failUnless(long_name==fname, \
"Expected long name ('%s') to be original name ('%s')" % (long_name, fname))
def testLongLongPathNames(self):
# We need filename where the FQN is > 256 - simplest way is to create a
# 250 character directory in the cwd (except - cwd may be on a drive
# not supporting \\\\?\\ (eg, network share) - so use temp.
import win32file
basename = "a" * 250
# but we need to ensure we use the 'long' version of the
# temp dir for later comparison.
long_temp_dir = win32api.GetLongPathNameW(tempfile.gettempdir())
fname = "\\\\?\\" + os.path.join(long_temp_dir, basename)
try:
win32file.CreateDirectoryW(fname, None)
except win32api.error as details:
if details.winerror!=winerror.ERROR_ALREADY_EXISTS:
raise
try:
# GetFileAttributes automatically calls GetFileAttributesW when
# passed unicode
try:
attr = win32api.GetFileAttributes(fname)
except win32api.error as details:
if details.winerror != winerror.ERROR_FILENAME_EXCED_RANGE:
raise
attr = win32api.GetFileAttributes(str(fname))
self.failUnless(attr & win32con.FILE_ATTRIBUTE_DIRECTORY, attr)
long_name = win32api.GetLongPathNameW(fname)
self.failUnlessEqual(long_name.lower(), fname.lower())
finally:
win32file.RemoveDirectory(fname)
class FormatMessage(unittest.TestCase):
def test_FromString(self):
msg = "Hello %1, how are you %2?"
inserts = ["Mark", "today"]
result = win32api.FormatMessage(win32con.FORMAT_MESSAGE_FROM_STRING,
msg, # source
0, # ID
0, # LangID
inserts)
self.assertEqual(result, "Hello Mark, how are you today?")
class Misc(unittest.TestCase):
def test_last_error(self):
for x in (0, 1, -1, winerror.TRUST_E_PROVIDER_UNKNOWN):
win32api.SetLastError(x)
self.failUnlessEqual(x, win32api.GetLastError())
def testVkKeyScan(self):
# hopefully ' ' doesn't depend on the locale!
self.failUnlessEqual(win32api.VkKeyScan(' '), 32)
def testVkKeyScanEx(self):
# hopefully ' ' doesn't depend on the locale!
self.failUnlessEqual(win32api.VkKeyScanEx(' ', 0), 32)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,32 @@
# Test module for win32crypt
import unittest
import win32crypt
from pywin32_testutil import str2bytes # py3k-friendly helper
class Crypt(unittest.TestCase):
def testSimple(self):
data = str2bytes("My test data")
entropy = None
desc = "My description"
flags = 0
ps = None
blob = win32crypt.CryptProtectData(data, desc, entropy, None, ps, flags)
got_desc, got_data = win32crypt.CryptUnprotectData(blob, entropy, None, ps, flags)
self.failUnlessEqual(data, got_data)
self.failUnlessEqual(desc, got_desc)
def testEntropy(self):
data = str2bytes("My test data")
entropy = str2bytes("My test entropy")
desc = "My description"
flags = 0
ps = None
blob = win32crypt.CryptProtectData(data, desc, entropy, None, ps, flags)
got_desc, got_data = win32crypt.CryptUnprotectData(blob, entropy, None, ps, flags)
self.failUnlessEqual(data, got_data)
self.failUnlessEqual(desc, got_desc)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,115 @@
import unittest
import win32event
import pywintypes
import time
import os
import sys
from pywin32_testutil import int2long
class TestWaitableTimer(unittest.TestCase):
def testWaitableFireLong(self):
h = win32event.CreateWaitableTimer(None, 0, None)
dt = int2long(-160) # 160 ns.
win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
rc = win32event.WaitForSingleObject(h, 1000)
self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)
def testWaitableFire(self):
h = win32event.CreateWaitableTimer(None, 0, None)
dt = -160 # 160 ns.
win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
rc = win32event.WaitForSingleObject(h, 1000)
self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)
def testWaitableTrigger(self):
h = win32event.CreateWaitableTimer(None, 0, None)
# for the sake of this, pass a long that doesn't fit in an int.
dt = -2000000000
win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
rc = win32event.WaitForSingleObject(h, 10) # 10 ms.
self.failUnlessEqual(rc, win32event.WAIT_TIMEOUT)
def testWaitableError(self):
h = win32event.CreateWaitableTimer(None, 0, None)
h.close()
self.assertRaises(pywintypes.error, win32event.SetWaitableTimer,
h, -42, 0, None, None, 0)
class TestWaitFunctions(unittest.TestCase):
def testMsgWaitForMultipleObjects(self):
# this function used to segfault when called with an empty list
res = win32event.MsgWaitForMultipleObjects([], 0, 0, 0)
self.assertEquals(res, win32event.WAIT_TIMEOUT)
def testMsgWaitForMultipleObjects2(self):
# test with non-empty list
event = win32event.CreateEvent(None, 0, 0, None)
res = win32event.MsgWaitForMultipleObjects([event], 0, 0, 0)
self.assertEquals(res, win32event.WAIT_TIMEOUT)
def testMsgWaitForMultipleObjectsEx(self):
# this function used to segfault when called with an empty list
res = win32event.MsgWaitForMultipleObjectsEx([], 0, 0, 0)
self.assertEquals(res, win32event.WAIT_TIMEOUT)
def testMsgWaitForMultipleObjectsEx2(self):
# test with non-empty list
event = win32event.CreateEvent(None, 0, 0, None)
res = win32event.MsgWaitForMultipleObjectsEx([event], 0, 0, 0)
self.assertEquals(res, win32event.WAIT_TIMEOUT)
class TestEvent(unittest.TestCase):
def assertSignaled(self, event):
self.assertEquals(win32event.WaitForSingleObject(event, 0),
win32event.WAIT_OBJECT_0)
def assertNotSignaled(self, event):
self.assertEquals(win32event.WaitForSingleObject(event, 0),
win32event.WAIT_TIMEOUT)
def testCreateEvent(self):
event = win32event.CreateEvent(None, False, False, None)
self.assertNotSignaled(event)
event = win32event.CreateEvent(None, False, True, None)
self.assertSignaled(event)
self.assertNotSignaled(event)
event = win32event.CreateEvent(None, True, True, None)
self.assertSignaled(event)
self.assertSignaled(event)
def testSetEvent(self):
event = win32event.CreateEvent(None, True, False, None)
self.assertNotSignaled(event)
res = win32event.SetEvent(event)
self.assertEquals(res, None)
self.assertSignaled(event)
event.close()
self.assertRaises(pywintypes.error, win32event.SetEvent, event)
def testResetEvent(self):
event = win32event.CreateEvent(None, True, True, None)
self.assertSignaled(event)
res = win32event.ResetEvent(event)
self.assertEquals(res, None)
self.assertNotSignaled(event)
event.close()
self.assertRaises(pywintypes.error, win32event.ResetEvent, event)
class TestMutex(unittest.TestCase):
def testReleaseMutex(self):
mutex = win32event.CreateMutex(None, True, None)
res = win32event.ReleaseMutex(mutex)
self.assertEqual(res, None)
res = win32event.WaitForSingleObject(mutex, 0)
self.assertEqual(res, win32event.WAIT_OBJECT_0)
mutex.close()
self.assertRaises(pywintypes.error, win32event.ReleaseMutex, mutex)
if __name__=='__main__':
unittest.main()

View file

@ -0,0 +1,960 @@
from __future__ import print_function
import unittest
from pywin32_testutil import str2bytes, TestSkipped, testmain
import win32api, win32file, win32pipe, pywintypes, winerror, win32event
import win32con, ntsecuritycon
import sys
import os
import tempfile
import threading
import time
import shutil
import socket
import datetime
import random
import win32timezone
try:
set
except NameError:
from sets import Set as set
class TestReadBuffer(unittest.TestCase):
def testLen(self):
buffer = win32file.AllocateReadBuffer(1)
self.failUnlessEqual(len(buffer), 1)
def testSimpleIndex(self):
val = str2bytes('\xFF')
buffer = win32file.AllocateReadBuffer(1)
buffer[0] = val
self.failUnlessEqual(buffer[0], val)
def testSimpleSlice(self):
buffer = win32file.AllocateReadBuffer(2)
val = str2bytes('\0\0')
buffer[:2] = val
self.failUnlessEqual(buffer[0:2], val)
class TestSimpleOps(unittest.TestCase):
def testSimpleFiles(self):
fd, filename = tempfile.mkstemp()
os.close(fd)
os.unlink(filename)
handle = win32file.CreateFile(filename, win32file.GENERIC_WRITE, 0, None, win32con.CREATE_NEW, 0, None)
test_data = str2bytes("Hello\0there")
try:
win32file.WriteFile(handle, test_data)
handle.Close()
# Try and open for read
handle = win32file.CreateFile(filename, win32file.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
rc, data = win32file.ReadFile(handle, 1024)
self.assertEquals(data, test_data)
finally:
handle.Close()
try:
os.unlink(filename)
except os.error:
pass
# A simple test using normal read/write operations.
def testMoreFiles(self):
# Create a file in the %TEMP% directory.
testName = os.path.join( win32api.GetTempPath(), "win32filetest.dat" )
desiredAccess = win32file.GENERIC_READ | win32file.GENERIC_WRITE
# Set a flag to delete the file automatically when it is closed.
fileFlags = win32file.FILE_FLAG_DELETE_ON_CLOSE
h = win32file.CreateFile( testName, desiredAccess, win32file.FILE_SHARE_READ, None, win32file.CREATE_ALWAYS, fileFlags, 0)
# Write a known number of bytes to the file.
data = str2bytes("z") * 1025
win32file.WriteFile(h, data)
self.failUnless(win32file.GetFileSize(h) == len(data), "WARNING: Written file does not have the same size as the length of the data in it!")
# Ensure we can read the data back.
win32file.SetFilePointer(h, 0, win32file.FILE_BEGIN)
hr, read_data = win32file.ReadFile(h, len(data)+10) # + 10 to get anything extra
self.failUnless(hr==0, "Readfile returned %d" % hr)
self.failUnless(read_data == data, "Read data is not what we wrote!")
# Now truncate the file at 1/2 its existing size.
newSize = len(data)//2
win32file.SetFilePointer(h, newSize, win32file.FILE_BEGIN)
win32file.SetEndOfFile(h)
self.failUnlessEqual(win32file.GetFileSize(h), newSize)
# GetFileAttributesEx/GetFileAttributesExW tests.
self.failUnlessEqual(win32file.GetFileAttributesEx(testName), win32file.GetFileAttributesExW(testName))
attr, ct, at, wt, size = win32file.GetFileAttributesEx(testName)
self.failUnless(size==newSize,
"Expected GetFileAttributesEx to return the same size as GetFileSize()")
self.failUnless(attr==win32file.GetFileAttributes(testName),
"Expected GetFileAttributesEx to return the same attributes as GetFileAttributes")
h = None # Close the file by removing the last reference to the handle!
self.failUnless(not os.path.isfile(testName), "After closing the file, it still exists!")
def testFilePointer(self):
# via [ 979270 ] SetFilePointer fails with negative offset
# Create a file in the %TEMP% directory.
filename = os.path.join( win32api.GetTempPath(), "win32filetest.dat" )
f = win32file.CreateFile(filename,
win32file.GENERIC_READ|win32file.GENERIC_WRITE,
0,
None,
win32file.CREATE_ALWAYS,
win32file.FILE_ATTRIBUTE_NORMAL,
0)
try:
#Write some data
data = str2bytes('Some data')
(res, written) = win32file.WriteFile(f, data)
self.failIf(res)
self.assertEqual(written, len(data))
#Move at the beginning and read the data
win32file.SetFilePointer(f, 0, win32file.FILE_BEGIN)
(res, s) = win32file.ReadFile(f, len(data))
self.failIf(res)
self.assertEqual(s, data)
#Move at the end and read the data
win32file.SetFilePointer(f, -len(data), win32file.FILE_END)
(res, s) = win32file.ReadFile(f, len(data))
self.failIf(res)
self.failUnlessEqual(s, data)
finally:
f.Close()
os.unlink(filename)
def testFileTimesTimezones(self):
if not issubclass(pywintypes.TimeType, datetime.datetime):
# maybe should report 'skipped', but that's not quite right as
# there is nothing you can do to avoid it being skipped!
return
filename = tempfile.mktemp("-testFileTimes")
# now() is always returning a timestamp with microseconds but the
# file APIs all have zero microseconds, so some comparisons fail.
now_utc = win32timezone.utcnow().replace(microsecond=0)
now_local = now_utc.astimezone(win32timezone.TimeZoneInfo.local())
h = win32file.CreateFile(filename,
win32file.GENERIC_READ|win32file.GENERIC_WRITE,
0, None, win32file.CREATE_ALWAYS, 0, 0)
try:
win32file.SetFileTime(h, now_utc, now_utc, now_utc)
ct, at, wt = win32file.GetFileTime(h)
self.failUnlessEqual(now_local, ct)
self.failUnlessEqual(now_local, at)
self.failUnlessEqual(now_local, wt)
# and the reverse - set local, check against utc
win32file.SetFileTime(h, now_local, now_local, now_local)
ct, at, wt = win32file.GetFileTime(h)
self.failUnlessEqual(now_utc, ct)
self.failUnlessEqual(now_utc, at)
self.failUnlessEqual(now_utc, wt)
finally:
h.close()
os.unlink(filename)
def testFileTimes(self):
if issubclass(pywintypes.TimeType, datetime.datetime):
from win32timezone import TimeZoneInfo
# now() is always returning a timestamp with microseconds but the
# file APIs all have zero microseconds, so some comparisons fail.
now = datetime.datetime.now(tz=TimeZoneInfo.utc()).replace(microsecond=0)
nowish = now + datetime.timedelta(seconds=1)
later = now + datetime.timedelta(seconds=120)
else:
rc, tzi = win32api.GetTimeZoneInformation()
bias = tzi[0]
if rc==2: # daylight-savings is in effect.
bias += tzi[-1]
bias *= 60 # minutes to seconds...
tick = int(time.time())
now = pywintypes.Time(tick+bias)
nowish = pywintypes.Time(tick+bias+1)
later = pywintypes.Time(tick+bias+120)
filename = tempfile.mktemp("-testFileTimes")
# Windows docs the 'last time' isn't valid until the last write
# handle is closed - so create the file, then re-open it to check.
open(filename,"w").close()
f = win32file.CreateFile(filename, win32file.GENERIC_READ|win32file.GENERIC_WRITE,
0, None,
win32con.OPEN_EXISTING, 0, None)
try:
ct, at, wt = win32file.GetFileTime(f)
self.failUnless(ct >= now, "File was created in the past - now=%s, created=%s" % (now, ct))
self.failUnless( now <= ct <= nowish, (now, ct))
self.failUnless(wt >= now, "File was written-to in the past now=%s, written=%s" % (now,wt))
self.failUnless( now <= wt <= nowish, (now, wt))
# Now set the times.
win32file.SetFileTime(f, later, later, later, UTCTimes=True)
# Get them back.
ct, at, wt = win32file.GetFileTime(f)
# XXX - the builtin PyTime type appears to be out by a dst offset.
# just ignore that type here...
self.failUnlessEqual(ct, later)
self.failUnlessEqual(at, later)
self.failUnlessEqual(wt, later)
finally:
f.Close()
os.unlink(filename)
class TestGetFileInfoByHandleEx(unittest.TestCase):
__handle = __filename = None
def setUp(self):
fd, self.__filename = tempfile.mkstemp()
os.close(fd)
def tearDown(self):
if self.__handle is not None:
self.__handle.Close()
if self.__filename is not None:
try:
os.unlink(self.__filename)
except OSError:
pass
self.__handle = self.__filename = None
def testFileBasicInfo(self):
attr = win32file.GetFileAttributes(self.__filename)
f = win32file.CreateFile(self.__filename, win32file.GENERIC_READ, 0, None,
win32con.OPEN_EXISTING, 0, None)
self.__handle = f
ct, at, wt = win32file.GetFileTime(f)
# bug #752: this throws ERROR_BAD_LENGTH (24) in x86 binaries of build 221
basic_info = win32file.GetFileInformationByHandleEx(f, win32file.FileBasicInfo)
self.assertEqual(ct, basic_info['CreationTime'])
self.assertEqual(at, basic_info['LastAccessTime'])
self.assertEqual(wt, basic_info['LastWriteTime'])
self.assertEqual(attr, basic_info['FileAttributes'])
class TestOverlapped(unittest.TestCase):
def testSimpleOverlapped(self):
# Create a file in the %TEMP% directory.
import win32event
testName = os.path.join( win32api.GetTempPath(), "win32filetest.dat" )
desiredAccess = win32file.GENERIC_WRITE
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
# Create the file and write shit-loads of data to it.
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.CREATE_ALWAYS, 0, 0)
chunk_data = str2bytes("z") * 0x8000
num_loops = 512
expected_size = num_loops * len(chunk_data)
for i in range(num_loops):
win32file.WriteFile(h, chunk_data, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(chunk_data)
h.Close()
# Now read the data back overlapped
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
desiredAccess = win32file.GENERIC_READ
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.OPEN_EXISTING, 0, 0)
buffer = win32file.AllocateReadBuffer(0xFFFF)
while 1:
try:
hr, data = win32file.ReadFile(h, buffer, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(data)
if not data is buffer:
self.fail("Unexpected result from ReadFile - should be the same buffer we passed it")
except win32api.error:
break
h.Close()
def testCompletionPortsMultiple(self):
# Mainly checking that we can "associate" an existing handle. This
# failed in build 203.
ioport = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE,
0, 0, 0)
socks = []
for PORT in range(9123, 9125):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('', PORT))
sock.listen(1)
socks.append(sock)
new = win32file.CreateIoCompletionPort(sock.fileno(), ioport, PORT, 0)
assert new is ioport
for s in socks:
s.close()
hv = int(ioport)
ioport = new = None
# The handle itself should be closed now (unless we leak references!)
# Check that.
try:
win32file.CloseHandle(hv)
raise RuntimeError("Expected close to fail!")
except win32file.error as details:
self.failUnlessEqual(details.winerror, winerror.ERROR_INVALID_HANDLE)
def testCompletionPortsQueued(self):
class Foo: pass
io_req_port = win32file.CreateIoCompletionPort(-1, None, 0, 0)
overlapped = pywintypes.OVERLAPPED()
overlapped.object = Foo()
win32file.PostQueuedCompletionStatus(io_req_port, 0, 99, overlapped)
errCode, bytes, key, overlapped = \
win32file.GetQueuedCompletionStatus(io_req_port, win32event.INFINITE)
self.failUnlessEqual(errCode, 0)
self.failUnless(isinstance(overlapped.object, Foo))
def _IOCPServerThread(self, handle, port, drop_overlapped_reference):
overlapped = pywintypes.OVERLAPPED()
win32pipe.ConnectNamedPipe(handle, overlapped)
if drop_overlapped_reference:
# Be naughty - the overlapped object is now dead, but
# GetQueuedCompletionStatus will still find it. Our check of
# reference counting should catch that error.
overlapped = None
# even if we fail, be sure to close the handle; prevents hangs
# on Vista 64...
try:
self.failUnlessRaises(RuntimeError,
win32file.GetQueuedCompletionStatus, port, -1)
finally:
handle.Close()
return
result = win32file.GetQueuedCompletionStatus(port, -1)
ol2 = result[-1]
self.failUnless(ol2 is overlapped)
data = win32file.ReadFile(handle, 512)[1]
win32file.WriteFile(handle, data)
def testCompletionPortsNonQueued(self, test_overlapped_death = 0):
# In 204 we had a reference count bug when OVERLAPPED objects were
# associated with a completion port other than via
# PostQueuedCompletionStatus. This test is based on the reproduction
# reported with that bug.
# Create the pipe.
BUFSIZE = 512
pipe_name = r"\\.\pipe\pywin32_test_pipe"
handle = win32pipe.CreateNamedPipe(pipe_name,
win32pipe.PIPE_ACCESS_DUPLEX|
win32file.FILE_FLAG_OVERLAPPED,
win32pipe.PIPE_TYPE_MESSAGE|
win32pipe.PIPE_READMODE_MESSAGE|
win32pipe.PIPE_WAIT,
1, BUFSIZE, BUFSIZE,
win32pipe.NMPWAIT_WAIT_FOREVER,
None)
# Create an IOCP and associate it with the handle.
port = win32file.CreateIoCompletionPort(-1, 0, 0, 0)
win32file.CreateIoCompletionPort(handle, port, 1, 0)
t = threading.Thread(target=self._IOCPServerThread, args=(handle,port, test_overlapped_death))
t.setDaemon(True) # avoid hanging entire test suite on failure.
t.start()
try:
time.sleep(0.1) # let thread do its thing.
try:
win32pipe.CallNamedPipe(r"\\.\pipe\pywin32_test_pipe", str2bytes("Hello there"), BUFSIZE, 0)
except win32pipe.error:
# Testing for overlapped death causes this
if not test_overlapped_death:
raise
finally:
if not test_overlapped_death:
handle.Close()
t.join(3)
self.failIf(t.isAlive(), "thread didn't finish")
def testCompletionPortsNonQueuedBadReference(self):
self.testCompletionPortsNonQueued(True)
def testHashable(self):
overlapped = pywintypes.OVERLAPPED()
d = {}
d[overlapped] = "hello"
self.failUnlessEqual(d[overlapped], "hello")
def testComparable(self):
overlapped = pywintypes.OVERLAPPED()
self.failUnlessEqual(overlapped, overlapped)
# ensure we explicitly test the operators.
self.failUnless(overlapped == overlapped)
self.failIf(overlapped != overlapped)
def testComparable2(self):
# 2 overlapped objects compare equal if their contents are the same.
overlapped1 = pywintypes.OVERLAPPED()
overlapped2 = pywintypes.OVERLAPPED()
self.failUnlessEqual(overlapped1, overlapped2)
# ensure we explicitly test the operators.
self.failUnless(overlapped1 == overlapped2)
self.failIf(overlapped1 != overlapped2)
# now change something in one of them - should no longer be equal.
overlapped1.hEvent = 1
self.failIfEqual(overlapped1, overlapped2)
# ensure we explicitly test the operators.
self.failIf(overlapped1 == overlapped2)
self.failUnless(overlapped1 != overlapped2)
class TestSocketExtensions(unittest.TestCase):
def acceptWorker(self, port, running_event, stopped_event):
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.bind(('', port))
listener.listen(200)
# create accept socket
accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# An overlapped
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
# accept the connection.
# We used to allow strings etc to be passed here, and they would be
# modified! Obviously this is evil :)
buffer = " " * 1024 # EVIL - SHOULD NOT BE ALLOWED.
self.assertRaises(TypeError, win32file.AcceptEx, listener, accepter, buffer, overlapped)
# This is the correct way to allocate the buffer...
buffer = win32file.AllocateReadBuffer(1024)
rc = win32file.AcceptEx(listener, accepter, buffer, overlapped)
self.failUnlessEqual(rc, winerror.ERROR_IO_PENDING)
# Set the event to say we are all ready
running_event.set()
# and wait for the connection.
rc = win32event.WaitForSingleObject(overlapped.hEvent, 2000)
if rc == win32event.WAIT_TIMEOUT:
self.fail("timed out waiting for a connection")
nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
#fam, loc, rem = win32file.GetAcceptExSockaddrs(accepter, buffer)
accepter.send(buffer[:nbytes])
# NOT set in a finally - this means *successfully* stopped!
stopped_event.set()
def testAcceptEx(self):
port = 4680
running = threading.Event()
stopped = threading.Event()
t = threading.Thread(target=self.acceptWorker, args=(port, running,stopped))
t.start()
running.wait(2)
if not running.isSet():
self.fail("AcceptEx Worker thread failed to start")
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('127.0.0.1', port))
win32file.WSASend(s, str2bytes("hello"), None)
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
# Like above - WSARecv used to allow strings as the receive buffer!!
buffer = " " * 10
self.assertRaises(TypeError, win32file.WSARecv, s, buffer, overlapped)
# This one should work :)
buffer = win32file.AllocateReadBuffer(10)
win32file.WSARecv(s, buffer, overlapped)
nbytes = win32file.GetOverlappedResult(s.fileno(), overlapped, True)
got = buffer[:nbytes]
self.failUnlessEqual(got, str2bytes("hello"))
# thread should have stopped
stopped.wait(2)
if not stopped.isSet():
self.fail("AcceptEx Worker thread failed to successfully stop")
class TestFindFiles(unittest.TestCase):
def testIter(self):
dir = os.path.join(os.getcwd(), "*")
files = win32file.FindFilesW(dir)
set1 = set()
set1.update(files)
set2 = set()
for file in win32file.FindFilesIterator(dir):
set2.add(file)
assert len(set2) > 5, "This directory has less than 5 files!?"
self.failUnlessEqual(set1, set2)
def testBadDir(self):
dir = os.path.join(os.getcwd(), "a dir that doesnt exist", "*")
self.assertRaises(win32file.error, win32file.FindFilesIterator, dir)
def testEmptySpec(self):
spec = os.path.join(os.getcwd(), "*.foo_bar")
num = 0
for i in win32file.FindFilesIterator(spec):
num += 1
self.failUnlessEqual(0, num)
def testEmptyDir(self):
test_path = os.path.join(win32api.GetTempPath(), "win32file_test_directory")
try:
# Note: previously used shutil.rmtree, but when looking for
# reference count leaks, that function showed leaks! os.rmdir
# doesn't have that problem.
os.rmdir(test_path)
except os.error:
pass
os.mkdir(test_path)
try:
num = 0
for i in win32file.FindFilesIterator(os.path.join(test_path, "*")):
num += 1
# Expecting "." and ".." only
self.failUnlessEqual(2, num)
finally:
os.rmdir(test_path)
class TestDirectoryChanges(unittest.TestCase):
num_test_dirs = 1
def setUp(self):
self.watcher_threads = []
self.watcher_thread_changes = []
self.dir_names = []
self.dir_handles = []
for i in range(self.num_test_dirs):
td = tempfile.mktemp("-test-directory-changes-%d" % i)
os.mkdir(td)
self.dir_names.append(td)
hdir = win32file.CreateFile(td,
ntsecuritycon.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ,
None, # security desc
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS |
win32con.FILE_FLAG_OVERLAPPED,
None)
self.dir_handles.append(hdir)
changes = []
t = threading.Thread(target=self._watcherThreadOverlapped,
args=(td, hdir, changes))
t.start()
self.watcher_threads.append(t)
self.watcher_thread_changes.append(changes)
def _watcherThread(self, dn, dh, changes):
# A synchronous version:
# XXX - not used - I was having a whole lot of problems trying to
# get this to work. Specifically:
# * ReadDirectoryChangesW without an OVERLAPPED blocks infinitely.
# * If another thread attempts to close the handle while
# ReadDirectoryChangesW is waiting on it, the ::CloseHandle() method
# blocks (which has nothing to do with the GIL - it is correctly
# managed)
# Which ends up with no way to kill the thread!
flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME
while 1:
try:
print("waiting", dh)
changes = win32file.ReadDirectoryChangesW(dh,
8192,
False, #sub-tree
flags)
print("got", changes)
except:
raise
changes.extend(changes)
def _watcherThreadOverlapped(self, dn, dh, changes):
flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME
buf = win32file.AllocateReadBuffer(8192)
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
while 1:
win32file.ReadDirectoryChangesW(dh,
buf,
False, #sub-tree
flags,
overlapped)
# Wait for our event, or for 5 seconds.
rc = win32event.WaitForSingleObject(overlapped.hEvent, 5000)
if rc == win32event.WAIT_OBJECT_0:
# got some data! Must use GetOverlappedResult to find out
# how much is valid! 0 generally means the handle has
# been closed. Blocking is OK here, as the event has
# already been set.
nbytes = win32file.GetOverlappedResult(dh, overlapped, True)
if nbytes:
bits = win32file.FILE_NOTIFY_INFORMATION(buf, nbytes)
changes.extend(bits)
else:
# This is "normal" exit - our 'tearDown' closes the
# handle.
# print "looks like dir handle was closed!"
return
else:
print("ERROR: Watcher thread timed-out!")
return # kill the thread!
def tearDown(self):
# be careful about raising errors at teardown!
for h in self.dir_handles:
# See comments in _watcherThread above - this appears to
# deadlock if a synchronous ReadDirectoryChangesW is waiting...
# (No such problems with an asynch ReadDirectoryChangesW)
h.Close()
for dn in self.dir_names:
try:
shutil.rmtree(dn)
except OSError:
print("FAILED to remove directory", dn)
for t in self.watcher_threads:
# closing dir handle should have killed threads!
t.join(5)
if t.isAlive():
print("FAILED to wait for thread termination")
def stablize(self):
time.sleep(0.5)
def testSimple(self):
self.stablize()
for dn in self.dir_names:
fn = os.path.join(dn, "test_file")
open(fn, "w").close()
self.stablize()
changes = self.watcher_thread_changes[0]
self.failUnlessEqual(changes, [(1, "test_file")])
def testSmall(self):
self.stablize()
for dn in self.dir_names:
fn = os.path.join(dn, "x")
open(fn, "w").close()
self.stablize()
changes = self.watcher_thread_changes[0]
self.failUnlessEqual(changes, [(1, "x")])
class TestEncrypt(unittest.TestCase):
def testEncrypt(self):
fname = tempfile.mktemp("win32file_test")
f = open(fname, "wb")
f.write(str2bytes("hello"))
f.close()
f = None
try:
try:
win32file.EncryptFile(fname)
except win32file.error as details:
if details.winerror != winerror.ERROR_ACCESS_DENIED:
raise
print("It appears this is not NTFS - cant encrypt/decrypt")
win32file.DecryptFile(fname)
finally:
if f is not None:
f.close()
os.unlink(fname)
class TestConnect(unittest.TestCase):
def connect_thread_runner(self, expect_payload, giveup_event):
# As Windows 2000 doesn't do ConnectEx, we need to use a non-blocking
# accept, as our test connection may never come. May as well use
# AcceptEx for this...
listener = socket.socket()
self.addr = ('localhost', random.randint(10000,64000))
listener.bind(self.addr)
listener.listen(1)
# create accept socket
accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# An overlapped
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
# accept the connection.
if expect_payload:
buf_size = 1024
else:
# when we don't expect data we must be careful to only pass the
# exact number of bytes for the endpoint data...
buf_size = win32file.CalculateSocketEndPointSize(listener)
buffer = win32file.AllocateReadBuffer(buf_size)
win32file.AcceptEx(listener, accepter, buffer, overlapped)
# wait for the connection or our test to fail.
events = giveup_event, overlapped.hEvent
rc = win32event.WaitForMultipleObjects(events, False, 2000)
if rc == win32event.WAIT_TIMEOUT:
self.fail("timed out waiting for a connection")
if rc == win32event.WAIT_OBJECT_0:
# Our main thread running the test failed and will never connect.
return
# must be a connection.
nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
if expect_payload:
self.request = buffer[:nbytes]
accepter.send(str2bytes('some expected response'))
def test_connect_with_payload(self):
giveup_event = win32event.CreateEvent(None, 0, 0, None)
t = threading.Thread(target=self.connect_thread_runner,
args=(True, giveup_event))
t.start()
time.sleep(0.1)
s2 = socket.socket()
ol = pywintypes.OVERLAPPED()
s2.bind(('0.0.0.0', 0)) # connectex requires the socket be bound beforehand
try:
win32file.ConnectEx(s2, self.addr, ol, str2bytes("some expected request"))
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == 10022: # WSAEINVAL
raise TestSkipped("ConnectEx is not available on this platform")
raise # some error error we don't expect.
win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
buff = win32file.AllocateReadBuffer(1024)
win32file.WSARecv(s2, buff, ol, 0)
length = win32file.GetOverlappedResult(s2.fileno(), ol, 1)
self.response = buff[:length]
self.assertEqual(self.response, str2bytes('some expected response'))
self.assertEqual(self.request, str2bytes('some expected request'))
t.join(5)
self.failIf(t.isAlive(), "worker thread didn't terminate")
def test_connect_without_payload(self):
giveup_event = win32event.CreateEvent(None, 0, 0, None)
t = threading.Thread(target=self.connect_thread_runner,
args=(False, giveup_event))
t.start()
time.sleep(0.1)
s2 = socket.socket()
ol = pywintypes.OVERLAPPED()
s2.bind(('0.0.0.0', 0)) # connectex requires the socket be bound beforehand
try:
win32file.ConnectEx(s2, self.addr, ol)
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == 10022: # WSAEINVAL
raise TestSkipped("ConnectEx is not available on this platform")
raise # some error error we don't expect.
win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
buff = win32file.AllocateReadBuffer(1024)
win32file.WSARecv(s2, buff, ol, 0)
length = win32file.GetOverlappedResult(s2.fileno(), ol, 1)
self.response = buff[:length]
self.assertEqual(self.response, str2bytes('some expected response'))
t.join(5)
self.failIf(t.isAlive(), "worker thread didn't terminate")
class TestTransmit(unittest.TestCase):
def test_transmit(self):
import binascii
bytes = os.urandom(1024*1024)
val = binascii.hexlify(bytes)
val_length = len(val)
f = tempfile.TemporaryFile()
f.write(val)
def runner():
s1 = socket.socket()
self.addr = ('localhost', random.randint(10000,64000))
s1.bind(self.addr)
s1.listen(1)
cli, addr = s1.accept()
buf = 1
self.request = []
while buf:
buf = cli.recv(1024*100)
self.request.append(buf)
th = threading.Thread(target=runner)
th.start()
time.sleep(0.5)
s2 = socket.socket()
s2.connect(self.addr)
length = 0
aaa = str2bytes("[AAA]")
bbb = str2bytes("[BBB]")
ccc = str2bytes("[CCC]")
ddd = str2bytes("[DDD]")
empty = str2bytes("")
ol = pywintypes.OVERLAPPED()
f.seek(0)
win32file.TransmitFile(s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0)
length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
f.seek(0)
win32file.TransmitFile(s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, aaa, bbb)
length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
f.seek(0)
win32file.TransmitFile(s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, empty, empty)
length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
f.seek(0)
win32file.TransmitFile(s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, None, ccc)
length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
ol = pywintypes.OVERLAPPED()
f.seek(0)
win32file.TransmitFile(s2, win32file._get_osfhandle(f.fileno()), val_length, 0, ol, 0, ddd)
length += win32file.GetOverlappedResult(s2.fileno(), ol, 1)
s2.close()
th.join()
buf = str2bytes('').join(self.request)
self.assertEqual(length, len(buf))
expected = val + aaa + val + bbb + val + val + ccc + ddd + val
self.assertEqual(type(expected), type(buf))
self.assert_(expected == buf)
class TestWSAEnumNetworkEvents(unittest.TestCase):
def test_basics(self):
s = socket.socket()
e = win32event.CreateEvent(None, 1, 0, None)
win32file.WSAEventSelect(s, e, 0)
self.assertEquals(win32file.WSAEnumNetworkEvents(s), {})
self.assertEquals(win32file.WSAEnumNetworkEvents(s, e), {})
self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, e, 3)
self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, "spam")
self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam", e)
self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam")
f = open("NUL")
h = win32file._get_osfhandle(f.fileno())
self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, h)
self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, s, h)
try:
win32file.WSAEnumNetworkEvents(h)
except win32file.error as e:
self.assertEquals(e.winerror, win32file.WSAENOTSOCK)
try:
win32file.WSAEnumNetworkEvents(s, h)
except win32file.error as e:
# According to the docs it would seem reasonable that
# this would fail with WSAEINVAL, but it doesn't.
self.assertEquals(e.winerror, win32file.WSAENOTSOCK)
def test_functional(self):
# This is not really a unit test, but it does exercise the code
# quite well and can serve as an example of WSAEventSelect and
# WSAEnumNetworkEvents usage.
port = socket.socket()
port.setblocking(0)
port_event = win32event.CreateEvent(None, 0, 0, None)
win32file.WSAEventSelect(port, port_event,
win32file.FD_ACCEPT |
win32file.FD_CLOSE)
port.bind(("127.0.0.1", 0))
port.listen(10)
client = socket.socket()
client.setblocking(0)
client_event = win32event.CreateEvent(None, 0, 0, None)
win32file.WSAEventSelect(client, client_event,
win32file.FD_CONNECT |
win32file.FD_READ |
win32file.FD_WRITE |
win32file.FD_CLOSE)
err = client.connect_ex(port.getsockname())
self.assertEquals(err, win32file.WSAEWOULDBLOCK)
res = win32event.WaitForSingleObject(port_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(port, port_event)
self.assertEquals(events, {win32file.FD_ACCEPT: 0})
server, addr = port.accept()
server.setblocking(0)
server_event = win32event.CreateEvent(None, 1, 0, None)
win32file.WSAEventSelect(server, server_event,
win32file.FD_READ |
win32file.FD_WRITE |
win32file.FD_CLOSE)
res = win32event.WaitForSingleObject(server_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(server, server_event)
self.assertEquals(events, {win32file.FD_WRITE: 0})
res = win32event.WaitForSingleObject(client_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(client, client_event)
self.assertEquals(events, {win32file.FD_CONNECT: 0,
win32file.FD_WRITE: 0})
sent = 0
data = str2bytes("x") * 16 * 1024
while sent < 16 * 1024 * 1024:
try:
sent += client.send(data)
except socket.error as e:
if e.args[0] == win32file.WSAEINTR:
continue
elif e.args[0] in (win32file.WSAEWOULDBLOCK, win32file.WSAENOBUFS):
break
else:
raise
else:
self.fail("could not find socket buffer limit")
events = win32file.WSAEnumNetworkEvents(client)
self.assertEquals(events, {})
res = win32event.WaitForSingleObject(server_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(server, server_event)
self.assertEquals(events, {win32file.FD_READ: 0})
received = 0
while received < sent:
try:
received += len(server.recv(16 * 1024))
except socket.error as e:
if e.args[0] in [win32file.WSAEINTR, win32file.WSAEWOULDBLOCK]:
continue
else:
raise
self.assertEquals(received, sent)
events = win32file.WSAEnumNetworkEvents(server)
self.assertEquals(events, {})
res = win32event.WaitForSingleObject(client_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(client, client_event)
self.assertEquals(events, {win32file.FD_WRITE: 0})
client.shutdown(socket.SHUT_WR)
res = win32event.WaitForSingleObject(server_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
# strange timing issues...
for i in range(5):
events = win32file.WSAEnumNetworkEvents(server, server_event)
if events: break
win32api.Sleep(100)
else:
raise AssertionError("failed to get events")
self.assertEquals(events, {win32file.FD_CLOSE: 0})
events = win32file.WSAEnumNetworkEvents(client)
self.assertEquals(events, {})
server.close()
res = win32event.WaitForSingleObject(client_event, 1000)
self.assertEquals(res, win32event.WAIT_OBJECT_0)
events = win32file.WSAEnumNetworkEvents(client, client_event)
self.assertEquals(events, {win32file.FD_CLOSE: 0})
client.close()
events = win32file.WSAEnumNetworkEvents(port)
self.assertEquals(events, {})
if __name__ == '__main__':
testmain()

View file

@ -0,0 +1,74 @@
# tests for win32gui
import unittest
import win32gui
import pywin32_testutil
import operator
import array
import sys
# theoretically should be in pywin32_testutil, but this is the only place
# that currently needs such a function...
def ob2bytes(ob):
if sys.version_info < (3,0):
return str(buffer(ob))
# py3k.
return bytes(ob)
class TestPyGetString(unittest.TestCase):
def test_get_string(self):
# test invalid addresses cause a ValueError rather than crash!
self.assertRaises(ValueError, win32gui.PyGetString, 0)
self.assertRaises(ValueError, win32gui.PyGetString, 1)
self.assertRaises(ValueError, win32gui.PyGetString, 1,1)
class TestPyGetMemory(unittest.TestCase):
def test_ob(self):
# Check the PyGetMemory result and a bytes string can be compared
test_data = pywin32_testutil.str2bytes("\0\1\2\3\4\5\6")
c = array.array("b", test_data)
addr, buflen = c.buffer_info()
got = win32gui.PyGetMemory(addr, buflen)
self.failUnlessEqual(len(got), len(test_data))
self.failUnlessEqual(ob2bytes(got), test_data)
def test_memory_index(self):
# Check we can index into the buffer object returned by PyGetMemory
test_data = pywin32_testutil.str2bytes("\0\1\2\3\4\5\6")
c = array.array("b", test_data)
addr, buflen = c.buffer_info()
got = win32gui.PyGetMemory(addr, buflen)
self.failUnlessEqual(got[0], pywin32_testutil.str2bytes('\0'))
def test_memory_slice(self):
# Check we can slice the buffer object returned by PyGetMemory
test_data = pywin32_testutil.str2bytes("\0\1\2\3\4\5\6")
c = array.array("b", test_data)
addr, buflen = c.buffer_info()
got = win32gui.PyGetMemory(addr, buflen)
self.failUnlessEqual(got[0:3], pywin32_testutil.str2bytes('\0\1\2'))
def test_real_view(self):
# Do the PyGetMemory, then change the original memory, then ensure
# the initial object we fetched sees the new value.
test_data = pywin32_testutil.str2bytes("\0\1\2\3\4\5\6")
c = array.array("b", test_data)
addr, buflen = c.buffer_info()
got = win32gui.PyGetMemory(addr, buflen)
self.failUnlessEqual(got[0], pywin32_testutil.str2bytes('\0'))
new = pywin32_testutil.str2bytes('\1')
c[0] = 1
self.failUnlessEqual(got[0], new)
def test_memory_not_writable(self):
# Check the buffer object fetched by PyGetMemory isn't writable.
test_data = pywin32_testutil.str2bytes("\0\1\2\3\4\5\6")
c = array.array("b", test_data)
addr, buflen = c.buffer_info()
got = win32gui.PyGetMemory(addr, buflen)
new = pywin32_testutil.str2bytes('\1')
self.failUnlessRaises(TypeError, operator.setitem, got, 0, new)
if __name__=='__main__':
unittest.main()

View file

@ -0,0 +1,223 @@
import unittest
import win32gui
import win32gui_struct
import win32con
import array
import pythoncom
class TestBase(unittest.TestCase):
def assertDictEquals(self, d, **kw):
checked = dict()
for n, v in kw.items():
self.failUnlessEqual(v, d[n],
"'%s' doesn't match: %r != %r" % (n, v, d[n]))
checked[n] = True
checked_keys = list(checked.keys())
passed_keys = list(kw.keys())
checked_keys.sort()
passed_keys.sort()
self.failUnlessEqual(checked_keys, passed_keys)
class TestMenuItemInfo(TestBase):
def _testPackUnpack(self, text):
vals = dict(fType=win32con.MFT_MENUBARBREAK,
fState=win32con.MFS_CHECKED,
wID=123,
hSubMenu=1234,
hbmpChecked=12345,
hbmpUnchecked=123456,
dwItemData=1234567,
text=text,
hbmpItem=321)
mii, extras = win32gui_struct.PackMENUITEMINFO(**vals)
fType, fState, wID, hSubMenu, hbmpChecked, hbmpUnchecked, \
dwItemData, text, hbmpItem = win32gui_struct.UnpackMENUITEMINFO(mii)
self.assertDictEquals(vals, fType=fType, fState=fState, wID=wID,
hSubMenu=hSubMenu, hbmpChecked=hbmpChecked,
hbmpUnchecked=hbmpUnchecked,
dwItemData=dwItemData, text=text,
hbmpItem=hbmpItem)
def testPackUnpack(self):
self._testPackUnpack("Hello")
def testPackUnpackNone(self):
self._testPackUnpack(None)
def testEmptyMenuItemInfo(self):
mii, extra = win32gui_struct.EmptyMENUITEMINFO()
fType, fState, wID, hSubMenu, hbmpChecked, hbmpUnchecked, \
dwItemData, text, hbmpItem = win32gui_struct.UnpackMENUITEMINFO(mii)
self.failUnlessEqual(fType, 0)
self.failUnlessEqual(fState, 0)
self.failUnlessEqual(wID, 0)
self.failUnlessEqual(hSubMenu, 0)
self.failUnlessEqual(hbmpChecked, 0)
self.failUnlessEqual(hbmpUnchecked, 0)
self.failUnlessEqual(dwItemData, 0)
self.failUnlessEqual(hbmpItem, 0)
# it's not clear if UnpackMENUITEMINFO() should ignore cch, instead
# assuming it is a buffer size rather than 'current length' - but it
# never has (and this gives us every \0 in the string), and actually
# helps us test the unicode/str semantics.
self.failUnlessEqual(text, '\0' * len(text))
class TestMenuInfo(TestBase):
def testPackUnpack(self):
vals = dict(dwStyle=1, cyMax=2, hbrBack=3, dwContextHelpID=4,
dwMenuData=5)
mi = win32gui_struct.PackMENUINFO(**vals)
dwStyle, cyMax, hbrBack, dwContextHelpID, dwMenuData = \
win32gui_struct.UnpackMENUINFO(mi)
self.assertDictEquals(vals, dwStyle=dwStyle, cyMax=cyMax,
hbrBack=hbrBack,
dwContextHelpID=dwContextHelpID,
dwMenuData=dwMenuData)
def testEmptyMenuItemInfo(self):
mi = win32gui_struct.EmptyMENUINFO()
dwStyle, cyMax, hbrBack, dwContextHelpID, dwMenuData = \
win32gui_struct.UnpackMENUINFO(mi)
self.failUnlessEqual(dwStyle, 0)
self.failUnlessEqual(cyMax, 0)
self.failUnlessEqual(hbrBack, 0)
self.failUnlessEqual(dwContextHelpID, 0)
self.failUnlessEqual(dwMenuData, 0)
class TestTreeViewItem(TestBase):
def _testPackUnpack(self, text):
vals = dict(hitem=1, state=2, stateMask=3, text=text, image=4,
selimage=5, citems=6, param=7)
ti, extra = win32gui_struct.PackTVITEM(**vals)
hitem, state, stateMask, text, image, selimage, citems, param = \
win32gui_struct.UnpackTVITEM(ti)
self.assertDictEquals(vals, hitem=hitem, state=state,
stateMask=stateMask, text=text, image=image,
selimage=selimage, citems=citems, param=param)
def testPackUnpack(self):
self._testPackUnpack("Hello")
def testPackUnpackNone(self):
self._testPackUnpack(None)
def testEmpty(self):
ti, extras = win32gui_struct.EmptyTVITEM(0)
hitem, state, stateMask, text, image, selimage, citems, param = \
win32gui_struct.UnpackTVITEM(ti)
self.failUnlessEqual(hitem, 0)
self.failUnlessEqual(state, 0)
self.failUnlessEqual(stateMask, 0)
self.failUnlessEqual(text, '')
self.failUnlessEqual(image, 0)
self.failUnlessEqual(selimage, 0)
self.failUnlessEqual(citems, 0)
self.failUnlessEqual(param, 0)
class TestListViewItem(TestBase):
def _testPackUnpack(self, text):
vals = dict(item=None, subItem=None, state=1, stateMask=2,
text=text, image=3, param=4, indent=5)
ti, extra = win32gui_struct.PackLVITEM(**vals)
item, subItem, state, stateMask, text, image, param, indent = \
win32gui_struct.UnpackLVITEM(ti)
# patch expected values.
vals['item'] = 0
vals['subItem'] = 0
self.assertDictEquals(vals, item=item, subItem=subItem, state=state,
stateMask=stateMask, text=text, image=image,
param=param, indent=indent)
def testPackUnpack(self):
self._testPackUnpack("Hello")
def testPackUnpackNone(self):
self._testPackUnpack(None)
def testEmpty(self):
ti, extras = win32gui_struct.EmptyLVITEM(1, 2)
item, subItem, state, stateMask, text, image, param, indent = \
win32gui_struct.UnpackLVITEM(ti)
self.failUnlessEqual(item, 1)
self.failUnlessEqual(subItem, 2)
self.failUnlessEqual(state, 0)
self.failUnlessEqual(stateMask, 0)
self.failUnlessEqual(text, '')
self.failUnlessEqual(image, 0)
self.failUnlessEqual(param, 0)
self.failUnlessEqual(indent, 0)
class TestLVColumn(TestBase):
def _testPackUnpack(self, text):
vals = dict(fmt=1, cx=2, text=text, subItem=3, image=4, order=5)
ti, extra = win32gui_struct.PackLVCOLUMN(**vals)
fmt, cx, text, subItem, image, order = \
win32gui_struct.UnpackLVCOLUMN(ti)
self.assertDictEquals(vals, fmt=fmt, cx=cx, text=text, subItem=subItem,
image=image, order=order)
def testPackUnpack(self):
self._testPackUnpack("Hello")
def testPackUnpackNone(self):
self._testPackUnpack(None)
def testEmpty(self):
ti, extras = win32gui_struct.EmptyLVCOLUMN()
fmt, cx, text, subItem, image, order = \
win32gui_struct.UnpackLVCOLUMN(ti)
self.failUnlessEqual(fmt, 0)
self.failUnlessEqual(cx, 0)
self.failUnlessEqual(text, '')
self.failUnlessEqual(subItem, 0)
self.failUnlessEqual(image, 0)
self.failUnlessEqual(order, 0)
class TestDEV_BROADCAST_HANDLE(TestBase):
def testPackUnpack(self):
s = win32gui_struct.PackDEV_BROADCAST_HANDLE(123)
c = array.array("b", s)
got = win32gui_struct.UnpackDEV_BROADCAST(c.buffer_info()[0])
self.failUnlessEqual(got.handle, 123)
def testGUID(self):
s = win32gui_struct.PackDEV_BROADCAST_HANDLE(123,
guid=pythoncom.IID_IUnknown)
c = array.array("b", s)
got = win32gui_struct.UnpackDEV_BROADCAST(c.buffer_info()[0])
self.failUnlessEqual(got.handle, 123)
self.failUnlessEqual(got.eventguid, pythoncom.IID_IUnknown)
class TestDEV_BROADCAST_DEVICEINTERFACE(TestBase):
def testPackUnpack(self):
s = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE(pythoncom.IID_IUnknown,
"hello")
c = array.array("b", s)
got = win32gui_struct.UnpackDEV_BROADCAST(c.buffer_info()[0])
self.failUnlessEqual(got.classguid, pythoncom.IID_IUnknown)
self.failUnlessEqual(got.name, "hello")
class TestDEV_BROADCAST_VOLUME(TestBase):
def testPackUnpack(self):
s = win32gui_struct.PackDEV_BROADCAST_VOLUME(123, 456)
c = array.array("b", s)
got = win32gui_struct.UnpackDEV_BROADCAST(c.buffer_info()[0])
self.failUnlessEqual(got.unitmask, 123)
self.failUnlessEqual(got.flags, 456)
if __name__=='__main__':
unittest.main()

View file

@ -0,0 +1,75 @@
from win32inet import *
from win32inetcon import *
import winerror
from pywin32_testutil import str2bytes # py3k-friendly helper
from pywin32_testutil import TestSkipped
import unittest
class CookieTests(unittest.TestCase):
def testCookies(self):
data = "TestData=Test"
InternetSetCookie("http://www.python.org", None, data)
got = InternetGetCookie("http://www.python.org", None)
self.assertEqual(got, data)
def testCookiesEmpty(self):
try:
InternetGetCookie("http://site-with-no-cookie.python.org", None)
self.fail("expected win32 exception")
except error as exc:
self.failUnlessEqual(exc.winerror, winerror.ERROR_NO_MORE_ITEMS)
class UrlTests(unittest.TestCase):
def testSimpleCanonicalize(self):
ret = InternetCanonicalizeUrl("foo bar")
self.assertEqual(ret, "foo%20bar")
def testLongCanonicalize(self):
# a 4k URL causes the underlying API to request a bigger buffer"
big = "x" * 2048
ret = InternetCanonicalizeUrl(big + " " + big)
self.assertEqual(ret, big + "%20" + big)
class TestNetwork(unittest.TestCase):
def setUp(self):
self.hi = InternetOpen("test", INTERNET_OPEN_TYPE_DIRECT, None, None, 0)
def tearDown(self):
self.hi.Close()
def testPythonDotOrg(self):
hdl = InternetOpenUrl(self.hi, "http://www.python.org", None,
INTERNET_FLAG_EXISTING_CONNECT)
chunks = []
while 1:
chunk = InternetReadFile(hdl, 1024)
if not chunk:
break
chunks.append(chunk)
data = str2bytes('').join(chunks)
assert data.find(str2bytes("Python"))>0, repr(data) # This must appear somewhere on the main page!
def testFtpCommand(self):
# ftp.python.org doesn't exist. ftp.gnu.org is what Python's urllib
# test code uses.
# (As of 2020 it doesn't! Unsurprisingly, it's difficult to find a good
# test server. This test sometimes works, but often doesn't - so handle
# failure here as a "skip")
try:
hcon = InternetConnect(self.hi, "ftp.gnu.org", INTERNET_INVALID_PORT_NUMBER,
None, None, # username/password
INTERNET_SERVICE_FTP, 0, 0)
try:
hftp = FtpCommand(hcon, True, FTP_TRANSFER_TYPE_ASCII, 'NLST', 0)
try:
print("Connected - response info is", InternetGetLastResponseInfo())
got = InternetReadFile(hftp, 2048)
print("Read", len(got), "bytes")
finally:
hftp.Close()
finally:
hcon.Close()
except error as e:
raise TestSkipped(e)
if __name__=='__main__':
unittest.main()

View file

@ -0,0 +1,21 @@
import unittest
import win32net, win32netcon
class TestCase(unittest.TestCase):
def testGroupsGoodResume(self, server=None):
res=0
level=0 #setting it to 1 will provide more detailed info
while True:
(user_list,total,res)=win32net.NetGroupEnum(server,level,res)
for i in user_list:
pass
if not res:
break
def testGroupsBadResume(self, server=None):
res=1 # Can't pass this first time round.
self.assertRaises(win32net.error, win32net.NetGroupEnum, server,0,res)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,132 @@
import unittest
import time
import threading
from pywin32_testutil import str2bytes # py3k-friendly helper
import win32pipe
import win32file
import win32event
import pywintypes
import winerror
import win32con
class PipeTests(unittest.TestCase):
pipename = "\\\\.\\pipe\\python_test_pipe"
def _serverThread(self, pipe_handle, event, wait_time):
# just do one connection and terminate.
hr = win32pipe.ConnectNamedPipe(pipe_handle)
self.failUnless(hr in (0, winerror.ERROR_PIPE_CONNECTED), "Got error code 0x%x" % (hr,))
hr, got = win32file.ReadFile(pipe_handle, 100)
self.failUnlessEqual(got, str2bytes("foo\0bar"))
time.sleep(wait_time)
win32file.WriteFile(pipe_handle, str2bytes("bar\0foo"))
pipe_handle.Close()
event.set()
def startPipeServer(self, event, wait_time = 0):
openMode = win32pipe.PIPE_ACCESS_DUPLEX
pipeMode = win32pipe.PIPE_TYPE_MESSAGE | win32pipe.PIPE_WAIT
sa = pywintypes.SECURITY_ATTRIBUTES()
sa.SetSecurityDescriptorDacl ( 1, None, 0 )
pipe_handle = win32pipe.CreateNamedPipe(self.pipename,
openMode,
pipeMode,
win32pipe.PIPE_UNLIMITED_INSTANCES,
0,
0,
2000,
sa)
threading.Thread(target=self._serverThread, args=(pipe_handle, event, wait_time)).start()
def testCallNamedPipe(self):
event = threading.Event()
self.startPipeServer(event)
got = win32pipe.CallNamedPipe(self.pipename,str2bytes("foo\0bar"), 1024, win32pipe.NMPWAIT_WAIT_FOREVER)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def testTransactNamedPipeBlocking(self):
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), 1024, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def testTransactNamedPipeBlockingBuffer(self):
# Like testTransactNamedPipeBlocking, but a pre-allocated buffer is
# passed (not really that useful, but it exercises the code path)
event = threading.Event()
self.startPipeServer(event)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
0, # win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, None)
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
def testTransactNamedPipeAsync(self):
event = threading.Event()
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
self.startPipeServer(event, 0.5)
open_mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE
hpipe = win32file.CreateFile(self.pipename,
open_mode,
0, # no sharing
None, # default security
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
None)
# set to message mode.
win32pipe.SetNamedPipeHandleState(
hpipe, win32pipe.PIPE_READMODE_MESSAGE, None, None)
buffer = win32file.AllocateReadBuffer(1024)
hr, got = win32pipe.TransactNamedPipe(hpipe, str2bytes("foo\0bar"), buffer, overlapped)
self.failUnlessEqual(hr, winerror.ERROR_IO_PENDING)
nbytes = win32file.GetOverlappedResult(hpipe, overlapped, True)
got = buffer[:nbytes]
self.failUnlessEqual(got, str2bytes("bar\0foo"))
event.wait(5)
self.failUnless(event.isSet(), "Pipe server thread didn't terminate")
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,67 @@
import sys, os
import unittest
import win32rcparser
import win32con
import tempfile
class TestParser(unittest.TestCase):
def setUp(self):
rc_file = os.path.join(os.path.dirname(__file__), "win32rcparser", "test.rc")
self.resources = win32rcparser.Parse(rc_file)
def testStrings(self):
for sid, expected in [
("IDS_TEST_STRING4", "Test 'single quoted' string"),
("IDS_TEST_STRING1", 'Test "quoted" string'),
("IDS_TEST_STRING3", 'String with single " quote'),
("IDS_TEST_STRING2", 'Test string'),
]:
got = self.resources.stringTable[sid].value
self.assertEqual(got, expected)
def testStandardIds(self):
for idc in "IDOK IDCANCEL".split():
correct = getattr(win32con, idc)
self.assertEqual(self.resources.names[correct], idc)
self.assertEqual(self.resources.ids[idc], correct)
def testTabStop(self):
d = self.resources.dialogs["IDD_TEST_DIALOG2"]
tabstop_names = ["IDC_EDIT1", "IDOK"] # should have WS_TABSTOP
tabstop_ids = [self.resources.ids[name] for name in tabstop_names]
notabstop_names = ["IDC_EDIT2"] # should have WS_TABSTOP
notabstop_ids = [self.resources.ids[name] for name in notabstop_names]
num_ok = 0
for cdef in d[1:]: # skip dlgdef
#print cdef
cid = cdef[2]
style = cdef[-2]
styleex = cdef[-1]
if cid in tabstop_ids:
self.failUnlessEqual(style & win32con.WS_TABSTOP, win32con.WS_TABSTOP)
num_ok += 1
elif cid in notabstop_ids:
self.failUnlessEqual(style & win32con.WS_TABSTOP, 0)
num_ok += 1
self.failUnlessEqual(num_ok, len(tabstop_ids) + len(notabstop_ids))
class TestGenerated(TestParser):
def setUp(self):
# don't call base!
rc_file = os.path.join(os.path.dirname(__file__), "win32rcparser", "test.rc")
py_file = tempfile.mktemp('test_win32rcparser.py')
try:
win32rcparser.GenerateFrozenResource(rc_file, py_file)
py_source = open(py_file).read()
finally:
if os.path.isfile(py_file):
os.unlink(py_file)
# poor-man's import :)
globs = {}
exec(py_source, globs, globs)
self.resources = globs["FakeParser"]()
if __name__=='__main__':
unittest.main()

View file

@ -0,0 +1,15 @@
# Test module for win32timezone
import unittest
import win32timezone
import doctest
class Win32TimeZoneTest(unittest.TestCase):
def testWin32TZ(self):
failed, total = doctest.testmod(win32timezone, verbose=False)
self.failIf(failed)
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,344 @@
import unittest
import win32trace
import threading
import time
import os
import sys
if __name__=='__main__':
this_file = sys.argv[0]
else:
this_file = __file__
def CheckNoOtherReaders():
win32trace.write("Hi")
time.sleep(0.05)
if win32trace.read() != "Hi":
# Reset everything so following tests still fail with this error!
win32trace.TermRead()
win32trace.TermWrite()
raise RuntimeError("An existing win32trace reader appears to be " \
"running - please stop this process and try again")
class TestInitOps(unittest.TestCase):
def setUp(self):
# clear old data
win32trace.InitRead()
win32trace.read()
win32trace.TermRead()
def tearDown(self):
try:
win32trace.TermRead()
except win32trace.error:
pass
try:
win32trace.TermWrite()
except win32trace.error:
pass
def testInitTermRead(self):
self.assertRaises(win32trace.error, win32trace.read)
win32trace.InitRead()
result = win32trace.read()
self.assertEquals(result, '')
win32trace.TermRead()
self.assertRaises(win32trace.error, win32trace.read)
win32trace.InitRead()
self.assertRaises(win32trace.error, win32trace.InitRead)
win32trace.InitWrite()
self.assertRaises(win32trace.error, win32trace.InitWrite)
win32trace.TermWrite()
win32trace.TermRead()
def testInitTermWrite(self):
self.assertRaises(win32trace.error, win32trace.write, 'Hei')
win32trace.InitWrite()
win32trace.write('Johan Galtung')
win32trace.TermWrite()
self.assertRaises(win32trace.error, win32trace.write, 'Hei')
def testTermSematics(self):
win32trace.InitWrite()
win32trace.write('Ta da')
# if we both Write and Read are terminated at the same time,
# we lose the data as the win32 object is closed. Note that
# if another writer is running, we do *not* lose the data - so
# test for either the correct data or an empty string
win32trace.TermWrite()
win32trace.InitRead()
self.failUnless(win32trace.read() in ['Ta da', ''])
win32trace.TermRead()
# we keep the data because we init read before terminating write
win32trace.InitWrite()
win32trace.write('Ta da')
win32trace.InitRead()
win32trace.TermWrite()
self.assertEquals('Ta da', win32trace.read())
win32trace.TermRead()
class BasicSetupTearDown(unittest.TestCase):
def setUp(self):
win32trace.InitRead()
# If any other writers are running (even if not actively writing),
# terminating the module will *not* close the handle, meaning old data
# will remain. This can cause other tests to fail.
win32trace.read()
win32trace.InitWrite()
def tearDown(self):
win32trace.TermWrite()
win32trace.TermRead()
class TestModuleOps(BasicSetupTearDown):
def testRoundTrip(self):
win32trace.write('Syver Enstad')
syverEnstad = win32trace.read()
self.assertEquals('Syver Enstad', syverEnstad)
def testRoundTripUnicode(self):
win32trace.write('\xa9opyright Syver Enstad')
syverEnstad = win32trace.read()
# str objects are always returned in py2k (latin-1 encoding was used
# on unicode objects)
self.assertEquals('\xa9opyright Syver Enstad', syverEnstad)
def testBlockingRead(self):
win32trace.write('Syver Enstad')
self.assertEquals('Syver Enstad', win32trace.blockingread())
def testBlockingReadUnicode(self):
win32trace.write('\xa9opyright Syver Enstad')
# str objects are always returned in py2k (latin-1 encoding was used
# on unicode objects)
self.assertEquals('\xa9opyright Syver Enstad', win32trace.blockingread())
def testFlush(self):
win32trace.flush()
class TestTraceObjectOps(BasicSetupTearDown):
def testInit(self):
win32trace.TermRead()
win32trace.TermWrite()
traceObject = win32trace.GetTracer()
self.assertRaises(win32trace.error, traceObject.read)
self.assertRaises(win32trace.error, traceObject.write, '')
win32trace.InitRead()
win32trace.InitWrite()
self.assertEquals('', traceObject.read())
traceObject.write('Syver')
def testFlush(self):
traceObject = win32trace.GetTracer()
traceObject.flush()
def testIsatty(self):
tracer = win32trace.GetTracer()
assert tracer.isatty() == False
def testRoundTrip(self):
traceObject = win32trace.GetTracer()
traceObject.write('Syver Enstad')
self.assertEquals('Syver Enstad', traceObject.read())
class WriterThread(threading.Thread):
def run(self):
self.writeCount = 0
for each in range(self.BucketCount):
win32trace.write(str(each))
self.writeCount = self.BucketCount
def verifyWritten(self):
return self.writeCount == self.BucketCount
class TestMultipleThreadsWriting(unittest.TestCase):
# FullBucket is the thread count
FullBucket = 50
BucketCount = 9 # buckets must be a single digit number (ie. less than 10)
def setUp(self):
WriterThread.BucketCount = self.BucketCount
win32trace.InitRead()
win32trace.read() # clear any old data.
win32trace.InitWrite()
CheckNoOtherReaders()
self.threads = [WriterThread() for each in range(self.FullBucket)]
self.buckets = list(range(self.BucketCount))
for each in self.buckets:
self.buckets[each] = 0
def tearDown(self):
win32trace.TermRead()
win32trace.TermWrite()
def areBucketsFull(self):
bucketsAreFull = True
for each in self.buckets:
assert each <= self.FullBucket, each
if each != self.FullBucket:
bucketsAreFull = False
break
return bucketsAreFull
def read(self):
while 1:
readString = win32trace.blockingread()
for ch in readString:
integer = int(ch)
count = self.buckets[integer]
assert count != -1
self.buckets[integer] = count + 1
if self.buckets[integer] == self.FullBucket:
if self.areBucketsFull():
return
def testThreads(self):
for each in self.threads:
each.start()
self.read()
for each in self.threads:
each.join()
for each in self.threads:
assert each.verifyWritten()
assert self.areBucketsFull()
class TestHugeChunks(unittest.TestCase):
# BiggestChunk is the size where we stop stressing the writer
BiggestChunk = 2**16 # 256k should do it.
def setUp(self):
win32trace.InitRead()
win32trace.read() # clear any old data
win32trace.InitWrite()
def testHugeChunks(self):
data = "*" * 1023 + "\n"
while len(data) <= self.BiggestChunk:
win32trace.write(data)
data = data + data
# If we made it here, we passed.
def tearDown(self):
win32trace.TermRead()
win32trace.TermWrite()
import win32event
import win32process
class TraceWriteProcess:
def __init__(self, threadCount):
self.exitCode = -1
self.threadCount = threadCount
def start(self):
procHandle, threadHandle, procId, threadId = win32process.CreateProcess(
None, # appName
'python.exe "%s" /run_test_process %s %s' % (this_file,
self.BucketCount,
self.threadCount),
None, # process security
None, # thread security
0, # inherit handles
win32process.NORMAL_PRIORITY_CLASS,
None, # new environment
None, # Current directory
win32process.STARTUPINFO(), # startup info
)
self.processHandle = procHandle
def join(self):
win32event.WaitForSingleObject(self.processHandle,
win32event.INFINITE)
self.exitCode = win32process.GetExitCodeProcess(self.processHandle)
def verifyWritten(self):
return self.exitCode == 0
class TestOutofProcess(unittest.TestCase):
BucketCount = 9
FullBucket = 50
def setUp(self):
win32trace.InitRead()
TraceWriteProcess.BucketCount = self.BucketCount
self.setUpWriters()
self.buckets = list(range(self.BucketCount))
for each in self.buckets:
self.buckets[each] = 0
def tearDown(self):
win32trace.TermRead()
def setUpWriters(self):
self.processes = []
# 5 processes, quot threads in each process
quot, remainder = divmod(self.FullBucket, 5)
for each in range(5):
self.processes.append(TraceWriteProcess(quot))
if remainder:
self.processes.append(TraceWriteProcess(remainder))
def areBucketsFull(self):
bucketsAreFull = True
for each in self.buckets:
assert each <= self.FullBucket, each
if each != self.FullBucket:
bucketsAreFull = False
break
return bucketsAreFull
def read(self):
while 1:
readString = win32trace.blockingread()
for ch in readString:
integer = int(ch)
count = self.buckets[integer]
assert count != -1
self.buckets[integer] = count + 1
if self.buckets[integer] == self.FullBucket:
if self.areBucketsFull():
return
def testProcesses(self):
for each in self.processes:
each.start()
self.read()
for each in self.processes:
each.join()
for each in self.processes:
assert each.verifyWritten()
assert self.areBucketsFull()
def _RunAsTestProcess():
# Run as an external process by the main tests.
WriterThread.BucketCount = int(sys.argv[2])
threadCount = int(sys.argv[3])
threads = [WriterThread() for each in range(threadCount)]
win32trace.InitWrite()
for t in threads:
t.start()
for t in threads:
t.join()
for t in threads:
if not t.verifyWritten():
sys.exit(-1)
if __name__ == '__main__':
if sys.argv[1:2]==["/run_test_process"]:
_RunAsTestProcess()
sys.exit(0)
# If some other win32traceutil reader is running, these tests fail
# badly (as the other reader sometimes sees the output!)
win32trace.InitRead()
win32trace.InitWrite()
CheckNoOtherReaders()
# reset state so test env is back to normal
win32trace.TermRead()
win32trace.TermWrite()
unittest.main()

View file

@ -0,0 +1,172 @@
import unittest
import win32wnet
import win32api
import netbios
from pywin32_testutil import str2bytes
RESOURCE_CONNECTED = 0x00000001
RESOURCE_GLOBALNET = 0x00000002
RESOURCE_REMEMBERED = 0x00000003
RESOURCE_RECENT = 0x00000004
RESOURCE_CONTEXT = 0x00000005
RESOURCETYPE_ANY = 0x00000000
RESOURCETYPE_DISK = 0x00000001
RESOURCETYPE_PRINT = 0x00000002
RESOURCETYPE_RESERVED = 0x00000008
RESOURCETYPE_UNKNOWN = 0xFFFFFFFF
RESOURCEUSAGE_CONNECTABLE = 0x00000001
RESOURCEUSAGE_CONTAINER = 0x00000002
RESOURCEDISPLAYTYPE_GENERIC = 0x00000000
RESOURCEDISPLAYTYPE_DOMAIN = 0x00000001
RESOURCEDISPLAYTYPE_SERVER = 0x00000002
RESOURCEDISPLAYTYPE_SHARE = 0x00000003
NETRESOURCE_attributes = [
("dwScope", int),
("dwType", int),
("dwDisplayType", int),
("dwUsage", int),
("lpLocalName", str),
("lpRemoteName", str),
("lpComment", str),
("lpProvider", str),
]
NCB_attributes = [
("Command", int),
("Retcode", int),
("Lsn", int),
("Num", int),
# ("Bufflen", int), - read-only
("Callname", str),
("Name", str),
("Rto", int),
("Sto", int),
("Lana_num", int),
("Cmd_cplt", int),
("Event", int),
("Post", int),
]
class TestCase(unittest.TestCase):
def testGetUser(self):
self.assertEquals(win32api.GetUserName(), win32wnet.WNetGetUser())
def _checkItemAttributes(self, item, attrs):
for attr, typ in attrs:
val = getattr(item, attr)
if typ is int:
self.failUnless(type(val) in (int,),
"Attr %r has value %r" % (attr, val))
new_val = val + 1
elif typ is str:
if val is not None:
# on py2k, must be string or unicode. py3k must be string or bytes.
self.failUnless(type(val) in (str, str),
"Attr %r has value %r" % (attr, val))
new_val = val + " new value"
else:
new_val = "new value"
else:
self.fail("Don't know what %s is" % (typ,))
# set the attribute just to make sure we can.
setattr(item, attr, new_val)
def testNETRESOURCE(self):
nr = win32wnet.NETRESOURCE()
self._checkItemAttributes(nr, NETRESOURCE_attributes)
def testWNetEnumResource(self):
handle = win32wnet.WNetOpenEnum(RESOURCE_GLOBALNET, RESOURCETYPE_ANY,
0, None)
try:
while 1:
items = win32wnet.WNetEnumResource(handle, 0)
if len(items)==0:
break
for item in items:
self._checkItemAttributes(item, NETRESOURCE_attributes)
finally:
handle.Close()
def testNCB(self):
ncb = win32wnet.NCB()
self._checkItemAttributes(ncb, NCB_attributes)
def testNetbios(self):
# taken from the demo code in netbios.py
ncb = win32wnet.NCB()
ncb.Command = netbios.NCBENUM
la_enum = netbios.LANA_ENUM()
ncb.Buffer = la_enum
rc = win32wnet.Netbios(ncb)
self.failUnlessEqual(rc, 0)
for i in range(la_enum.length):
ncb.Reset()
ncb.Command = netbios.NCBRESET
ncb.Lana_num = netbios.byte_to_int(la_enum.lana[i])
rc = Netbios(ncb)
self.failUnlessEqual(rc, 0)
ncb.Reset()
ncb.Command = netbios.NCBASTAT
ncb.Lana_num = byte_to_int(la_enum.lana[i])
ncb.Callname = str2bytes("* ") # ensure bytes on py2x and 3k
adapter = netbios.ADAPTER_STATUS()
ncb.Buffer = adapter
Netbios(ncb)
# expect 6 bytes in the mac address.
self.failUnless(len(adapter.adapter_address), 6)
def iterConnectableShares(self):
nr = win32wnet.NETRESOURCE()
nr.dwScope = RESOURCE_GLOBALNET
nr.dwUsage = RESOURCEUSAGE_CONTAINER
nr.lpRemoteName = "\\\\" + win32api.GetComputerName()
handle = win32wnet.WNetOpenEnum(RESOURCE_GLOBALNET, RESOURCETYPE_ANY,
0, nr)
while 1:
items = win32wnet.WNetEnumResource(handle, 0)
if len(items)==0:
break
for item in items:
if item.dwDisplayType == RESOURCEDISPLAYTYPE_SHARE:
yield item
def findUnusedDriveLetter(self):
existing = [x[0].lower() for x in win32api.GetLogicalDriveStrings().split('\0') if x]
handle = win32wnet.WNetOpenEnum(RESOURCE_REMEMBERED,RESOURCETYPE_DISK,0,None)
try:
while 1:
items = win32wnet.WNetEnumResource(handle, 0)
if len(items)==0:
break
xtra = [i.lpLocalName[0].lower() for i in items if i.lpLocalName]
existing.extend(xtra)
finally:
handle.Close()
for maybe in 'defghijklmnopqrstuvwxyz':
if maybe not in existing:
return maybe
self.fail("All drive mappings are taken?")
def testAddConnection(self):
localName = self.findUnusedDriveLetter() + ':'
for share in self.iterConnectableShares():
share.lpLocalName = localName
win32wnet.WNetAddConnection2(share)
win32wnet.WNetCancelConnection2(localName, 0, 0)
break
def testAddConnectionOld(self):
localName = self.findUnusedDriveLetter() + ':'
for share in self.iterConnectableShares():
win32wnet.WNetAddConnection2(share.dwType, localName, share.lpRemoteName)
win32wnet.WNetCancelConnection2(localName, 0, 0)
break
if __name__ == '__main__':
unittest.main()

View file

@ -0,0 +1,189 @@
import sys, os
import re
import unittest
import traceback
import pywin32_testutil
# A list of demos that depend on user-interface of *any* kind. Tests listed
# here are not suitable for unattended testing.
ui_demos = """GetSaveFileName print_desktop win32cred_demo win32gui_demo
win32gui_dialog win32gui_menu win32gui_taskbar
win32rcparser_demo winprocess win32console_demo
win32clipboard_bitmapdemo
win32gui_devicenotify
NetValidatePasswordPolicy""".split()
# Other demos known as 'bad' (or at least highly unlikely to work)
# cerapi: no CE module is built (CE via pywin32 appears dead)
# desktopmanager: hangs (well, hangs for 60secs or so...)
# EvtSubscribe_*: must be run together
bad_demos = """cerapi desktopmanager win32comport_demo
EvtSubscribe_pull EvtSubscribe_push
""".split()
argvs = {
"rastest": ("-l",),
}
no_user_interaction = False
# re to pull apart an exception line into the exception type and the args.
re_exception = re.compile("([a-zA-Z0-9_.]*): (.*)$")
def find_exception_in_output(data):
have_traceback = False
for line in data.splitlines():
line = line.decode('ascii') # not sure what the correct encoding is...
if line.startswith("Traceback ("):
have_traceback = True
continue
if line.startswith(" "):
continue
if have_traceback:
# first line not starting with a space since the traceback.
# must be the exception!
m = re_exception.match(line)
if m:
exc_type, args = m.groups()
# get hacky - get the *real* exception object from the name.
bits = exc_type.split(".", 1)
if len(bits) > 1:
mod = __import__(bits[0])
exc = getattr(mod, bits[1])
else:
# probably builtin
exc = eval(bits[0])
else:
# hrm - probably just an exception with no args
try:
exc = eval(line.strip())
args = "()"
except:
return None
# try and turn the args into real args.
try:
args = eval(args)
except:
pass
if not isinstance(args, tuple):
args = (args,)
# try and instantiate the exception.
try:
ret = exc(*args)
except:
ret = None
return ret
# apparently not - keep looking...
have_traceback = False
class TestRunner:
def __init__(self, argv):
self.argv = argv
self.__name__ = "Test Runner for cmdline {}".format(argv)
def __call__(self):
import subprocess
p = subprocess.Popen(self.argv,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
output, _ = p.communicate()
rc = p.returncode
if rc:
base = os.path.basename(self.argv[1])
# See if we can detect and reconstruct an exception in the output.
reconstituted = find_exception_in_output(output)
if reconstituted is not None:
raise reconstituted
raise AssertionError("%s failed with exit code %s. Output is:\n%s" % (base, rc, output))
def get_demo_tests():
import win32api
ret = []
demo_dir = os.path.abspath(os.path.join(os.path.dirname(win32api.__file__), "Demos"))
assert os.path.isdir(demo_dir), demo_dir
for name in os.listdir(demo_dir):
base, ext = os.path.splitext(name)
if base in ui_demos and no_user_interaction:
continue
# Skip any other files than .py and bad tests in any case
if ext != ".py" or base in bad_demos:
continue
argv = (sys.executable, os.path.join(demo_dir, base+".py")) + \
argvs.get(base, ())
ret.append(unittest.FunctionTestCase(TestRunner(argv), description="win32/demos/" + name))
return ret
def import_all():
# Some hacks for import order - dde depends on win32ui
try:
import win32ui
except ImportError:
pass # 'what-ev-a....'
import win32api
dir = os.path.dirname(win32api.__file__)
num = 0
is_debug = os.path.basename(win32api.__file__).endswith("_d")
for name in os.listdir(dir):
base, ext = os.path.splitext(name)
if (ext==".pyd") and \
name != "_winxptheme.pyd" and \
(is_debug and base.endswith("_d") or \
not is_debug and not base.endswith("_d")):
try:
__import__(base)
except:
print(("FAILED to import", name))
raise
num += 1
def suite():
# Loop over all .py files here, except me :)
try:
me = __file__
except NameError:
me = sys.argv[0]
me = os.path.abspath(me)
files = os.listdir(os.path.dirname(me))
suite = unittest.TestSuite()
suite.addTest(unittest.FunctionTestCase(import_all))
for file in files:
base, ext = os.path.splitext(file)
if ext=='.py' and os.path.basename(me) != file:
try:
mod = __import__(base)
except:
print("FAILED to import test module %r" % base)
traceback.print_exc()
continue
if hasattr(mod, "suite"):
test = mod.suite()
else:
test = unittest.defaultTestLoader.loadTestsFromModule(mod)
suite.addTest(test)
for test in get_demo_tests():
suite.addTest(test)
return suite
class CustomLoader(pywin32_testutil.TestLoader):
def loadTestsFromModule(self, module):
return self.fixupTestsForLeakTests(suite())
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description="Test runner for PyWin32/win32")
parser.add_argument("-no-user-interaction",
default=no_user_interaction,
action='store_true',
help="Run all tests without user interaction")
parsed_args, remains = parser.parse_known_args()
no_user_interaction = parsed_args.no_user_interaction
sys.argv = [sys.argv[0]] + remains
pywin32_testutil.testmain(testLoader=CustomLoader())

Binary file not shown.

After

Width:  |  Height:  |  Size: 778 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 766 B

View file

@ -0,0 +1,46 @@
//{{NO_DEPENDENCIES}}
// Microsoft Developer Studio generated include file.
// Used by test.rc
//
#define IDS_TEST_STRING1 51
#define IDS_TEST_STRING2 52
#define IDS_TEST_STRING3 53
#define IDS_TEST_STRING4 54
#define IDS_TEST_STRING5 55
#define IDS_TEST_STRING6 56
#define IDS_TEST_STRING7 57
#define IDD_TEST_DIALOG1 101
#define IDD_TEST_DIALOG2 102
#define IDB_PYTHON 103
#define IDI_PYTHON 105
#define IDD_TEST_DIALOG3 105
#define IDC_EDIT1 1000
#define IDC_CHECK1 1001
#define IDC_EDIT2 1001
#define IDC_COMBO1 1002
#define IDC_SPIN1 1003
#define IDC_PROGRESS1 1004
#define IDC_SLIDER1 1005
#define IDC_LIST1 1006
#define IDC_TREE1 1007
#define IDC_TAB1 1008
#define IDC_ANIMATE1 1009
#define IDC_RICHEDIT1 1010
#define IDC_DATETIMEPICKER1 1011
#define IDC_MONTHCALENDAR1 1012
#define IDC_SCROLLBAR1 1013
#define IDC_SCROLLBAR2 1014
#define IDC_LIST2 1015
#define IDC_HELLO 1016
#define IDC_HELLO2 1017
// Next default values for new objects
//
#ifdef APSTUDIO_INVOKED
#ifndef APSTUDIO_READONLY_SYMBOLS
#define _APS_NEXT_RESOURCE_VALUE 107
#define _APS_NEXT_COMMAND_VALUE 40002
#define _APS_NEXT_CONTROL_VALUE 1018
#define _APS_NEXT_SYMED_VALUE 101
#endif
#endif

View file

@ -0,0 +1,216 @@
//Microsoft Developer Studio generated resource script.
//
#include "test.h"
#define APSTUDIO_READONLY_SYMBOLS
/////////////////////////////////////////////////////////////////////////////
//
// Generated from the TEXTINCLUDE 2 resource.
//
#include "afxres.h"
/////////////////////////////////////////////////////////////////////////////
#undef APSTUDIO_READONLY_SYMBOLS
/////////////////////////////////////////////////////////////////////////////
// English (Australia) resources
#if !defined(AFX_RESOURCE_DLL) || defined(AFX_TARG_ENA)
#ifdef _WIN32
LANGUAGE LANG_ENGLISH, SUBLANG_ENGLISH_AUS
#pragma code_page(1252)
#endif //_WIN32
#ifdef APSTUDIO_INVOKED
/////////////////////////////////////////////////////////////////////////////
//
// TEXTINCLUDE
//
1 TEXTINCLUDE DISCARDABLE
BEGIN
"test.h\0"
END
2 TEXTINCLUDE DISCARDABLE
BEGIN
"#include ""afxres.h""\r\n"
"\0"
END
3 TEXTINCLUDE DISCARDABLE
BEGIN
"\r\n"
"\0"
END
#endif // APSTUDIO_INVOKED
/////////////////////////////////////////////////////////////////////////////
//
// Dialog
//
IDD_TEST_DIALOG1 DIALOG DISCARDABLE 0, 0, 186, 95
STYLE DS_MODALFRAME | WS_POPUP | WS_CAPTION | WS_SYSMENU
CAPTION "Test Dialog"
FONT 8, "MS Sans Serif"
BEGIN
DEFPUSHBUTTON "OK",IDOK,129,7,50,14
PUSHBUTTON "Cancel",IDCANCEL,129,24,50,14
ICON IDI_PYTHON,IDC_STATIC,142,47,21,20
LTEXT "An icon",IDC_STATIC,140,70,34,9
END
IDD_TEST_DIALOG2 DIALOG DISCARDABLE 0, 0, 186, 95
STYLE DS_MODALFRAME | WS_POPUP | WS_CAPTION | WS_SYSMENU
CAPTION "Test Dialog"
FONT 8, "MS Sans Serif"
BEGIN
DEFPUSHBUTTON "OK",IDOK,129,7,50,14
PUSHBUTTON "Cancel",IDCANCEL,129,24,50,14,NOT WS_TABSTOP
CONTROL 103,IDC_STATIC,"Static",SS_BITMAP,139,49,32,32
LTEXT "A bitmap",IDC_STATIC,135,72,34,9
EDITTEXT IDC_EDIT1,59,7,59,14,ES_AUTOHSCROLL
EDITTEXT IDC_EDIT2,59,31,60,15,ES_AUTOHSCROLL | NOT WS_TABSTOP
LTEXT "Tabstop",IDC_STATIC,7,9,43,10
LTEXT "Not Tabstop",IDC_STATIC,7,33,43,10
END
IDD_TEST_DIALOG3 DIALOGEX 0, 0, 232, 310
STYLE DS_MODALFRAME | WS_POPUP | WS_CAPTION | WS_SYSMENU
CAPTION "Dialog"
FONT 8, "MS Sans Serif", 0, 0, 0x1
BEGIN
GROUPBOX "Frame",IDC_STATIC,7,7,218,41
LTEXT "Left Static",IDC_STATIC,16,17,73,11
EDITTEXT IDC_EDIT1,103,15,112,12,ES_AUTOHSCROLL
LTEXT "Right Static",IDC_STATIC,16,30,73,11,0,WS_EX_RIGHT
CONTROL "",IDC_RICHEDIT1,"RICHEDIT",ES_AUTOHSCROLL | WS_BORDER |
WS_TABSTOP,103,31,113,14
CONTROL "Check1",IDC_CHECK1,"Button",BS_AUTOCHECKBOX |
WS_TABSTOP,7,52,68,12
COMBOBOX IDC_COMBO1,85,52,82,35,CBS_DROPDOWNLIST | CBS_SORT |
WS_VSCROLL | WS_TABSTOP
CONTROL "Spin1",IDC_SPIN1,"msctls_updown32",UDS_ARROWKEYS,7,71,
14,22
CONTROL "Progress1",IDC_PROGRESS1,"msctls_progress32",WS_BORDER,
39,72,153,13
SCROLLBAR IDC_SCROLLBAR2,207,55,13,57,SBS_VERT
CONTROL "Slider1",IDC_SLIDER1,"msctls_trackbar32",TBS_BOTH |
TBS_NOTICKS | WS_TABSTOP,35,91,159,7
SCROLLBAR IDC_SCROLLBAR1,37,102,155,11
CONTROL "Tab1",IDC_TAB1,"SysTabControl32",0x0,7,120,217,43
CONTROL "Animate1",IDC_ANIMATE1,"SysAnimate32",WS_BORDER |
WS_TABSTOP,7,171,46,42
CONTROL "List1",IDC_LIST1,"SysListView32",WS_BORDER | WS_TABSTOP,
63,171,53,43
CONTROL "Tree1",IDC_TREE1,"SysTreeView32",WS_BORDER | WS_TABSTOP,
126,171,50,43
CONTROL "MonthCalendar1",IDC_MONTHCALENDAR1,"SysMonthCal32",
MCS_NOTODAY | WS_TABSTOP,7,219,140,84
CONTROL "DateTimePicker1",IDC_DATETIMEPICKER1,"SysDateTimePick32",
DTS_RIGHTALIGN | WS_TABSTOP,174,221,51,15
DEFPUSHBUTTON "OK",IDOK,175,289,50,14
PUSHBUTTON "Hello",IDC_HELLO,175,271,50,14
PUSHBUTTON "Hello",IDC_HELLO2,175,240,50,26,BS_ICON
LISTBOX IDC_LIST2,184,171,40,45,LBS_SORT | LBS_NOINTEGRALHEIGHT |
WS_VSCROLL | WS_TABSTOP
END
/////////////////////////////////////////////////////////////////////////////
//
// DESIGNINFO
//
#ifdef APSTUDIO_INVOKED
GUIDELINES DESIGNINFO DISCARDABLE
BEGIN
IDD_TEST_DIALOG1, DIALOG
BEGIN
LEFTMARGIN, 7
RIGHTMARGIN, 179
TOPMARGIN, 7
BOTTOMMARGIN, 88
END
IDD_TEST_DIALOG2, DIALOG
BEGIN
LEFTMARGIN, 7
RIGHTMARGIN, 179
TOPMARGIN, 7
BOTTOMMARGIN, 88
END
IDD_TEST_DIALOG3, DIALOG
BEGIN
LEFTMARGIN, 7
RIGHTMARGIN, 225
TOPMARGIN, 7
BOTTOMMARGIN, 303
END
END
#endif // APSTUDIO_INVOKED
/////////////////////////////////////////////////////////////////////////////
//
// Icon
//
// Icon with lowest ID value placed first to ensure application icon
// remains consistent on all systems.
IDI_PYTHON ICON DISCARDABLE "python.ico"
/////////////////////////////////////////////////////////////////////////////
//
// Bitmap
//
IDB_PYTHON BITMAP DISCARDABLE "python.bmp"
/////////////////////////////////////////////////////////////////////////////
//
// Dialog Info
//
IDD_TEST_DIALOG3 DLGINIT
BEGIN
IDC_COMBO1, 0x403, 6, 0
0x7449, 0x6d65, 0x0031,
IDC_COMBO1, 0x403, 6, 0
0x7449, 0x6d65, 0x0032,
0
END
/////////////////////////////////////////////////////////////////////////////
//
// String Table
//
STRINGTABLE DISCARDABLE
BEGIN
IDS_TEST_STRING1 "Test ""quoted"" string"
IDS_TEST_STRING2 "Test string"
IDS_TEST_STRING3 "String with single "" quote"
IDS_TEST_STRING4 "Test 'single quoted' string"
END
#endif // English (Australia) resources
/////////////////////////////////////////////////////////////////////////////
#ifndef APSTUDIO_INVOKED
/////////////////////////////////////////////////////////////////////////////
//
// Generated from the TEXTINCLUDE 3 resource.
//
/////////////////////////////////////////////////////////////////////////////
#endif // not APSTUDIO_INVOKED