def test_walk(self):
     """
     Verify that walking the path gives the same result as the known file
     hierarchy.
     """
     x = [foo.path for foo in self.path.walk()]
     self.assertEquals(set(x), set(self.all))

    def __init__(self, glib_module, gtk_module, useGtk=False):
        self._simtag = None
        self._reads = set()
        self._writes = set()
        self._sources = {}
        self._glib = glib_module
        self._gtk = gtk_module
        posixbase.PosixReactorBase.__init__(self)

        self._source_remove = self._glib.source_remove
        self._timeout_add = self._glib.timeout_add

        def _mainquit():
            if self._gtk.main_level():
                self._gtk.main_quit()

        if useGtk:
            self._pending = self._gtk.events_pending
            self._iteration = self._gtk.main_iteration_do
            self._crash = _mainquit
            self._run = self._gtk.main
        else:
            self.context = self._glib.main_context_default()
            self._pending = self.context.pending
            self._iteration = self.context.iteration
            self.loop = self._glib.MainLoop()
            self._crash = lambda: self._glib.idle_add(self.loop.quit)
            self._run = self.loop.run

 def test_iteration(self):
     """
     L{_DictHeaders.__iter__} returns an iterator the elements of which
     are the lowercase name of each header present.
     """
     headers, wrapper = self.headers(foo=["lemur", "panda"], bar=["baz"])
     self.assertEqual(set(list(wrapper)), set(["foo", "bar"]))

 def test_keys(self, _method='keys', _requireList=True):
     """
     L{_DictHeaders.keys} will return a list of all present header names.
     """
     headers, wrapper = self.headers(test=["lemur"], foo=["bar"])
     keys = getattr(wrapper, _method)()
     if _requireList:
         self.assertIsInstance(keys, list)
     self.assertEqual(set(keys), set(["foo", "test"]))

 def test_values(self, _method='values', _requireList=True):
     """
     L{_DictHeaders.values} will return a list of all present header values,
     returning only the last value for headers with more than one.
     """
     headers, wrapper = self.headers(foo=["lemur"], bar=["marmot", "panda"])
     values = getattr(wrapper, _method)()
     if _requireList:
         self.assertIsInstance(values, list)
     self.assertEqual(set(values), set(["lemur", "panda"]))

 def test_walkObeysDescend(self):
     """
     Verify that when the supplied C{descend} predicate returns C{False},
     the target is not traversed.
     """
     self.createLinks()
     def noSymLinks(path):
         return not path.islink()
     x = [foo.path for foo in self.path.walk(descend=noSymLinks)]
     self.assertEquals(set(x), set(self.all))

 def test_items(self, _method='items', _requireList=True):
     """
     L{_DictHeaders.items} will return a list of all present header names
     and values as tuples, returning only the last value for headers with
     more than one.
     """
     headers, wrapper = self.headers(foo=["lemur"], bar=["marmot", "panda"])
     items = getattr(wrapper, _method)()
     if _requireList:
         self.assertIsInstance(items, list)
     self.assertEqual(set(items), set([("foo", "lemur"), ("bar", "panda")]))

    def test_getDelayedCalls(self):
        """
        Test that we can get a list of all delayed calls
        """
        c = task.Clock()
        call = c.callLater(1, lambda x: None)
        call2 = c.callLater(2, lambda x: None)

        calls = c.getDelayedCalls()

        self.assertEquals(set([call, call2]), set(calls))

 def test_getPrivateKeys(self):
     """
     L{OpenSSHFactory.getPrivateKeys} should return the available private
     keys in the data directory.
     """
     keys = self.factory.getPrivateKeys()
     self.assertEquals(len(keys), 2)
     keyTypes = keys.keys()
     self.assertEqual(set(keyTypes), set(['ssh-rsa', 'ssh-dss']))
     self.assertEquals(self.mockos.seteuidCalls, [])
     self.assertEquals(self.mockos.setegidCalls, [])

 def test_checkersWithoutPamAuth(self):
     """
     The L{OpenSSHFactory} built by L{tap.makeService} has a portal with
     L{ISSHPrivateKey} and L{IUsernamePassword} interfaces registered as
     checkers if C{pamauth} is not available.
     """
     # Fake the absence of pamauth, even if PyPAM is installed
     self.patch(tap, "pamauth", None)
     config = tap.Options()
     service = tap.makeService(config)
     portal = service.factory.portal
     self.assertEquals(set(portal.checkers.keys()), set([ISSHPrivateKey, IUsernamePassword]))

 def test_unzip(self):
     """
     L{twisted.python.zipstream.unzip} should extract all files from a zip
     archive
     """
     numfiles = 3
     zpfilename = self.makeZipFile([str(i) for i in range(numfiles)])
     zipstream.unzip(zpfilename, self.unzipdir.path)
     self.assertEqual(
         set(self.unzipdir.listdir()),
         set(map(str, range(numfiles))))
     for i in range(numfiles):
         self.assertEqual(self.unzipdir.child(str(i)).getContent(), str(i))

    def __init__(self, useGtk=True):
        self._simtag = None
        self._reads = set()
        self._writes = set()
        self._sources = {}
        posixbase.PosixReactorBase.__init__(self)

        self.context = gobject.main_context_default()
        self.__pending = self.context.pending
        self.__iteration = self.context.iteration
        self.loop = gobject.MainLoop()
        self.__crash = self.loop.quit
        self.__run = self.loop.run

    def test_getAllRawHeaders(self):
        """
        L{Headers.getAllRawHeaders} returns an iterable of (k, v) pairs, where
        C{k} is the canonicalized representation of the header name, and C{v}
        is a sequence of values.
        """
        h = Headers()
        h.setRawHeaders(b"test", [b"lemurs"])
        h.setRawHeaders(b"www-authenticate", [b"basic aksljdlk="])

        allHeaders = set([(k, tuple(v)) for k, v in h.getAllRawHeaders()])

        self.assertEqual(allHeaders, set([(b"WWW-Authenticate", (b"basic aksljdlk=",)), (b"Test", (b"lemurs",))]))

 def test_removeAllReturnsRemovedDescriptors(self):
     """
     L{PosixReactorBase._removeAll} returns a list of removed
     L{IReadDescriptor} and L{IWriteDescriptor} objects.
     """
     reactor = TrivialReactor()
     reader = object()
     writer = object()
     reactor.addReader(reader)
     reactor.addWriter(writer)
     removed = reactor._removeAll(reactor._readers, reactor._writers)
     self.assertEqual(set(removed), set([reader, writer]))
     self.assertNotIn(reader, reactor._readers)
     self.assertNotIn(writer, reactor._writers)

    def test_unzipIterChunky(self):
        """
        L{twisted.python.zipstream.unzipIterChunky} returns an iterator which
        must be exhausted to completely unzip the input archive.
        """
        numfiles = 10
        contents = ["This is test file %d!" % i for i in range(numfiles)]
        zpfilename = self.makeZipFile(contents)
        list(zipstream.unzipIterChunky(zpfilename, self.unzipdir.path))
        self.assertEqual(set(self.unzipdir.listdir()), set(map(str, range(numfiles))))

        for child in self.unzipdir.children():
            num = int(child.basename())
            self.assertEqual(child.getContent(), contents[num])

 def test_unzipDirectory(self):
     """
     The path to which a file is extracted by L{zipstream.unzip} is
     determined by joining the C{directory} argument to C{unzip} with the
     path within the archive of the file being extracted.
     """
     numfiles = 3
     zpfilename = self.makeZipFile([str(i) for i in range(numfiles)], 'foo')
     zipstream.unzip(zpfilename, self.unzipdir.path)
     self.assertEqual(
         set(self.unzipdir.child('foo').listdir()),
         set(map(str, range(numfiles))))
     for i in range(numfiles):
         self.assertEqual(
             self.unzipdir.child('foo').child(str(i)).getContent(), str(i))

    def test_unzipIterChunkyDirectory(self):
        """
        The path to which a file is extracted by L{zipstream.unzipIterChunky}
        is determined by joining the C{directory} argument to C{unzip} with the
        path within the archive of the file being extracted.
        """
        numfiles = 10
        contents = ["This is test file %d!" % i for i in range(numfiles)]
        zpfilename = self.makeZipFile(contents, "foo")
        list(zipstream.unzipIterChunky(zpfilename, self.unzipdir.path))
        self.assertEqual(set(self.unzipdir.child("foo").listdir()), set(map(str, range(numfiles))))

        for child in self.unzipdir.child("foo").children():
            num = int(child.basename())
            self.assertEqual(child.getContent(), contents[num])

 def test_checkersPamAuth(self):
     """
     The L{OpenSSHFactory} built by L{tap.makeService} has a portal with
     L{IPluggableAuthenticationModules}, L{ISSHPrivateKey} and
     L{IUsernamePassword} interfaces registered as checkers if C{pamauth} is
     available.
     """
     # Fake the presence of pamauth, even if PyPAM is not installed
     self.patch(tap, "pamauth", object())
     config = tap.Options()
     service = tap.makeService(config)
     portal = service.factory.portal
     self.assertEqual(
         set(portal.checkers.keys()),
         set([IPluggableAuthenticationModules, ISSHPrivateKey,
              IUsernamePassword]))

def spawnProcess(processProtocol, bootstrap, args=(), env={},
                 path=None, uid=None, gid=None, usePTY=0,
                 packages=()):
    env = env.copy()

    pythonpath = []
    for pkg in packages:
        pkg_path, name = os.path.split(pkg)
        p = os.path.split(imp.find_module(name, [pkg_path] if pkg_path else None)[1])[0]
        if p.startswith(os.path.join(sys.prefix, 'lib')):
            continue
        pythonpath.append(p)
    pythonpath = list(set(pythonpath))
    pythonpath.extend(env.get('PYTHONPATH', '').split(os.pathsep))
    env['PYTHONPATH'] = os.pathsep.join(pythonpath)
    args = (sys.executable, '-c', bootstrap) + args
    # childFDs variable is needed because sometimes child processes
    # misbehave and use stdout to output stuff that should really go
    # to stderr. Of course child process might even use the wrong FDs
    # that I'm using here, 3 and 4, so we are going to fix all these
    # issues when I add support for the configuration object that can
    # fix this stuff in a more configurable way.
    if IS_WINDOWS:
        return reactor.spawnProcess(processProtocol, sys.executable, args,
                                    env, path, uid, gid, usePTY)
    else:
        return reactor.spawnProcess(processProtocol, sys.executable, args,
                                    env, path, uid, gid, usePTY,
                                    childFDs={0:"w", 1:"r", 2:"r", 3:"w", 4:"r"})

    def __init__(self):
        self.threadCallQueue = []
        self._eventTriggers = {}
        self._pendingTimedCalls = []
        self._newTimedCalls = []
        self._cancellations = 0
        self.running = False
        self._started = False
        self._justStopped = False
        self._startedBefore = False
        # reactor internal readers, e.g. the waker.
        self._internalReaders = set()
        self.waker = None

        # Arrange for the running attribute to change to True at the right time
        # and let a subclass possibly do other things at that time (eg install
        # signal handlers).
        self.addSystemEventTrigger(
            'during', 'startup', self._reallyStartRunning)
        self.addSystemEventTrigger('during', 'shutdown', self.crash)
        self.addSystemEventTrigger('during', 'shutdown', self.disconnectAll)

        if platform.supportsThreads():
            self._initThreads()
        self.installWaker()