Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions com/win32comext/directsound/test/ds_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import win32api
import win32com.directsound.directsound as ds
import win32event
from pywin32_testutil import TestSkipped, find_test_fixture
from pywin32_testutil import find_test_fixture

# next two lines are for for debugging:
# import win32com
Expand Down Expand Up @@ -37,7 +37,7 @@ def wav_header_unpack(data):

assert riff == b"RIFF", "invalid wav header"

# fmt chuck is not first chunk, directly followed by data chuck
# fmt chunk is not first chunk, directly followed by data chunk
# It is nowhere required that they are, it is just very common
assert (
fmtsize == 16 and fmt == b"fmt " and data == b"data"
Expand Down Expand Up @@ -298,7 +298,7 @@ def testCreate(self):
except pythoncom.com_error as exc:
if exc.hresult != ds.DSERR_NODRIVER:
raise
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))

def testPlay(self):
"""Mesdames et Messieurs, la cour de Devin Dazzle"""
Expand All @@ -318,7 +318,7 @@ def testPlay(self):
except pythoncom.com_error as exc:
if exc.hresult != ds.DSERR_NODRIVER:
raise
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))
d.SetCooperativeLevel(None, ds.DSSCL_PRIORITY)

sdesc = ds.DSBUFFERDESC()
Expand Down Expand Up @@ -358,15 +358,15 @@ def testCreate(self):
except pythoncom.com_error as exc:
if exc.hresult != ds.DSERR_NODRIVER:
raise
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))

def testRecord(self):
try:
d = ds.DirectSoundCaptureCreate(None, None)
except pythoncom.com_error as exc:
if exc.hresult != ds.DSERR_NODRIVER:
raise
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))

sdesc = ds.DSCBUFFERDESC()
sdesc.dwBufferBytes = 352800 # 2 seconds
Expand Down
88 changes: 34 additions & 54 deletions win32/Lib/pywin32_testutil.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,23 @@
# Utilities for the pywin32 tests
"""Utilities for the pywin32 tests"""

from __future__ import annotations

import gc
import os
import site
import sys
import unittest
from collections import Counter
from typing import TYPE_CHECKING

import pythoncom
import pywintypes
import winerror
from pythoncom import _GetGatewayCount, _GetInterfaceCount
from win32com.shell.shell import IsUserAnAdmin

if TYPE_CHECKING:
from _typeshed import OptExcInfo

##
## unittest related stuff
Expand Down Expand Up @@ -37,10 +49,7 @@ def countTestCases(self):
return self.num_test_cases

def __call__(self, result=None):
# For the COM suite's sake, always ensure we don't leak
# gateways/interfaces
from pythoncom import _GetGatewayCount, _GetInterfaceCount

# For the COM suite's sake, always ensure we don't leak gateways/interfaces
gc.collect()
ni = _GetInterfaceCount()
ng = _GetGatewayCount()
Expand Down Expand Up @@ -145,20 +154,17 @@ def loadTestsFromName(self, name, module=None):

# win32 error codes that probably mean we need to be elevated (ie, if we
# aren't elevated, we treat these error codes as 'skipped')
non_admin_error_codes = [
non_admin_error_codes = {
winerror.ERROR_ACCESS_DENIED,
winerror.ERROR_PRIVILEGE_NOT_HELD,
]
}

_is_admin = None
_is_admin: bool | None = None


def check_is_admin():
def check_is_admin() -> bool:
global _is_admin
if _is_admin is None:
import pythoncom
from win32com.shell.shell import IsUserAnAdmin

try:
_is_admin = IsUserAnAdmin()
except pythoncom.com_error as exc:
Expand Down Expand Up @@ -197,30 +203,19 @@ def find_test_fixture(basename, extra_dir="."):
d = os.path.normcase(d)
if os.path.commonprefix([this_file, d]) == d:
# looks like we are in an installed Python, so skip the text.
raise TestSkipped(f"Can't find test fixture '{fname}'")
raise unittest.SkipTest(f"Can't find test fixture '{fname}'")
# Looks like we are running from source, so this is fatal.
raise RuntimeError(f"Can't find test fixture '{fname}'")


# If this exception is raised by a test, the test is reported as a 'skip'
class TestSkipped(Exception):
pass


# The 'TestResult' subclass that records the failures and has the special
# handling for the TestSkipped exception.
# handling for the unittest.SkipTest exception.
class TestResult(unittest.TextTestResult):
def __init__(self, *args, **kw):
super().__init__(*args, **kw)
self.skips = {} # count of skips for each reason.
def addError(self, test: unittest.TestCase, err: OptExcInfo) -> None:
"""Called when an error has occurred.

def addError(self, test, err):
"""Called when an error has occurred. 'err' is a tuple of values as
returned by sys.exc_info().
Translate a couple of 'well-known' exceptions into 'skipped'
"""
# translate a couple of 'well-known' exceptions into 'skipped'
import pywintypes

exc_val = err[1]
# translate ERROR_ACCESS_DENIED for non-admin users to be skipped.
# (access denied errors for an admin user aren't expected.)
Expand All @@ -229,42 +224,27 @@ def addError(self, test, err):
and exc_val.winerror in non_admin_error_codes
and not check_is_admin()
):
exc_val = TestSkipped(exc_val)
return self.addSkip(test, str(exc_val))
# and COM errors due to objects not being registered (the com test
# suite will attempt to catch this and handle it itself if the user
# is admin)
elif isinstance(exc_val, pywintypes.com_error) and exc_val.hresult in [
elif isinstance(exc_val, pywintypes.com_error) and exc_val.hresult in {
winerror.CO_E_CLASSSTRING,
winerror.REGDB_E_CLASSNOTREG,
winerror.TYPE_E_LIBNOTREGISTERED,
]:
exc_val = TestSkipped(exc_val)
# NotImplemented generally means the platform doesn't support the
# functionality.
}:
return self.addSkip(test, str(exc_val))
# NotImplemented generally means the platform doesn't support the functionality.
elif isinstance(exc_val, NotImplementedError):
exc_val = TestSkipped(NotImplementedError)

if isinstance(exc_val, TestSkipped):
reason = exc_val.args[0]
# if the reason itself is another exception, get its args.
try:
reason = tuple(reason.args)
except (AttributeError, TypeError):
pass
self.skips.setdefault(reason, 0)
self.skips[reason] += 1
if self.showAll:
self.stream.writeln(f"SKIP ({reason})")
elif self.dots:
self.stream.write("S")
self.stream.flush()
return
return self.addSkip(test, str(exc_val))

super().addError(test, err)

def printErrors(self):
def printErrors(self) -> None:
super().printErrors()
for reason, num_skipped in self.skips.items():
self.stream.writeln("SKIPPED: %d tests - %s" % (num_skipped, reason))
reasons = [reason for (_, reason) in self.skipped]
for reason, num_skipped in Counter(reasons).items():
self.stream.writeln(f"SKIPPED: {num_skipped} tests - {reason}")


# TestRunner subclass necessary just to get our TestResult hooked up.
Expand Down
3 changes: 1 addition & 2 deletions win32/test/test_odbc.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@

import odbc
import pythoncom
from pywin32_testutil import TestSkipped
from win32com.client import constants

# We use the DAO ODBC driver
Expand All @@ -33,7 +32,7 @@ def setUp(self):
except pythoncom.com_error:
pass
else:
raise TestSkipped("Can't find a DB engine")
raise unittest.SkipTest("Can't find a DB engine")

workspace = dbe.Workspaces(0)

Expand Down
8 changes: 4 additions & 4 deletions win32/test/test_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import win32con
import win32security
import winerror
from pywin32_testutil import TestSkipped, testmain
from pywin32_testutil import testmain


class SecurityTests(unittest.TestCase):
Expand All @@ -27,7 +27,7 @@ def tearDown(self):

def testEqual(self):
if self.admin_sid is None:
raise TestSkipped("No 'Administrator' account is available")
raise unittest.SkipTest("No 'Administrator' account is available")
self.assertEqual(
win32security.LookupAccountName("", "Administrator")[0],
win32security.LookupAccountName("", "Administrator")[0],
Expand All @@ -51,7 +51,7 @@ def testSIDInDict(self):

def testBuffer(self):
if self.admin_sid is None:
raise TestSkipped("No 'Administrator' account is available")
raise unittest.SkipTest("No 'Administrator' account is available")
self.assertEqual(
memoryview(win32security.LookupAccountName("", "Administrator")[0]),
memoryview(win32security.LookupAccountName("", "Administrator")[0]),
Expand Down Expand Up @@ -97,7 +97,7 @@ def setUp(self):
except win32security.error as exc:
if exc.winerror != winerror.ERROR_NO_SUCH_DOMAIN:
raise
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))

def tearDown(self):
if self.ds_handle is not None:
Expand Down
4 changes: 2 additions & 2 deletions win32/test/test_sspi.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import sspicon
import win32api
import win32security
from pywin32_testutil import TestSkipped, testmain
from pywin32_testutil import testmain


# It is quite likely that the Kerberos tests will fail due to not being
Expand All @@ -21,7 +21,7 @@ def applyHandlingSkips(func, *args):
sspicon.SEC_E_NO_CREDENTIALS,
sspicon.SEC_E_NO_AUTHENTICATING_AUTHORITY,
]:
raise TestSkipped(exc)
raise unittest.SkipTest(str(exc))
raise


Expand Down
3 changes: 1 addition & 2 deletions win32/test/test_win32api.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
import win32con
import win32event
import winerror
from pywin32_testutil import TestSkipped


class TestError(Exception):
Expand All @@ -23,7 +22,7 @@ def testGetCurrentUser(self):
domain = win32api.GetDomainName()
if domain == "NT AUTHORITY":
# Running as a service account, so the comparison will fail
raise TestSkipped("running as service account")
raise unittest.SkipTest("running as service account")
name = f"{domain}\\{win32api.GetUserName()}"
self.assertEqual(name, win32api.GetUserNameEx(win32api.NameSamCompatible))

Expand Down
14 changes: 9 additions & 5 deletions win32/test/test_win32file.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
import win32pipe
import win32timezone
import winerror
from pywin32_testutil import TestSkipped, testmain
from pywin32_testutil import testmain


class TestReadBuffer(unittest.TestCase):
Expand Down Expand Up @@ -802,15 +802,17 @@ def test_connect_with_payload(self):
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == 10022: # WSAEINVAL
raise TestSkipped("ConnectEx is not available on this platform")
raise unittest.SkipTest("ConnectEx is not available on this platform")
raise # some error error we don't expect.
# We occasionally see ERROR_CONNECTION_REFUSED in automation
try:
win32file.GetOverlappedResult(s2.fileno(), ol, 1)
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == winerror.ERROR_CONNECTION_REFUSED:
raise TestSkipped("Assuming ERROR_CONNECTION_REFUSED is transient")
raise unittest.SkipTest(
"Assuming ERROR_CONNECTION_REFUSED is transient"
)
raise
ol = pywintypes.OVERLAPPED()
buff = win32file.AllocateReadBuffer(1024)
Expand All @@ -837,15 +839,17 @@ def test_connect_without_payload(self):
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == 10022: # WSAEINVAL
raise TestSkipped("ConnectEx is not available on this platform")
raise unittest.SkipTest("ConnectEx is not available on this platform")
raise # some error error we don't expect.
# We occasionally see ERROR_CONNECTION_REFUSED in automation
try:
win32file.GetOverlappedResult(s2.fileno(), ol, 1)
except win32file.error as exc:
win32event.SetEvent(giveup_event)
if exc.winerror == winerror.ERROR_CONNECTION_REFUSED:
raise TestSkipped("Assuming ERROR_CONNECTION_REFUSED is transient")
raise unittest.SkipTest(
"Assuming ERROR_CONNECTION_REFUSED is transient"
)
raise

ol = pywintypes.OVERLAPPED()
Expand Down
4 changes: 2 additions & 2 deletions win32/test/test_win32inet.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import unittest

import winerror
from pywin32_testutil import TestSkipped, testmain
from pywin32_testutil import testmain
from win32inet import (
FtpCommand,
InternetCanonicalizeUrl,
Expand Down Expand Up @@ -101,7 +101,7 @@ def testFtpCommand(self):
finally:
hcon.Close()
except error as e:
raise TestSkipped(e)
raise unittest.SkipTest(str(e))


if __name__ == "__main__":
Expand Down
13 changes: 7 additions & 6 deletions win32/test/test_win32trace.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,20 +5,21 @@
import unittest

import win32trace
from pywin32_testutil import TestSkipped

if __name__ == "__main__":
this_file = sys.argv[0]
else:
this_file = __file__


def SkipIfCI():
# This test often fails in CI, probably when it is being run multiple times
# (ie, for different Python versions)
# Github actions always have a `CI` variable.
def SkipIfCI() -> None:
"""For tests that often fails in CI, probably when it is being run multiple times
(ie, for different Python versions)

Github actions always have a `CI` variable.
"""
if "CI" in os.environ:
raise TestSkipped("We skip this test on CI")
raise unittest.SkipTest("We skip this test on CI")


def CheckNoOtherReaders():
Expand Down