Python win32api 模块，error() 实例源码

def GetDomainName():
    try:
        tok = win32security.OpenThreadToken(win32api.GetCurrentThread(),
                                            TOKEN_QUERY, 1)
    except win32api.error, details:
        if details[0] != winerror.ERROR_NO_TOKEN:
            raise
        # attempt to open the process token, since no thread token
        # exists
        tok = win32security.OpenProcessToken(win32api.GetCurrentProcess(),
                                             TOKEN_QUERY)
    sid, attr = win32security.GetTokenInformation(tok, TokenUser)
    win32api.CloseHandle(tok)

    name, dom, typ = win32security.LookupAccountSid(None, sid)
    return dom

def __init__ (self):
        wx.Frame.__init__ (self, None, title="Clipboard viewer", size=(250,150))

        self.first   = True
        self.nextWnd = None

        # Get native window handle of this wxWidget Frame.
        self.hwnd    = self.GetHandle ()

        # Set the WndProc to our function.
        self.oldWndProc = win32gui.SetWindowLong (self.hwnd,
                                                  win32con.GWL_WNDPROC,
                                                  self.MyWndProc)

        try:
            self.nextWnd = win32clipboard.SetClipboardViewer (self.hwnd)
        except win32api.error:
            if win32api.GetLastError () == 0:
                # information that there is no other window in chain
                pass
            else:
                raise

def MyWndProc (self, hWnd, msg, wParam, lParam):
        if msg == win32con.WM_CHANGECBCHAIN:
            self.OnChangeCBChain (msg, wParam, lParam)
        elif msg == win32con.WM_DRAWCLIPBOARD:
            self.OnDrawClipboard (msg, wParam, lParam)

        # Restore the old WndProc. Notice the use of win32api
        # instead of win32gui here. This is to avoid an error due to
        # not passing a callable object.
        if msg == win32con.WM_DESTROY:
            if self.nextWnd:
               win32clipboard.ChangeClipboardChain (self.hwnd, self.nextWnd)
            else:
               win32clipboard.ChangeClipboardChain (self.hwnd, 0)

            win32api.SetWindowLong (self.hwnd,
                                    win32con.GWL_WNDPROC,
                                    self.oldWndProc)

        # Pass all messages (in this case, yours may be different) on
        # to the original WndProc
        return win32gui.CallWindowProc (self.oldWndProc,
                                        hWnd, msg, wParam, lParam)

def checkWork(self):
        finished = 0
        fullDataRead = []

        while 1:
            try:
                buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.pipe, 1)
                # finished = (result == -1)
                if not bytesToRead:
                    break
                hr, data = win32file.ReadFile(self.pipe, bytesToRead, None)
                fullDataRead.append(data)
            except win32api.error:
                finished = 1
                break

        dataBuf = ''.join(fullDataRead)
        if dataBuf:
            self.receivedCallback(dataBuf)
        if finished:
            self.cleanup()
        return len(dataBuf)

def RegisterCoreDLL(coredllName = None):
    """Registers the core DLL in the registry.

        If no params are passed, the name of the Python DLL used in 
        the current process is used and registered.
    """
    if coredllName is None:
        coredllName = win32api.GetModuleFileName(sys.dllhandle)
        # must exist!
    else:
        try:
            os.stat(coredllName)
        except os.error:
            print "Warning: Registering non-existant core DLL %s" % coredllName

    hKey = win32api.RegCreateKey(GetRootKey() , BuildDefaultPythonKey())
    try:
        win32api.RegSetValue(hKey, "Dll", win32con.REG_SZ, coredllName)
    finally:
        win32api.RegCloseKey(hKey)
    # Lastly, setup the current version to point to me.
    win32api.RegSetValue(GetRootKey(), "Software\\Python\\PythonCore\\CurrentVersion", win32con.REG_SZ, sys.winver)

def __init__(self, writePipe, lostCallback):
        self.disconnecting = False
        self.producer = None
        self.producerPaused = 0
        self.streamingProducer = 0
        self.outQueue = []
        self.writePipe = writePipe
        self.lostCallback = lostCallback
        try:
            win32pipe.SetNamedPipeHandleState(writePipe,
                                              win32pipe.PIPE_NOWAIT,
                                              None,
                                              None)
        except pywintypes.error:
            # Maybe it's an invalid handle.  Who knows.
            pass

def RegisterPythonExe(exeFullPath, exeAlias = None, exeAppPath = None):
    """Register a .exe file that uses Python.

       Registers the .exe with the OS.  This allows the specified .exe to
       be run from the command-line or start button without using the full path,
       and also to setup application specific path (ie, os.environ['PATH']).

       Currently the exeAppPath is not supported, so this function is general
       purpose, and not specific to Python at all.  Later, exeAppPath may provide
       a reasonable default that is used.

       exeFullPath -- The full path to the .exe
       exeAlias = None -- An alias for the exe - if none, the base portion
                 of the filename is used.
       exeAppPath -- Not supported.
    """
    # Note - Dont work on win32s (but we dont care anymore!)
    if exeAppPath:
        raise error("Do not support exeAppPath argument currently")
    if exeAlias is None:
        exeAlias = os.path.basename(exeFullPath)
    win32api.RegSetValue(GetRootKey(), GetAppPathsKey() + "\\" + exeAlias, win32con.REG_SZ, exeFullPath)

def RegisterModule(modName, modPath):
    """Register an explicit module in the registry.  This forces the Python import
           mechanism to locate this module directly, without a sys.path search.  Thus
           a registered module need not appear in sys.path at all.

       modName -- The name of the module, as used by import.
       modPath -- The full path and file name of the module.
    """
    try:
        import os
        os.stat(modPath)
    except os.error:
        print "Warning: Registering non-existant module %s" % modPath
    win32api.RegSetValue(GetRootKey(), 
                         BuildDefaultPythonKey() + "\\Modules\\%s" % modName,
        win32con.REG_SZ, modPath)

def addcounter(self,object, counter, instance = None, inum=-1, machine=None):
        '''
        Adds a single counter path to the paths attribute.  Normally
        this will be called by a child class' speciality functions,
        rather than being called directly by the user. (Though it isn't
        hard to call manually, since almost everything is given a default)
        This method is only functional when the query is closed (or hasn't
        yet been opened).  This is to prevent conflict in multi-threaded
        query applications).
        e.g.:
            query.addcounter('Memory','Available Bytes')
        '''
        if not self.active:
            try:
                self.rawaddcounter(object, counter, instance, inum, machine)
                return 0
            except win32api.error:
                return -1
        else:
            return -1

def collectdataslave(self,format = win32pdh.PDH_FMT_LONG):
        '''
        ### Not a public method
        Called only when the Query is known to be open, runs over
        the whole set of counters, appending results to the temp,
        returns the values as a list.
        '''
        try:
            win32pdh.CollectQueryData(self._base)
            temp = []
            for counter in self.counters:
                ok = 0
                try:
                    if counter:
                        temp.append(win32pdh.GetFormattedCounterValue(counter, format)[1])
                        ok = 1
                except win32api.error:
                    pass
                if not ok:
                    temp.append(-1) # a better way to signal failure???
            return temp
        except win32api.error: # will happen if, for instance, no counters are part of the query and we attempt to collect data for it.
            return [-1] * len(self.counters)
    # pickle functions

def open(self,*args,**namedargs):
        '''
        Explicitly open a query:
        When you are needing to make multiple calls to the same query,
        it is most efficient to open the query, run all of the calls,
        then close the query, instead of having the collectdata method
        automatically open and close the query each time it runs.
        There are currently no arguments to open.
        '''
        # do all the normal opening stuff, self._base is now the query object
        BaseQuery.open(*(self,)+args, **namedargs)
        # should rewrite getinstpaths to take a single tuple
        paths = []
        for tup in self.volatilecounters:
            paths[len(paths):] = self.getinstpaths(*tup)
        for path in paths:
            try:
                self.counters.append(win32pdh.AddCounter(self._base, path))
                self.curpaths.append(path) # if we fail on the line above, this path won't be in the table or the counters
            except win32api.error:
                pass # again, what to do with a malformed path???

def _GetServiceShortName(longName):
    # looks up a services name
    # from the display name
    # Thanks to Andy McKay for this code.
    access = win32con.KEY_READ | win32con.KEY_ENUMERATE_SUB_KEYS | win32con.KEY_QUERY_VALUE
    hkey = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "SYSTEM\\CurrentControlSet\\Services", 0, access)
    num = win32api.RegQueryInfoKey(hkey)[0]
    longName = longName.lower()
    # loop through number of subkeys
    for x in range(0, num):
    # find service name, open subkey
        svc = win32api.RegEnumKey(hkey, x)
        skey = win32api.RegOpenKey(hkey, svc, 0, access)
        try:
            # find display name
            thisName = str(win32api.RegQueryValueEx(skey, "DisplayName")[0])
            if thisName.lower() == longName:
                return svc
        except win32api.error:
            # in case there is no key called DisplayName
            pass
    return None

# Open a service given either it's long or short name.

def RemoveService(serviceName):
    try:
        import perfmon
        perfmon.UnloadPerfCounterTextStrings("python.exe "+serviceName)
    except (ImportError, win32api.error):
        pass

    hscm = win32service.OpenSCManager(None,None,win32service.SC_MANAGER_ALL_ACCESS)
    try:
        hs = SmartOpenService(hscm, serviceName, win32service.SERVICE_ALL_ACCESS)
        win32service.DeleteService(hs)
        win32service.CloseServiceHandle(hs)
    finally:
        win32service.CloseServiceHandle(hscm)

    import win32evtlogutil
    try:
        win32evtlogutil.RemoveSourceFromRegistry(serviceName)
    except win32api.error:
        pass

def __FindSvcDeps(findName):
    if type(findName) is pywintypes.UnicodeType: findName = str(findName)
    dict = {}
    k = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "SYSTEM\\CurrentControlSet\\Services")
    num = 0
    while 1:
        try:
            svc = win32api.RegEnumKey(k, num)
        except win32api.error:
            break
        num = num + 1
        sk = win32api.RegOpenKey(k, svc)
        try:
            deps, typ = win32api.RegQueryValueEx(sk, "DependOnService")
        except win32api.error:
            deps = ()
        for dep in deps:
            dep = dep.lower()
            dep_on = dict.get(dep, [])
            dep_on.append(svc)
            dict[dep]=dep_on

    return __ResolveDeps(findName, dict)

def RestartService(serviceName, args = None, waitSeconds = 30, machine = None):
    "Stop the service, and then start it again (with some tolerance for allowing it to stop.)"
    try:
        StopService(serviceName, machine)
    except pywintypes.error, exc:
        # Allow only "service not running" error
        if exc.winerror!=winerror.ERROR_SERVICE_NOT_ACTIVE:
            raise
    # Give it a few goes, as the service may take time to stop
    for i in range(waitSeconds):
        try:
            StartService(serviceName, args, machine)
            break
        except pywintypes.error, exc:
            if exc.winerror!=winerror.ERROR_SERVICE_ALREADY_RUNNING:
                raise
            win32api.Sleep(1000)
    else:
        print "Gave up waiting for the old service to stop!"

def GetServiceClassString(cls, argv = None):
    if argv is None:
        argv = sys.argv
    import pickle
    modName = pickle.whichmodule(cls, cls.__name__)
    if modName == '__main__':
        try:
            fname = win32api.GetFullPathName(argv[0])
            path = os.path.split(fname)[0]
            # Eaaaahhhh - sometimes this will be a short filename, which causes
            # problems with 1.5.1 and the silly filename case rule.
            # Get the long name
            fname = os.path.join(path, win32api.FindFiles(fname)[0][8])
        except win32api.error:
            raise error("Could not resolve the path name '%s' to a full path" % (argv[0]))
        modName = os.path.splitext(fname)[0]
    return modName + "." + cls.__name__

def _find_localserver_module():
  import win32com.server
  path = win32com.server.__path__[0]
  baseName = "localserver"
  pyfile = os.path.join(path, baseName + ".py")
  try:
    os.stat(pyfile)
  except os.error:
    # See if we have a compiled extension
    if __debug__:
      ext = ".pyc"
    else:
      ext = ".pyo"
    pyfile = os.path.join(path, baseName + ext)
    try:
      os.stat(pyfile)
    except os.error:
      raise RuntimeError("Can not locate the Python module 'win32com.server.%s'" % baseName)
  return pyfile

def EnumKeys(root):
    index = 0
    ret = []
    while 1:
        try:
            item = win32api.RegEnumKey(root, index)
        except win32api.error:
            break
        try:
            # Note this doesn't handle REG_EXPAND_SZ, but the implementation
            # here doesn't need to - that is handled as the data is read.
            val = win32api.RegQueryValue(root, item)
        except win32api.error:
            val = "" # code using this assumes a string.

        ret.append((item, val))
        index = index + 1
    return ret

def GetDefaultProfileName():
    import win32api, win32con
    try:
        key = win32api.RegOpenKey(win32con.HKEY_CURRENT_USER, "Software\\Microsoft\\Windows NT\\CurrentVersion\\Windows Messaging Subsystem\\Profiles")
        try:
            return win32api.RegQueryValueEx(key, "DefaultProfile")[0]
        finally:
            key.Close()
    except win32api.error:
        return None

#
# Recursive dump of folders.
#

def test():
    import win32com.client
    oldcwd = os.getcwd()
    try:
        session = gencache.EnsureDispatch("MAPI.Session")
        try:
            session.Logon(GetDefaultProfileName())
        except pythoncom.com_error, details:
            print "Could not log on to MAPI:", details
            return
    except pythoncom.error:
        # no mapi.session - let's try outlook
        app = gencache.EnsureDispatch("Outlook.Application")
        session = app.Session

    try:
        TestUser(session)
        TestAddress(session)
        DumpFolders(session)
    finally:
        session.Logoff()
        # It appears Exchange will change the cwd on us :(
        os.chdir(oldcwd)

def _find_localserver_module():
  import win32com.server
  path = win32com.server.__path__[0]
  baseName = "localserver"
  pyfile = os.path.join(path, baseName + ".py")
  try:
    os.stat(pyfile)
  except os.error:
    # See if we have a compiled extension
    if __debug__:
      ext = ".pyc"
    else:
      ext = ".pyo"
    pyfile = os.path.join(path, baseName + ext)
    try:
      os.stat(pyfile)
    except os.error:
      raise RuntimeError("Can not locate the Python module 'win32com.server.%s'" % baseName)
  return pyfile

def EnumKeys(root):
    index = 0
    ret = []
    while 1:
        try:
            item = win32api.RegEnumKey(root, index)
        except win32api.error:
            break
        try:
            # Note this doesn't handle REG_EXPAND_SZ, but the implementation
            # here doesn't need to - that is handled as the data is read.
            val = win32api.RegQueryValue(root, item)
        except win32api.error:
            val = "" # code using this assumes a string.

        ret.append((item, val))
        index = index + 1
    return ret

def test():
    import win32com.client
    oldcwd = os.getcwd()
    try:
        session = gencache.EnsureDispatch("MAPI.Session")
        try:
            session.Logon(GetDefaultProfileName())
        except pythoncom.com_error, details:
            print "Could not log on to MAPI:", details
            return
    except pythoncom.error:
        # no mapi.session - let's try outlook
        app = gencache.EnsureDispatch("Outlook.Application")
        session = app.Session

    try:
        TestUser(session)
        TestAddress(session)
        DumpFolders(session)
    finally:
        session.Logoff()
        # It appears Exchange will change the cwd on us :(
        os.chdir(oldcwd)

def checkWork(self):
        finished = 0
        fullDataRead = []

        while 1:
            try:
                buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.pipe, 1)
                # finished = (result == -1)
                if not bytesToRead:
                    break
                hr, data = win32file.ReadFile(self.pipe, bytesToRead, None)
                fullDataRead.append(data)
            except win32api.error:
                finished = 1
                break

        dataBuf = ''.join(fullDataRead)
        if dataBuf:
            self.receivedCallback(dataBuf)
        if finished:
            self.cleanup()
        return len(dataBuf)

def __init__(self, writePipe, lostCallback):
        self.disconnecting = False
        self.producer = None
        self.producerPaused = 0
        self.streamingProducer = 0
        self.outQueue = []
        self.writePipe = writePipe
        self.lostCallback = lostCallback
        try:
            win32pipe.SetNamedPipeHandleState(writePipe,
                                              win32pipe.PIPE_NOWAIT,
                                              None,
                                              None)
        except pywintypes.error:
            # Maybe it's an invalid handle.  Who knows.
            pass

def JobError(self, job, error):
        print 'Job Error', job, error
        f = error.GetFile()
        print 'While downloading', f.GetRemoteName()
        print 'To', f.GetLocalName()
        print 'The following error happened:'
        self._print_error(error)
        if f.GetRemoteName().endswith('missing-favicon.ico'):
            print 'Changing to point to correct file'
            f2 = f.QueryInterface(bits.IID_IBackgroundCopyFile2)
            favicon = 'http://www.python.org/favicon.ico'
            print 'Changing RemoteName from', f2.GetRemoteName(), 'to', favicon
            f2.SetRemoteName(favicon)
            job.Resume()
        else:
            job.Cancel()

def __init__(self, module):
        self.module = module
        if hasattr(module, '__file__'):
            fname = self.module.__file__
            # Check for .pyc or .pyo or even .pys!
            if fname[-1] in ['O','o','C','c', 'S', 's']: fname = fname[:-1]
            try:
                fname = win32api.GetFullPathName(fname)
            except win32api.error:
                pass
        else:
            if module.__name__=='__main__' and len(sys.argv)>0:
                fname = sys.argv[0]
            else:
                fname = "<Unknown!>"
        SourceCodeContainer.__init__(self, None, fname)

def FindPackagePath(packageName, knownFileName, searchPaths):
    """Find a package.

       Given a ni style package name, check the package is registered.

       First place looked is the registry for an existing entry.  Then
       the searchPaths are searched.
    """
    import regutil, os
    pathLook = regutil.GetRegisteredNamedPath(packageName)
    if pathLook and IsPackageDir(pathLook, packageName, knownFileName):
        return pathLook, None # The currently registered one is good.
    # Search down the search paths.
    for pathLook in searchPaths:
        if IsPackageDir(pathLook, packageName, knownFileName):
            # Found it
            ret = os.path.abspath(pathLook)
            return ret, ret
    raise error("The package %s can not be located" % packageName)

def FindHelpPath(helpFile, helpDesc, searchPaths):
    # See if the current registry entry is OK
    import os, win32api, win32con
    try:
        key = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "Software\\Microsoft\\Windows\\Help", 0, win32con.KEY_ALL_ACCESS)
        try:
            try:
                path = win32api.RegQueryValueEx(key, helpDesc)[0]
                if FileExists(os.path.join(path, helpFile)):
                    return os.path.abspath(path)
            except win32api.error:
                pass # no registry entry.
        finally:
            key.Close()
    except win32api.error:
        pass
    for pathLook in searchPaths:
        if FileExists(os.path.join(pathLook, helpFile)):
            return os.path.abspath(pathLook)
        pathLook = os.path.join(pathLook, "Help")
        if FileExists(os.path.join( pathLook, helpFile)):
            return os.path.abspath(pathLook)
    raise error("The help file %s can not be located" % helpFile)

def FindAppPath(appName, knownFileName, searchPaths):
    """Find an application.

     First place looked is the registry for an existing entry.  Then
     the searchPaths are searched.
    """
    # Look in the first path.
    import regutil, string, os
    regPath = regutil.GetRegisteredNamedPath(appName)
    if regPath:
        pathLook = regPath.split(";")[0]
    if regPath and FileExists(os.path.join(pathLook, knownFileName)):
        return None # The currently registered one is good.
    # Search down the search paths.
    for pathLook in searchPaths:
        if FileExists(os.path.join(pathLook, knownFileName)):
            # Found it
            return os.path.abspath(pathLook)
    raise error("The file %s can not be located for application %s" % (knownFileName, appName))

def FindRegisterApp(appName, knownFiles, searchPaths):
    """Find and Register a package.

       Assumes the core registry setup correctly.

    """
    import regutil, string
    if type(knownFiles)==type(''):
        knownFiles = [knownFiles]
    paths=[]
    try:
        for knownFile in knownFiles:
            pathLook = FindAppPath(appName, knownFile, searchPaths)
            if pathLook:
                paths.append(pathLook)
    except error, details:
        print "*** ", details
        return

    regutil.RegisterNamedPath(appName, ";".join(paths))

def dir(args):
    """dir directory_name ...
    Perform a directory listing on the remote device
    """
    bRecurse = 0
    try:
        opts, args = getopt.getopt(args, "r")
    except getopt.error, details:
        raise InvalidUsage(details)
    for o, v in opts:
        if o=="-r":
            bRecurse=1
    for arg in args:
        print "Directory of WCE:%s" % arg
        files = BuildFileList(arg, 0, bRecurse, _dirfilter, None)
        total_size=0
        for full_name, info, rel_name in files:
            date_str = info[3].Format("%d-%b-%Y %H:%M")
            attr_string = "     "
            if info[0] & win32con.FILE_ATTRIBUTE_DIRECTORY: attr_string = "<DIR>"
            print "%s  %s %10d %s" % (date_str, attr_string, info[5], rel_name)
            total_size = total_size + info[5]
        print " " * 14 + "%3d files, %10d bytes" % (len(files), total_size)

def walk(arg, dirname, names):
  global numStamped
  vars, debug, descriptions = arg
  for name in names:
    for pat in g_patterns:
      if fnmatch.fnmatch(name, pat):
        # Handle the "_d" thing.
        pathname = os.path.join(dirname, name)
        base, ext = os.path.splitext(name)
        if base[-2:]=='_d':
          name = base[:-2] + ext
        is_dll = ext.lower() != ".exe"
        if os.path.normcase(name) in descriptions:
          desc = descriptions[os.path.normcase(name)]
          try:
            verstamp.stamp(vars, pathname, desc, is_dll=is_dll)
            numStamped = numStamped + 1
          except win32api.error, exc:
            print "Could not stamp", pathname, "Error", exc.winerror, "-", exc.strerror
        else:
          print 'WARNING: description not provided for:', name
          # skip branding this - assume already branded or handled elsewhere
#        print "Stamped", pathname

def verify_request(self, sock, ca):
        # Do the sspi auth dance
        self.sa.reset()
        while 1:
            data = _get_msg(sock)
            if data is None:
                return False
            try:
                err, sec_buffer = self.sa.authorize(data)
            except sspi.error, details:
                print "FAILED to authorize client:", details
                return False

            if err==0:
                break
            _send_msg(sock, sec_buffer[0].Buffer)
        return True

def DumpRegistry(root, level=0):
    # A recursive dump of the remote registry to test most functions.
    h = wincerapi.CeRegOpenKeyEx(win32con.HKEY_LOCAL_MACHINE, None)
    level_prefix = " " * level
    index = 0
    # Enumerate values.
    while 1:
        try:
            name, data, typ = wincerapi.CeRegEnumValue(root, index)
        except win32api.error:
            break
        print "%s%s=%s" % (level_prefix, name, repr(str(data)))
        index = index+1
    # Now enumerate all keys.
    index=0
    while 1:
        try:
            name, klass = wincerapi.CeRegEnumKeyEx(root, index)
        except win32api.error:
            break
        print "%s%s\\" % (level_prefix, name)
        subkey = wincerapi.CeRegOpenKeyEx(root, name)
        DumpRegistry(subkey, level+1)
        index = index+1

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 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 _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 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)

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, 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)

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, 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")

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, e:
            self.assertEquals(e.winerror, win32file.WSAENOTSOCK)
        try:
            win32file.WSAEnumNetworkEvents(s, h)
        except win32file.error, 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 CheckLoaderModule(dll_name):
    suffix = ""
    if is_debug_build: suffix = "_d"
    template = os.path.join(this_dir,
                            "PyISAPI_loader" + suffix + ".dll")
    if not os.path.isfile(template):
        raise ConfigurationError(
              "Template loader '%s' does not exist" % (template,))
    # We can't do a simple "is newer" check, as the DLL is specific to the
    # Python version.  So we check the date-time and size are identical,
    # and skip the copy in that case.
    src_stat = os.stat(template)
    try:
        dest_stat = os.stat(dll_name)
    except os.error:
        same = 0
    else:
        same = src_stat[stat.ST_SIZE]==dest_stat[stat.ST_SIZE] and \
               src_stat[stat.ST_MTIME]==dest_stat[stat.ST_MTIME]
    if not same:
        log(2, "Updating %s->%s" % (template, dll_name))
        shutil.copyfile(template, dll_name)
        shutil.copystat(template, dll_name)
    else:
        log(2, "%s is up to date." % (dll_name,))

def RegisterPythonExe(exeFullPath, exeAlias = None, exeAppPath = None):
    """Register a .exe file that uses Python.

       Registers the .exe with the OS.  This allows the specified .exe to
       be run from the command-line or start button without using the full path,
       and also to setup application specific path (ie, os.environ['PATH']).

       Currently the exeAppPath is not supported, so this function is general
       purpose, and not specific to Python at all.  Later, exeAppPath may provide
       a reasonable default that is used.

       exeFullPath -- The full path to the .exe
       exeAlias = None -- An alias for the exe - if none, the base portion
                 of the filename is used.
       exeAppPath -- Not supported.
    """
    # Note - Dont work on win32s (but we dont care anymore!)
    if exeAppPath:
        raise error("Do not support exeAppPath argument currently")
    if exeAlias is None:
        exeAlias = os.path.basename(exeFullPath)
    win32api.RegSetValue(GetRootKey(), GetAppPathsKey() + "\\" + exeAlias, win32con.REG_SZ, exeFullPath)

def RegisterModule(modName, modPath):
    """Register an explicit module in the registry.  This forces the Python import
           mechanism to locate this module directly, without a sys.path search.  Thus
           a registered module need not appear in sys.path at all.

       modName -- The name of the module, as used by import.
       modPath -- The full path and file name of the module.
    """
    try:
        import os
        os.stat(modPath)
    except os.error:
        print "Warning: Registering non-existant module %s" % modPath
    win32api.RegSetValue(GetRootKey(), 
                         BuildDefaultPythonKey() + "\\Modules\\%s" % modName,
        win32con.REG_SZ, modPath)

def RegisterHelpFile(helpFile, helpPath, helpDesc = None, bCheckFile = 1):
    """Register a help file in the registry.

         Note that this used to support writing to the Windows Help
         key, however this is no longer done, as it seems to be incompatible.

           helpFile -- the base name of the help file.
           helpPath -- the path to the help file
           helpDesc -- A description for the help file.  If None, the helpFile param is used.
           bCheckFile -- A flag indicating if the file existence should be checked.
    """
    if helpDesc is None: helpDesc = helpFile
    fullHelpFile = os.path.join(helpPath, helpFile)
    try:
        if bCheckFile: os.stat(fullHelpFile)
    except os.error:
        raise ValueError("Help file does not exist")
    # Now register with Python itself.
    win32api.RegSetValue(GetRootKey(), 
                         BuildDefaultPythonKey() + "\\Help\\%s" % helpDesc, win32con.REG_SZ, fullHelpFile)

def RegisterCoreDLL(coredllName = None):
    """Registers the core DLL in the registry.

        If no params are passed, the name of the Python DLL used in 
        the current process is used and registered.
    """
    if coredllName is None:
        coredllName = win32api.GetModuleFileName(sys.dllhandle)
        # must exist!
    else:
        try:
            os.stat(coredllName)
        except os.error:
            print "Warning: Registering non-existant core DLL %s" % coredllName

    hKey = win32api.RegCreateKey(GetRootKey() , BuildDefaultPythonKey())
    try:
        win32api.RegSetValue(hKey, "Dll", win32con.REG_SZ, coredllName)
    finally:
        win32api.RegCloseKey(hKey)
    # Lastly, setup the current version to point to me.
    win32api.RegSetValue(GetRootKey(), "Software\\Python\\PythonCore\\CurrentVersion", win32con.REG_SZ, sys.winver)

def addcounter(self,object, counter, instance = None, inum=-1, machine=None):
        '''
        Adds a single counter path to the paths attribute.  Normally
        this will be called by a child class' speciality functions,
        rather than being called directly by the user. (Though it isn't
        hard to call manually, since almost everything is given a default)
        This method is only functional when the query is closed (or hasn't
        yet been opened).  This is to prevent conflict in multi-threaded
        query applications).
        e.g.:
            query.addcounter('Memory','Available Bytes')
        '''
        if not self.active:
            try:
                self.rawaddcounter(object, counter, instance, inum, machine)
                return 0
            except win32api.error:
                return -1
        else:
            return -1