[pywin32-checkins] pywin32/com/win32comext/bits/test show_all_jobs.py, 1.1, 1.2 test_bits.py, 1.1, 1.2

[Sidnei da S. <dre...@us...>]

Update of /cvsroot/pywin32/pywin32/com/win32comext/bits/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv3887/com/win32comext/bits/test

Added Files:
	show_all_jobs.py test_bits.py 
Log Message:
 - Merge sidnei-bits branch, adds support for BITS (Background Intelligent Transfer Service)

--- NEW FILE: show_all_jobs.py ---
# Dump lots of info about BITS jobs.
from win32com.bits import bits
import pythoncom

states = dict([(val, (name[13:]))
               for name, val in vars(bits).items()
               if name.startswith('BG_JOB_STATE_')])

job_types = dict([(val, (name[12:]))
               for name, val in vars(bits).items()
               if name.startswith('BG_JOB_TYPE_')])

bcm = pythoncom.CoCreateInstance(bits.CLSID_BackgroundCopyManager, 
                                 None,
                                 pythoncom.CLSCTX_LOCAL_SERVER,
                                 bits.IID_IBackgroundCopyManager)

try:
    enum = bcm.EnumJobs(bits.BG_JOB_ENUM_ALL_USERS)
except pythoncom.error:
    print "Failed to get jobs for all users - trying for current user"
    enum = bcm.EnumJobs(0)

for job in enum:
    print "Job:", job.GetDisplayName()
    print "Description:", job.GetDescription()
    print "Id:", job.GetId()
    print "State:", states.get(job.GetState())
    print "Type:", job_types.get(job.GetType())
    print "Owner:", job.GetOwner()
    print "Errors:", job.GetErrorCount()
    print "Created/Modified/Finished times:", [str(t) for t in job.GetTimes()]
    bytes_tot, bytes_xf, files_tot, files_xf = job.GetProgress()
    print "Bytes: %d complete of %d total" % (bytes_xf, bytes_tot)
    print "Files: %d complete of %d total" % (files_xf, files_tot)
    for f in job.EnumFiles():
        bytes, total, done = f.GetProgress()
        print "  Remote:", f.GetRemoteName()
        print "  Local:", f.GetLocalName()
        print "  Progress: %d of %d bytes - completed=%s)" % (bytes, total, done)
        print
    print

--- NEW FILE: test_bits.py ---


from win32com.server.util import wrap
import pythoncom, sys, os, time, win32api, win32event, tempfile
from win32com.bits import bits

TIMEOUT = 200 # ms
StopEvent = win32event.CreateEvent(None, 0, 0, None)

job_name = 'bits-pywin32-test'
states = dict([(val, (name[13:]))
               for name, val in vars(bits).items()
               if name.startswith('BG_JOB_STATE_')])

bcm = pythoncom.CoCreateInstance(bits.CLSID_BackgroundCopyManager, 
                                 None,
                                 pythoncom.CLSCTX_LOCAL_SERVER,
                                 bits.IID_IBackgroundCopyManager)

class BackgroundJobCallback:
    _com_interfaces_ = [bits.IID_IBackgroundCopyCallback]
    _public_methods_ = ["JobTransferred", "JobError", "JobModification"]
    
    def JobTransferred(self, job):
        print 'Job Transferred', job
        job.Complete()
        win32event.SetEvent(StopEvent) # exit msg pump

    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 _print_error(self, err):
        ctx, hresult = err.GetError()
        try:
            hresult_msg = win32api.FormatMessage(hresult)
        except win32api.error:
            hresult_msg  = ""
        print "Context=0x%x, hresult=0x%x (%s)" % (ctx, hresult, hresult_msg)
        print err.GetErrorDescription()

    def JobModification(self, job, reserved):
        state = job.GetState()
        print 'Job Modification', job.GetDisplayName(), states.get(state)
        # Need to catch TRANSIENT_ERROR here, as JobError doesn't get
        # called (apparently) when the error is transient.
        if state == bits.BG_JOB_STATE_TRANSIENT_ERROR:
            print "Error details:"
            err = job.GetError()
            self._print_error(err)

job = bcm.CreateJob(job_name, bits.BG_JOB_TYPE_DOWNLOAD)

job.SetNotifyInterface(wrap(BackgroundJobCallback()))
job.SetNotifyFlags(bits.BG_NOTIFY_JOB_TRANSFERRED |
                   bits.BG_NOTIFY_JOB_ERROR |
                   bits.BG_NOTIFY_JOB_MODIFICATION)


# The idea here is to intentionally make one of the files fail to be
# downloaded. Then the JobError notification will be triggered, where
# we do fix the failing file by calling SetRemoteName to a valid URL
# and call Resume() on the job, making the job finish successfully.
#
# Note to self: A domain that cannot be resolved will cause
# TRANSIENT_ERROR instead of ERROR, and the JobError notification will
# not be triggered! This can bite you during testing depending on how
# your DNS is configured. For example, if you use OpenDNS.org's DNS
# servers, an invalid hostname will *always* be resolved (they
# redirect you to a search page), so be careful when testing.
job.AddFile('http://www.python.org/favicon.ico', os.path.join(tempfile.gettempdir(), 'bits-favicon.ico'))
job.AddFile('http://www.python.org/missing-favicon.ico', os.path.join(tempfile.gettempdir(), 'bits-missing-favicon.ico'))

for f in job.EnumFiles():
    print 'Downloading', f.GetRemoteName()
    print 'To', f.GetLocalName()

job.Resume()

while True:
    rc = win32event.MsgWaitForMultipleObjects(
        (StopEvent,), 
        0,
        TIMEOUT,
        win32event.QS_ALLEVENTS)

    if rc == win32event.WAIT_OBJECT_0:
        break
    elif rc == win32event.WAIT_OBJECT_0+1:
        if pythoncom.PumpWaitingMessages():
            break # wm_quit