OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 __init__.py	103 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📁 __pycache__	-	03/31/2022 06:22:39 AM	rwxr-xr-x
📄 cert.pem.no_trailing_newline	1.38 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 crash_test_dummy.py	543 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 iosim.py	17.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 key.pem.no_trailing_newline	1.67 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 mock_win32process.py	1.46 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 myrebuilder1.py	158 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 myrebuilder2.py	158 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 plugin_basic.py	943 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 plugin_extra1.py	407 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 plugin_extra2.py	579 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_cmdline.py	162 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_echoer.py	214 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_fds.py	945 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 process_getargv.py	283 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 process_getenv.py	268 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 process_linger.py	286 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_reader.py	188 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_signal.py	214 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_stdinreader.py	857 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 process_tester.py	1.01 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_tty.py	130 bytes	09/08/2017 10:38:35 AM	rw-r--r--
📄 process_twisted.py	1.18 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 proto_helpers.py	26.33 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 raiser.c	93.05 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 raiser.cpython-36m-x86_64-linux-gnu.so	19.16 KB	03/22/2022 11:03:56 AM	rw-r--r--
📄 raiser.pyx	466 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 reflect_helper_IE.py	61 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 reflect_helper_VE.py	82 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 reflect_helper_ZDE.py	47 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 server.pem	4.34 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 ssl_helpers.py	1.01 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_consumer.py	1.19 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_halfclose.py	1.89 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_hostpeer.py	1021 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_lastwrite.py	1.18 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_loseconn.py	1.51 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_producer.py	1.47 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_write.py	923 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 stdio_test_writeseq.py	915 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_abstract.py	3.42 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_adbapi.py	25.53 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_amp.py	107.96 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_application.py	32.05 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_compat.py	27.32 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_context.py	1.48 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_cooperator.py	20.96 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_defer.py	100.93 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_defgen.py	10.45 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_dict.py	1.41 KB	09/23/2017 05:52:22 AM	rw-r--r--
📄 test_dirdbm.py	6.76 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_error.py	8.39 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_factories.py	4.53 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_failure.py	29.92 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_fdesc.py	7.2 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_finger.py	1.95 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_formmethod.py	3.56 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_ftp.py	127.27 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_ftp_options.py	2.62 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_htb.py	3.12 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_ident.py	6.85 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_internet.py	45.33 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_iosim.py	8.49 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_iutils.py	13.13 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_lockfile.py	15.14 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_log.py	35.48 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_logfile.py	17.8 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_loopback.py	14.15 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_main.py	2.44 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_memcache.py	24.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_modules.py	17.47 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_monkey.py	5.5 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_nooldstyle.py	5.82 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_paths.py	72.61 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_pcp.py	12.26 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_persisted.py	14.28 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_plugin.py	25.5 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_policies.py	32.04 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_postfix.py	3.53 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_process.py	84.1 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_protocols.py	7.28 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_randbytes.py	3.28 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_rebuild.py	8.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_reflect.py	25.47 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_roots.py	1.77 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_shortcut.py	1.89 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_sip.py	24.69 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_sob.py	5.5 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_socks.py	17.32 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_ssl.py	23.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_sslverify.py	104.28 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stateful.py	1.97 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stdio.py	12.85 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_strerror.py	5.06 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stringtransport.py	12.95 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_strports.py	1.75 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_task.py	38.4 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tcp.py	64.07 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tcp_internals.py	8.54 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_text.py	6.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threadable.py	3.65 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threadpool.py	22.47 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threads.py	12.96 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tpfile.py	1.56 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_twistd.py	61.05 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_twisted.py	18.42 KB	09/09/2017 05:11:54 AM	rw-r--r--
📄 test_udp.py	24.1 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_unix.py	14.8 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_usage.py	23.09 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 testutils.py	5.19 KB	09/08/2017 10:38:36 AM	rw-r--r--

Editing: test_iutils.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Test running processes with the APIs in L{twisted.internet.utils}.
"""

from __future__ import division, absolute_import

import warnings, os, stat, sys, signal

from twisted.python.compat import _PY3
from twisted.python.runtime import platform
from twisted.trial import unittest
from twisted.internet import error, reactor, utils, interfaces
from twisted.internet.defer import Deferred
from twisted.python.test.test_util import SuppressedWarningsTests

class ProcessUtilsTests(unittest.TestCase):
    """
    Test running a process using L{getProcessOutput}, L{getProcessValue}, and
    L{getProcessOutputAndValue}.
    """

if interfaces.IReactorProcess(reactor, None) is None:
        skip = "reactor doesn't implement IReactorProcess"

output = None
    value = None
    exe = sys.executable

def makeSourceFile(self, sourceLines):
        """
        Write the given list of lines to a text file and return the absolute
        path to it.
        """
        script = self.mktemp()
        with open(script, 'wt') as scriptFile:
            scriptFile.write(os.linesep.join(sourceLines) + os.linesep)
        return os.path.abspath(script)

def test_output(self):
        """
        L{getProcessOutput} returns a L{Deferred} which fires with the complete
        output of the process it runs after that process exits.
        """
        scriptFile = self.makeSourceFile([
                "import sys",
                "for s in b'hello world\\n':",
                "    if hasattr(sys.stdout, 'buffer'):",
                "        # Python 3",
                "        s = bytes([s])",
                "        sys.stdout.buffer.write(s)",
                "    else:",
                "        # Python 2",
                "        sys.stdout.write(s)",
                "    sys.stdout.flush()"])
        d = utils.getProcessOutput(self.exe, ['-u', scriptFile])
        return d.addCallback(self.assertEqual, b"hello world\n")

def test_outputWithErrorIgnored(self):
        """
        The L{Deferred} returned by L{getProcessOutput} is fired with an
        L{IOError} L{Failure} if the child process writes to stderr.
        """
        # make sure stderr raises an error normally
        scriptFile = self.makeSourceFile([
            'import sys',
            'sys.stderr.write("hello world\\n")'
            ])

d = utils.getProcessOutput(self.exe, ['-u', scriptFile])
        d = self.assertFailure(d, IOError)
        def cbFailed(err):
            return self.assertFailure(err.processEnded, error.ProcessDone)
        d.addCallback(cbFailed)
        return d

def test_outputWithErrorCollected(self):
        """
        If a C{True} value is supplied for the C{errortoo} parameter to
        L{getProcessOutput}, the returned L{Deferred} fires with the child's
        stderr output as well as its stdout output.
        """
        scriptFile = self.makeSourceFile([
            'import sys',
            # Write the same value to both because ordering isn't guaranteed so
            # this simplifies the test.
            'sys.stdout.write("foo")',
            'sys.stdout.flush()',
            'sys.stderr.write("foo")',
            'sys.stderr.flush()'])

d = utils.getProcessOutput(self.exe, ['-u', scriptFile], errortoo=True)
        return d.addCallback(self.assertEqual, b"foofoo")

def test_value(self):
        """
        The L{Deferred} returned by L{getProcessValue} is fired with the exit
        status of the child process.
        """
        scriptFile = self.makeSourceFile(["raise SystemExit(1)"])

d = utils.getProcessValue(self.exe, ['-u', scriptFile])
        return d.addCallback(self.assertEqual, 1)

def test_outputAndValue(self):
        """
        The L{Deferred} returned by L{getProcessOutputAndValue} fires with a
        three-tuple, the elements of which give the data written to the child's
        stdout, the data written to the child's stderr, and the exit status of
        the child.
        """
        scriptFile = self.makeSourceFile([
            "import sys",
            "if hasattr(sys.stdout, 'buffer'):",
            "    # Python 3",
            "    sys.stdout.buffer.write(b'hello world!\\n')",
            "    sys.stderr.buffer.write(b'goodbye world!\\n')",
            "else:",
            "    # Python 2",
            "    sys.stdout.write(b'hello world!\\n')",
            "    sys.stderr.write(b'goodbye world!\\n')",
            "sys.exit(1)"
            ])

def gotOutputAndValue(out_err_code):
            out, err, code = out_err_code
            self.assertEqual(out, b"hello world!\n")
            if _PY3:
                self.assertEqual(err, b"goodbye world!\n")
            else:
                self.assertEqual(err, b"goodbye world!" +
                                      os.linesep)
            self.assertEqual(code, 1)
        d = utils.getProcessOutputAndValue(self.exe, ["-u", scriptFile])
        return d.addCallback(gotOutputAndValue)

def test_outputSignal(self):
        """
        If the child process exits because of a signal, the L{Deferred}
        returned by L{getProcessOutputAndValue} fires a L{Failure} of a tuple
        containing the child's stdout, stderr, and the signal which caused
        it to exit.
        """
        # Use SIGKILL here because it's guaranteed to be delivered. Using
        # SIGHUP might not work in, e.g., a buildbot slave run under the
        # 'nohup' command.
        scriptFile = self.makeSourceFile([
            "import sys, os, signal",
            "sys.stdout.write('stdout bytes\\n')",
            "sys.stderr.write('stderr bytes\\n')",
            "sys.stdout.flush()",
            "sys.stderr.flush()",
            "os.kill(os.getpid(), signal.SIGKILL)"])

def gotOutputAndValue(out_err_sig):
            out, err, sig = out_err_sig
            self.assertEqual(out, b"stdout bytes\n")
            self.assertEqual(err, b"stderr bytes\n")
            self.assertEqual(sig, signal.SIGKILL)

d = utils.getProcessOutputAndValue(self.exe, ['-u', scriptFile])
        d = self.assertFailure(d, tuple)
        return d.addCallback(gotOutputAndValue)
    if platform.isWindows():
        test_outputSignal.skip = "Windows doesn't have real signals."

def _pathTest(self, utilFunc, check):
        dir = os.path.abspath(self.mktemp())
        os.makedirs(dir)
        scriptFile = self.makeSourceFile([
                "import os, sys",
                "sys.stdout.write(os.getcwd())"])
        d = utilFunc(self.exe, ['-u', scriptFile], path=dir)
        d.addCallback(check, dir.encode(sys.getfilesystemencoding()))
        return d

def test_getProcessOutputPath(self):
        """
        L{getProcessOutput} runs the given command with the working directory
        given by the C{path} parameter.
        """
        return self._pathTest(utils.getProcessOutput, self.assertEqual)

def test_getProcessValuePath(self):
        """
        L{getProcessValue} runs the given command with the working directory
        given by the C{path} parameter.
        """
        def check(result, ignored):
            self.assertEqual(result, 0)
        return self._pathTest(utils.getProcessValue, check)

def test_getProcessOutputAndValuePath(self):
        """
        L{getProcessOutputAndValue} runs the given command with the working
        directory given by the C{path} parameter.
        """
        def check(out_err_status, dir):
            out, err, status = out_err_status
            self.assertEqual(out, dir)
            self.assertEqual(status, 0)
        return self._pathTest(utils.getProcessOutputAndValue, check)

def _defaultPathTest(self, utilFunc, check):
        # Make another directory to mess around with.
        dir = os.path.abspath(self.mktemp())
        os.makedirs(dir)

scriptFile = self.makeSourceFile([
                "import os, sys, stat",
                # Fix the permissions so we can report the working directory.
                # On OS X (and maybe elsewhere), os.getcwd() fails with EACCES
                # if +x is missing from the working directory.
                "os.chmod(%r, stat.S_IXUSR)" % (dir,),
                "sys.stdout.write(os.getcwd())"])

# Switch to it, but make sure we switch back
        self.addCleanup(os.chdir, os.getcwd())
        os.chdir(dir)

# Get rid of all its permissions, but make sure they get cleaned up
        # later, because otherwise it might be hard to delete the trial
        # temporary directory.
        self.addCleanup(
            os.chmod, dir, stat.S_IMODE(os.stat('.').st_mode))
        os.chmod(dir, 0)

# Pass in -S so that if run using the coverage .pth trick, it won't be
        # loaded and cause Coverage to try and get the current working
        # directory (see the comments above why this can be a problem) on OSX.
        d = utilFunc(self.exe, ['-S', '-u', scriptFile])
        d.addCallback(check, dir.encode(sys.getfilesystemencoding()))
        return d

def test_getProcessOutputDefaultPath(self):
        """
        If no value is supplied for the C{path} parameter, L{getProcessOutput}
        runs the given command in the same working directory as the parent
        process and succeeds even if the current working directory is not
        accessible.
        """
        return self._defaultPathTest(utils.getProcessOutput, self.assertEqual)

def test_getProcessValueDefaultPath(self):
        """
        If no value is supplied for the C{path} parameter, L{getProcessValue}
        runs the given command in the same working directory as the parent
        process and succeeds even if the current working directory is not
        accessible.
        """
        def check(result, ignored):
            self.assertEqual(result, 0)
        return self._defaultPathTest(utils.getProcessValue, check)

def test_getProcessOutputAndValueDefaultPath(self):
        """
        If no value is supplied for the C{path} parameter,
        L{getProcessOutputAndValue} runs the given command in the same working
        directory as the parent process and succeeds even if the current
        working directory is not accessible.
        """
        def check(out_err_status, dir):
            out, err, status = out_err_status
            self.assertEqual(out, dir)
            self.assertEqual(status, 0)
        return self._defaultPathTest(
            utils.getProcessOutputAndValue, check)

class SuppressWarningsTests(unittest.SynchronousTestCase):
    """
    Tests for L{utils.suppressWarnings}.
    """
    def test_suppressWarnings(self):
        """
        L{utils.suppressWarnings} decorates a function so that the given
        warnings are suppressed.
        """
        result = []
        def showwarning(self, *a, **kw):
            result.append((a, kw))
        self.patch(warnings, "showwarning", showwarning)

def f(msg):
            warnings.warn(msg)
        g = utils.suppressWarnings(f, (('ignore',), dict(message="This is message")))

# Start off with a sanity check - calling the original function
        # should emit the warning.
        f("Sanity check message")
        self.assertEqual(len(result), 1)

# Now that that's out of the way, call the wrapped function, and
        # make sure no new warnings show up.
        g("This is message")
        self.assertEqual(len(result), 1)

# Finally, emit another warning which should not be ignored, and
        # make sure it is not.
        g("Unignored message")
        self.assertEqual(len(result), 2)

class DeferredSuppressedWarningsTests(SuppressedWarningsTests):
    """
    Tests for L{utils.runWithWarningsSuppressed}, the version that supports
    Deferreds.
    """
    # Override the non-Deferred-supporting function from the base class with
    # the function we are testing in this class:
    runWithWarningsSuppressed = staticmethod(utils.runWithWarningsSuppressed)

def test_deferredCallback(self):
        """
        If the function called by L{utils.runWithWarningsSuppressed} returns a
        C{Deferred}, the warning filters aren't removed until the Deferred
        fires.
        """
        filters = [(("ignore", ".*foo.*"), {}),
                   (("ignore", ".*bar.*"), {})]
        result = Deferred()
        self.runWithWarningsSuppressed(filters, lambda: result)
        warnings.warn("ignore foo")
        result.callback(3)
        warnings.warn("ignore foo 2")
        self.assertEqual(
            ["ignore foo 2"], [w['message'] for w in self.flushWarnings()])

def test_deferredErrback(self):
        """
        If the function called by L{utils.runWithWarningsSuppressed} returns a
        C{Deferred}, the warning filters aren't removed until the Deferred
        fires with an errback.
        """
        filters = [(("ignore", ".*foo.*"), {}),
                   (("ignore", ".*bar.*"), {})]
        result = Deferred()
        d = self.runWithWarningsSuppressed(filters, lambda: result)
        warnings.warn("ignore foo")
        result.errback(ZeroDivisionError())
        d.addErrback(lambda f: f.trap(ZeroDivisionError))
        warnings.warn("ignore foo 2")
        self.assertEqual(
            ["ignore foo 2"], [w['message'] for w in self.flushWarnings()])