Gecko [ gandhischool.edu.np ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	475 B	-rw-r--r--
cert.pem.no_trailing_newline	1.38 KB	-rw-r--r--
crash_test_dummy.py	549 B	-rw-r--r--
ignore_test_failure.py	30.72 KB	-rw-r--r--
iosim.py	17.74 KB	-rw-r--r--
key.pem.no_trailing_newline	1.67 KB	-rw-r--r--
mock_win32process.py	1.27 KB	-rw-r--r--
myrebuilder1.py	151 B	-rw-r--r--
myrebuilder2.py	151 B	-rw-r--r--
plugin_basic.py	901 B	-rw-r--r--
plugin_extra1.py	392 B	-rw-r--r--
plugin_extra2.py	550 B	-rw-r--r--
process_cmdline.py	123 B	-rw-r--r--
process_echoer.py	214 B	-rw-r--r--
process_fds.py	984 B	-rw-r--r--
process_getargv.py	234 B	-rw-r--r--
process_getenv.py	268 B	-rw-r--r--
process_linger.py	297 B	-rw-r--r--
process_reader.py	178 B	-rw-r--r--
process_signal.py	220 B	-rw-r--r--
process_stdinreader.py	739 B	-rw-r--r--
process_tester.py	779 B	-rw-r--r--
process_tty.py	130 B	-rw-r--r--
process_twisted.py	1.15 KB	-rw-r--r--
proto_helpers.py	1.34 KB	-rw-r--r--
reflect_helper_IE.py	60 B	-rw-r--r--
reflect_helper_VE.py	81 B	-rw-r--r--
reflect_helper_ZDE.py	48 B	-rw-r--r--
server.pem	4.34 KB	-rw-r--r--
ssl_helpers.py	1.61 KB	-rw-r--r--
stdio_test_consumer.py	1.14 KB	-rw-r--r--
stdio_test_halfclose.py	2 KB	-rw-r--r--
stdio_test_hostpeer.py	1.06 KB	-rw-r--r--
stdio_test_lastwrite.py	1.13 KB	-rw-r--r--
stdio_test_loseconn.py	1.55 KB	-rw-r--r--
stdio_test_producer.py	1.45 KB	-rw-r--r--
stdio_test_write.py	902 B	-rw-r--r--
stdio_test_writeseq.py	894 B	-rw-r--r--
test_abstract.py	3.35 KB	-rw-r--r--
test_adbapi.py	25.56 KB	-rw-r--r--
test_amp.py	107.92 KB	-rw-r--r--
test_application.py	33.34 KB	-rw-r--r--
test_compat.py	17.75 KB	-rw-r--r--
test_context.py	1.43 KB	-rw-r--r--
test_cooperator.py	20.84 KB	-rw-r--r--
test_defer.py	121.86 KB	-rw-r--r--
test_defgen.py	10.38 KB	-rw-r--r--
test_dict.py	1.49 KB	-rw-r--r--
test_dirdbm.py	6.69 KB	-rw-r--r--
test_error.py	8.72 KB	-rw-r--r--
test_factories.py	4.46 KB	-rw-r--r--
test_fdesc.py	7.28 KB	-rw-r--r--
test_finger.py	1.85 KB	-rw-r--r--
test_formmethod.py	3.87 KB	-rw-r--r--
test_ftp.py	126.96 KB	-rw-r--r--
test_ftp_options.py	2.63 KB	-rw-r--r--
test_htb.py	3.08 KB	-rw-r--r--
test_ident.py	6.56 KB	-rw-r--r--
test_internet.py	45.38 KB	-rw-r--r--
test_iosim.py	8.7 KB	-rw-r--r--
test_iutils.py	13.31 KB	-rw-r--r--
test_lockfile.py	14.98 KB	-rw-r--r--
test_log.py	35.23 KB	-rw-r--r--
test_logfile.py	17.47 KB	-rw-r--r--
test_loopback.py	13.99 KB	-rw-r--r--
test_main.py	2.12 KB	-rw-r--r--
test_memcache.py	24.69 KB	-rw-r--r--
test_modules.py	17.02 KB	-rw-r--r--
test_monkey.py	5.39 KB	-rw-r--r--
test_paths.py	69.31 KB	-rw-r--r--
test_pcp.py	12.23 KB	-rw-r--r--
test_persisted.py	13.09 KB	-rw-r--r--
test_plugin.py	24.79 KB	-rw-r--r--
test_policies.py	32.29 KB	-rw-r--r--
test_postfix.py	4.32 KB	-rw-r--r--
test_process.py	85.16 KB	-rw-r--r--
test_protocols.py	7.16 KB	-rw-r--r--
test_randbytes.py	3.19 KB	-rw-r--r--
test_rebuild.py	7.15 KB	-rw-r--r--
test_reflect.py	23.9 KB	-rw-r--r--
test_roots.py	1.6 KB	-rw-r--r--
test_shortcut.py	1.92 KB	-rw-r--r--
test_sip.py	24.9 KB	-rw-r--r--
test_sob.py	5.53 KB	-rw-r--r--
test_socks.py	17.09 KB	-rw-r--r--
test_ssl.py	22.75 KB	-rw-r--r--
test_sslverify.py	113.89 KB	-rw-r--r--
test_stateful.py	1.97 KB	-rw-r--r--
test_stdio.py	12.43 KB	-rw-r--r--
test_strerror.py	5.12 KB	-rw-r--r--
test_strports.py	1.67 KB	-rw-r--r--
test_task.py	45.26 KB	-rw-r--r--
test_tcp.py	64.27 KB	-rw-r--r--
test_tcp_internals.py	12.73 KB	-rw-r--r--
test_text.py	6.22 KB	-rw-r--r--
test_threadable.py	3.26 KB	-rw-r--r--
test_threadpool.py	21.44 KB	-rw-r--r--
test_threads.py	12.9 KB	-rw-r--r--
test_tpfile.py	1.51 KB	-rw-r--r--
test_twistd.py	72.04 KB	-rw-r--r--
test_twisted.py	6.13 KB	-rw-r--r--
test_udp.py	24.43 KB	-rw-r--r--
test_unix.py	13.26 KB	-rw-r--r--
test_usage.py	22.76 KB	-rw-r--r--
testutils.py	5.06 KB	-rw-r--r--

Code Editor : test_iutils.py

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

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

import os
import signal
import stat
import sys
import warnings
from unittest import skipIf

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

class ProcessUtilsTests(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':",
                "    s = bytes([s])",
                "    sys.stdout.buffer.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",
                "sys.stdout.buffer.write(b'hello world!\\n')",
                "sys.stderr.buffer.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")
            self.assertEqual(err, b"goodbye world!\n")
            self.assertEqual(code, 1)

d = utils.getProcessOutputAndValue(self.exe, ["-u", scriptFile])
        return d.addCallback(gotOutputAndValue)

@skipIf(platform.isWindows(), "Windows doesn't have real signals.")
    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)

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", "cdir = os.getcwd()", "sys.stdout.write(cdir)"]
        )

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

# Remember its default permissions.
        originalMode = stat.S_IMODE(os.stat(".").st_mode)

# On macOS Catalina (and maybe elsewhere), os.getcwd() sometimes fails
        # with EACCES if u+rx is missing from the working directory, so don't
        # reduce it further than this.
        os.chmod(dir, stat.S_IXUSR | stat.S_IRUSR)

# Restore the permissions to their original state later (probably
        # adding at least u+w), because otherwise it might be hard to delete
        # the trial temporary directory.
        self.addCleanup(os.chmod, dir, originalMode)

d = utilFunc(self.exe, ["-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)

def test_get_processOutputAndValueStdin(self):
        """
        Standard input can be made available to the child process by passing
        bytes for the `stdinBytes` parameter.
        """
        scriptFile = self.makeSourceFile(
            [
                "import sys",
                "sys.stdout.write(sys.stdin.read())",
            ]
        )
        stdinBytes = b"These are the bytes to see."
        d = utils.getProcessOutputAndValue(
            self.exe,
            ["-u", scriptFile],
            stdinBytes=stdinBytes,
        )

def gotOutputAndValue(out_err_code):
            out, err, code = out_err_code
            # Avoid making an exact equality comparison in case there is extra
            # random output on stdout (warnings, stray print statements,
            # logging, who knows).
            self.assertIn(stdinBytes, out)
            self.assertEqual(0, code)

d.addCallback(gotOutputAndValue)
        return d

class SuppressWarningsTests(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()])

${this.title}

Code Editor : test_iutils.py