def SimpleFileDemo():
testName = os.path.join( win32api.GetTempPath(), "win32file_demo_test_file")
if os.path.exists(testName): os.unlink(testName)
# Open the file for writing.
handle = win32file.CreateFile(testName,
win32file.GENERIC_WRITE,
0,
None,
win32con.CREATE_NEW,
0,
None)
test_data = "Hello\0there".encode("ascii")
win32file.WriteFile(handle, test_data)
handle.Close()
# Open it for reading.
handle = win32file.CreateFile(testName, win32file.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
rc, data = win32file.ReadFile(handle, 1024)
handle.Close()
if data == test_data:
print "Successfully wrote and read a file"
else:
raise Exception("Got different data back???")
os.unlink(testName)
python类CreateFile()的实例源码
def load(self, drive=''):
'''Closes cd drive door and waits until cd is readable.'''
drive = drive or self.drive
device = self.__getDeviceHandle(drive)
hdevice = win32file.CreateFile(device, GENERIC_READ,
FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0)
win32file.DeviceIoControl(hdevice,2967564,"", 0, None)
win32file.CloseHandle(hdevice)
# Poll drive for loaded and give up after timeout period
i=0
while i < 20:
if self.__is_cd_inserted(drive) == 1:
return 1
else:
time.sleep(1)
i = i+1
return 0
def CopyFileToCe(src_name, dest_name, progress = None):
sh = win32file.CreateFile(src_name, win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
bytes=0
try:
dh = wincerapi.CeCreateFile(dest_name, win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
try:
while 1:
hr, data = win32file.ReadFile(sh, 2048)
if not data:
break
wincerapi.CeWriteFile(dh, data)
bytes = bytes + len(data)
if progress is not None: progress(bytes)
finally:
pass
dh.Close()
finally:
sh.Close()
return bytes
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 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 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 CopyFileToCe(src_name, dest_name, progress = None):
sh = win32file.CreateFile(src_name, win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
bytes=0
try:
dh = wincerapi.CeCreateFile(dest_name, win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
try:
while 1:
hr, data = win32file.ReadFile(sh, 2048)
if not data:
break
wincerapi.CeWriteFile(dh, data)
bytes = bytes + len(data)
if progress is not None: progress(bytes)
finally:
pass
dh.Close()
finally:
sh.Close()
return bytes
def SimpleFileDemo():
testName = os.path.join( win32api.GetTempPath(), "win32file_demo_test_file")
if os.path.exists(testName): os.unlink(testName)
# Open the file for writing.
handle = win32file.CreateFile(testName,
win32file.GENERIC_WRITE,
0,
None,
win32con.CREATE_NEW,
0,
None)
test_data = "Hello\0there".encode("ascii")
win32file.WriteFile(handle, test_data)
handle.Close()
# Open it for reading.
handle = win32file.CreateFile(testName, win32file.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
rc, data = win32file.ReadFile(handle, 1024)
handle.Close()
if data == test_data:
print("Successfully wrote and read a file")
else:
raise Exception("Got different data back???")
os.unlink(testName)
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 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 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 _OpenFileForRead(self, path):
try:
fhandle = self.fhandle = win32file.CreateFile(
path,
win32file.GENERIC_READ,
win32file.FILE_SHARE_READ | win32file.FILE_SHARE_WRITE,
None,
win32file.OPEN_EXISTING,
win32file.FILE_ATTRIBUTE_NORMAL,
None)
self._closer = weakref.ref(
self, lambda x: win32file.CloseHandle(fhandle))
self.write_enabled = False
return fhandle
except pywintypes.error as e:
raise IOError("Unable to open %s: %s" % (path, e))
def _OpenFileForWrite(self, path):
try:
fhandle = self.fhandle = win32file.CreateFile(
path,
win32file.GENERIC_READ | win32file.GENERIC_WRITE,
win32file.FILE_SHARE_READ | win32file.FILE_SHARE_WRITE,
None,
win32file.OPEN_EXISTING,
win32file.FILE_ATTRIBUTE_NORMAL,
None)
self.write_enabled = True
self._closer = weakref.ref(
self, lambda x: win32file.CloseHandle(fhandle))
return fhandle
except pywintypes.error as e:
raise IOError("Unable to open %s: %s" % (path, e))
def changeTimestamp(path,fileName,daylimit):
print '[*] Inside changeTimestamp'
dates = {}
fileName = os.path.join(path, fileName)
dates['tdata'] = getDate(fileName,daylimit)
dates['ctime'] = datetime.utcfromtimestamp(os.path.getctime(fileName))
# print "[*] Original time: ",str(dates['ctime'])
if __use_win_32:
filehandle = win32file.CreateFile(fileName, win32file.GENERIC_WRITE, 0, None, win32con.OPEN_EXISTING, 0, None)
win32file.SetFileTime(filehandle, dates['tdata'],dates['tdata'],dates['tdata'])
filehandle.close()
print "[*] Timestamps changed!!"
else:
os.utime(fileName, (time.mktime(dates['tdata'].utctimetuple()),)*2)
dates['mtime'] = datetime.utcfromtimestamp(os.path.getmtime(fileName))
# print "[*] Modified time: ",str(dates['mtime'])
def __init__(self):
self.pty = Pty()
self.ready_f = Future()
self._input_ready_callbacks = []
self.loop = get_event_loop()
self.stdout_handle = win32file.CreateFile(
self.pty.conout_name(),
win32con.GENERIC_READ,
0,
win32security.SECURITY_ATTRIBUTES(),
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
0)
self.stdin_handle = win32file.CreateFile(
self.pty.conin_name(),
win32con.GENERIC_WRITE,
0,
win32security.SECURITY_ATTRIBUTES(),
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_OVERLAPPED,
0)
self._buffer = []
def pipe(bufsize=8192):
"""Creates overlapped (asynchronous) pipe.
"""
name = r'\\.\pipe\pycos-pipe-%d-%d' % (os.getpid(), next(_pipe_id))
openmode = (win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED |
FILE_FLAG_FIRST_PIPE_INSTANCE)
pipemode = (win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE)
rh = wh = None
try:
rh = win32pipe.CreateNamedPipe(
name, openmode, pipemode, 1, bufsize, bufsize,
win32pipe.NMPWAIT_USE_DEFAULT_WAIT, None)
wh = win32file.CreateFile(
name, win32file.GENERIC_WRITE | winnt.FILE_READ_ATTRIBUTES, 0, None,
win32file.OPEN_EXISTING, win32file.FILE_FLAG_OVERLAPPED, None)
overlapped = pywintypes.OVERLAPPED()
# 'yield' can't be used in constructor so use sync wait
# (in this case it is should be okay)
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
rc = win32pipe.ConnectNamedPipe(rh, overlapped)
if rc == winerror.ERROR_PIPE_CONNECTED:
win32event.SetEvent(overlapped.hEvent)
rc = win32event.WaitForSingleObject(overlapped.hEvent, 1000)
overlapped = None
if rc != win32event.WAIT_OBJECT_0:
pycos.logger.warning('connect failed: %s' % rc)
raise Exception(rc)
return (rh, wh)
except:
if rh is not None:
win32file.CloseHandle(rh)
if wh is not None:
win32file.CloseHandle(wh)
raise
def pipe(bufsize=8192):
"""Creates overlapped (asynchronous) pipe.
"""
name = r'\\.\pipe\pycos-pipe-%d-%d' % (os.getpid(), next(_pipe_id))
openmode = (win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED |
FILE_FLAG_FIRST_PIPE_INSTANCE)
pipemode = (win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE)
rh = wh = None
try:
rh = win32pipe.CreateNamedPipe(
name, openmode, pipemode, 1, bufsize, bufsize,
win32pipe.NMPWAIT_USE_DEFAULT_WAIT, None)
wh = win32file.CreateFile(
name, win32file.GENERIC_WRITE | winnt.FILE_READ_ATTRIBUTES, 0, None,
win32file.OPEN_EXISTING, win32file.FILE_FLAG_OVERLAPPED, None)
overlapped = pywintypes.OVERLAPPED()
# 'yield' can't be used in constructor so use sync wait
# (in this case it is should be okay)
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
rc = win32pipe.ConnectNamedPipe(rh, overlapped)
if rc == winerror.ERROR_PIPE_CONNECTED:
win32event.SetEvent(overlapped.hEvent)
rc = win32event.WaitForSingleObject(overlapped.hEvent, 1000)
overlapped = None
if rc != win32event.WAIT_OBJECT_0:
pycos.logger.warning('connect failed: %s' % rc)
raise Exception(rc)
return (rh, wh)
except:
if rh is not None:
win32file.CloseHandle(rh)
if wh is not None:
win32file.CloseHandle(wh)
raise
def opencom1():
# returns a handle to the Windows file
return win32file.CreateFile(u'COM1:', win32file.GENERIC_READ, \
0, None, win32file.OPEN_EXISTING, 0, None)
def eject(self, drive=''):
'''Opens the cd drive door.'''
drive = drive or self.drive
device = self.__getDeviceHandle(drive)
hdevice = win32file.CreateFile(device, GENERIC_READ,
FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0)
win32file.DeviceIoControl(hdevice,2967560,"", 0, None)
win32file.CloseHandle(hdevice)
def close(self, drive=''):
'''Closes the cd drive door.'''
drive = drive or self.drive
device = self.__getDeviceHandle(drive)
hdevice = win32file.CreateFile(device, GENERIC_READ,
FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0)
win32file.DeviceIoControl(hdevice,2967564,"", 0, None)
win32file.CloseHandle(hdevice)
def watchos():
#get path or maintain current path of app
FILE_LIST_DIRECTORY = 0x0001
try: path_to_watch = myos.get() or "."
except: path_to_watch = "."
path_to_watch = os.path.abspath(path_to_watch)
textbox.insert(END, "Watching %s at %s" % (path_to_watch, time.asctime()) + "\n\n")
# FindFirstChangeNotification sets up a handle for watching
# file changes.
while 1:
hDir = win32file.CreateFile (
path_to_watch,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
change_handle = win32file.ReadDirectoryChangesW (
hDir,
1024,
True,#Heap Size include_subdirectories,
win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
win32con.FILE_NOTIFY_CHANGE_SIZE |
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
win32con.FILE_NOTIFY_CHANGE_SECURITY,
None,
None
)
# Loop forever, listing any file changes. The WaitFor... will
# time out every half a second allowing for keyboard interrupts
# to terminate the loop.
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed from something",
5 : "Renamed to something"
}
results = change_handle
for action, files in results:
full_filename = os.path.join(path_to_watch, files)
theact = ACTIONS.get(action, "Unknown")
textbox.insert(END, str(full_filename) + "\t" + str(theact) +"\n")
def run(self):
if running_on_linux()==True:
print("thread: start")
wm = pyinotify.WatchManager()
print("wathcing path",self.watch_path)
ret=wm.add_watch(self.watch_path, pyinotify.IN_CLOSE_WRITE, self.onChange,False,False)
print(ret)
print("thread: start notifyer",self.notifier)
self.notifier = pyinotify.Notifier(wm)
try:
while 1:
self.notifier.process_events()
if self.notifier.check_events():
self.notifier.read_events()
#self.notifier.loop()
except:
print("error in notify",sys.exc_info()[0])
else:
hDir = win32file.CreateFile (self.watch_path,FILE_LIST_DIRECTORY,win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE | win32con.FILE_SHARE_DELETE,None,win32con.OPEN_EXISTING,win32con.FILE_FLAG_BACKUP_SEMANTICS,None)
while 1:
results = win32file.ReadDirectoryChangesW (hDir,1024,True,
win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
win32con.FILE_NOTIFY_CHANGE_SIZE |
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
win32con.FILE_NOTIFY_CHANGE_SECURITY,
None,
None)
for action, file in results:
full_filename = os.path.join (self.watch_path, file)
self.onChange(full_filename)
def connect(self, address):
win32pipe.WaitNamedPipe(address, self._timeout)
try:
handle = win32file.CreateFile(
address,
win32file.GENERIC_READ | win32file.GENERIC_WRITE,
0,
None,
win32file.OPEN_EXISTING,
cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,
0
)
except win32pipe.error as e:
# See Remarks:
# https://msdn.microsoft.com/en-us/library/aa365800.aspx
if e.winerror == cERROR_PIPE_BUSY:
# Another program or thread has grabbed our pipe instance
# before we got to it. Wait for availability and attempt to
# connect again.
win32pipe.WaitNamedPipe(address, RETRY_WAIT_TIMEOUT)
return self.connect(address)
raise e
self.flags = win32pipe.GetNamedPipeInfo(handle)[0]
self._handle = handle
self._address = address
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_utc = win32timezone.utcnow()
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 HttpExtensionProc(self, ecb):
# NOTE: If you use a ThreadPoolExtension, you must still perform
# this check in HttpExtensionProc - raising the exception from
# The "Dispatch" method will just cause the exception to be
# rendered to the browser.
if self.reload_watcher.change_detected:
print "Doing reload"
raise InternalReloadException
if ecb.GetServerVariable("UNICODE_URL").endswith("test.py"):
file_flags = win32con.FILE_FLAG_SEQUENTIAL_SCAN | win32con.FILE_FLAG_OVERLAPPED
hfile = win32file.CreateFile(__file__, win32con.GENERIC_READ,
0, None, win32con.OPEN_EXISTING,
file_flags, None)
flags = isapicon.HSE_IO_ASYNC | isapicon.HSE_IO_DISCONNECT_AFTER_SEND | \
isapicon.HSE_IO_SEND_HEADERS
# We pass hFile to the callback simply as a way of keeping it alive
# for the duration of the transmission
try:
ecb.TransmitFile(TransmitFileCallback, hfile,
int(hfile),
"200 OK",
0, 0, None, None, flags)
except:
# Errors keep this source file open!
hfile.Close()
raise
else:
# default response
ecb.SendResponseHeaders("200 OK", "Content-Type: text/html\r\n\r\n", 0)
print >> ecb, "<HTML><BODY>"
print >> ecb, "The root of this site is at", ecb.MapURLToPath("/")
print >> ecb, "</BODY></HTML>"
ecb.close()
return isapicon.HSE_STATUS_SUCCESS
def HttpExtensionProc(self, ecb):
# NOTE: If you use a ThreadPoolExtension, you must still perform
# this check in HttpExtensionProc - raising the exception from
# The "Dispatch" method will just cause the exception to be
# rendered to the browser.
if self.reload_watcher.change_detected:
print("Doing reload")
raise InternalReloadException
if ecb.GetServerVariable("UNICODE_URL").endswith("test.py"):
file_flags = win32con.FILE_FLAG_SEQUENTIAL_SCAN | win32con.FILE_FLAG_OVERLAPPED
hfile = win32file.CreateFile(__file__, win32con.GENERIC_READ,
0, None, win32con.OPEN_EXISTING,
file_flags, None)
flags = isapicon.HSE_IO_ASYNC | isapicon.HSE_IO_DISCONNECT_AFTER_SEND | \
isapicon.HSE_IO_SEND_HEADERS
# We pass hFile to the callback simply as a way of keeping it alive
# for the duration of the transmission
try:
ecb.TransmitFile(TransmitFileCallback, hfile,
int(hfile),
"200 OK",
0, 0, None, None, flags)
except:
# Errors keep this source file open!
hfile.Close()
raise
else:
# default response
ecb.SendResponseHeaders("200 OK", "Content-Type: text/html\r\n\r\n", 0)
print("<HTML><BODY>", file=ecb)
print("The root of this site is at", ecb.MapURLToPath("/"), file=ecb)
print("</BODY></HTML>", file=ecb)
ecb.close()
return isapicon.HSE_STATUS_SUCCESS
def TestDeviceNotifications(dir_names):
wc = win32gui.WNDCLASS()
wc.lpszClassName = 'test_devicenotify'
wc.style = win32con.CS_GLOBALCLASS|win32con.CS_VREDRAW | win32con.CS_HREDRAW
wc.hbrBackground = win32con.COLOR_WINDOW+1
wc.lpfnWndProc={win32con.WM_DEVICECHANGE:OnDeviceChange}
class_atom=win32gui.RegisterClass(wc)
hwnd = win32gui.CreateWindow(wc.lpszClassName,
'Testing some devices',
# no need for it to be visible.
win32con.WS_CAPTION,
100,100,900,900, 0, 0, 0, None)
hdevs = []
# Watch for all USB device notifications
filter = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE(
GUID_DEVINTERFACE_USB_DEVICE)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# and create handles for all specified directories
for d in dir_names:
hdir = win32file.CreateFile(d,
winnt.FILE_LIST_DIRECTORY,
winnt.FILE_SHARE_READ | winnt.FILE_SHARE_WRITE | winnt.FILE_SHARE_DELETE,
None, # security attributes
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS | # required privileges: SE_BACKUP_NAME and SE_RESTORE_NAME.
win32con.FILE_FLAG_OVERLAPPED,
None)
filter = win32gui_struct.PackDEV_BROADCAST_HANDLE(hdir)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# now start a message pump and wait for messages to be delivered.
print("Watching", len(hdevs), "handles - press Ctrl+C to terminate, or")
print("add and remove some USB devices...")
if not dir_names:
print("(Note you can also pass paths to watch on the command-line - eg,")
print("pass the root of an inserted USB stick to see events specific to")
print("that volume)")
while 1:
win32gui.PumpWaitingMessages()
time.sleep(0.01)
win32gui.DestroyWindow(hwnd)
win32gui.UnregisterClass(wc.lpszClassName, None)
