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_pcp.py

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

__version__ = "$Revision: 1.5 $"[11:-2]

from twisted.protocols import pcp
from twisted.trial import unittest

# Goal:

# Take a Protocol instance.  Own all outgoing data - anything that
# would go to p.transport.write.  Own all incoming data - anything
# that comes to p.dataReceived.

# I need:
# Something with the AbstractFileDescriptor interface.
# That is:
#  - acts as a Transport
#    - has a method write()
#    - which buffers
#  - acts as a Consumer
#    - has a registerProducer, unRegisterProducer
#    - tells the Producer to back off (pauseProducing) when its buffer is full.
#    - tells the Producer to resumeProducing when its buffer is not so full.
#  - acts as a Producer
#    - calls registerProducer
#    - calls write() on consumers
#    - honors requests to pause/resume producing
#    - honors stopProducing, and passes it along to upstream Producers

class DummyTransport:
    """A dumb transport to wrap around."""

def __init__(self):
        self._writes = []

def write(self, data):
        self._writes.append(data)

def getvalue(self):
        return "".join(self._writes)

class DummyProducer:
    resumed = False
    stopped = False
    paused = False

def __init__(self, consumer):
        self.consumer = consumer

def resumeProducing(self):
        self.resumed = True
        self.paused = False

def pauseProducing(self):
        self.paused = True

def stopProducing(self):
        self.stopped = True

class DummyConsumer(DummyTransport):
    producer = None
    finished = False
    unregistered = True

def registerProducer(self, producer, streaming):
        self.producer = (producer, streaming)

def unregisterProducer(self):
        self.unregistered = True

def finish(self):
        self.finished = True

class TransportInterfaceTests(unittest.TestCase):
    proxyClass = pcp.BasicProducerConsumerProxy

def setUp(self):
        self.underlying = DummyConsumer()
        self.transport = self.proxyClass(self.underlying)

def testWrite(self):
        self.transport.write("some bytes")

class ConsumerInterfaceTest:
    """Test ProducerConsumerProxy as a Consumer.

Normally we have ProducingServer -> ConsumingTransport.

If I am to go between (Server -> Shaper -> Transport), I have to
    play the role of Consumer convincingly for the ProducingServer.
    """

def setUp(self):
        self.underlying = DummyConsumer()
        self.consumer = self.proxyClass(self.underlying)
        self.producer = DummyProducer(self.consumer)

def testRegisterPush(self):
        self.consumer.registerProducer(self.producer, True)
        ## Consumer should NOT have called PushProducer.resumeProducing
        self.assertFalse(self.producer.resumed)

## I'm I'm just a proxy, should I only do resumeProducing when
    ## I get poked myself?
    # def testRegisterPull(self):
    #    self.consumer.registerProducer(self.producer, False)
    #    ## Consumer SHOULD have called PushProducer.resumeProducing
    #    self.assertTrue(self.producer.resumed)

def testUnregister(self):
        self.consumer.registerProducer(self.producer, False)
        self.consumer.unregisterProducer()
        # Now when the consumer would ordinarily want more data, it
        # shouldn't ask producer for it.
        # The most succinct way to trigger "want more data" is to proxy for
        # a PullProducer and have someone ask me for data.
        self.producer.resumed = False
        self.consumer.resumeProducing()
        self.assertFalse(self.producer.resumed)

def testFinish(self):
        self.consumer.registerProducer(self.producer, False)
        self.consumer.finish()
        # I guess finish should behave like unregister?
        self.producer.resumed = False
        self.consumer.resumeProducing()
        self.assertFalse(self.producer.resumed)

class ProducerInterfaceTest:
    """Test ProducerConsumerProxy as a Producer.

Normally we have ProducingServer -> ConsumingTransport.

If I am to go between (Server -> Shaper -> Transport), I have to
    play the role of Producer convincingly for the ConsumingTransport.
    """

def setUp(self):
        self.consumer = DummyConsumer()
        self.producer = self.proxyClass(self.consumer)

def testRegistersProducer(self):
        self.assertEqual(self.consumer.producer[0], self.producer)

def testPause(self):
        self.producer.pauseProducing()
        self.producer.write("yakkity yak")
        self.assertFalse(
            self.consumer.getvalue(), "Paused producer should not have sent data."
        )

def testResume(self):
        self.producer.pauseProducing()
        self.producer.resumeProducing()
        self.producer.write("yakkity yak")
        self.assertEqual(self.consumer.getvalue(), "yakkity yak")

def testResumeNoEmptyWrite(self):
        self.producer.pauseProducing()
        self.producer.resumeProducing()
        self.assertEqual(
            len(self.consumer._writes), 0, "Resume triggered an empty write."
        )

def testResumeBuffer(self):
        self.producer.pauseProducing()
        self.producer.write("buffer this")
        self.producer.resumeProducing()
        self.assertEqual(self.consumer.getvalue(), "buffer this")

def testStop(self):
        self.producer.stopProducing()
        self.producer.write("yakkity yak")
        self.assertFalse(
            self.consumer.getvalue(), "Stopped producer should not have sent data."
        )

class PCP_ConsumerInterfaceTests(ConsumerInterfaceTest, unittest.TestCase):
    proxyClass = pcp.BasicProducerConsumerProxy

class PCPII_ConsumerInterfaceTests(ConsumerInterfaceTest, unittest.TestCase):
    proxyClass = pcp.ProducerConsumerProxy

class PCP_ProducerInterfaceTests(ProducerInterfaceTest, unittest.TestCase):
    proxyClass = pcp.BasicProducerConsumerProxy

class PCPII_ProducerInterfaceTests(ProducerInterfaceTest, unittest.TestCase):
    proxyClass = pcp.ProducerConsumerProxy

class ProducerProxyTests(unittest.TestCase):
    """Producer methods on me should be relayed to the Producer I proxy."""

proxyClass = pcp.BasicProducerConsumerProxy

def setUp(self):
        self.proxy = self.proxyClass(None)
        self.parentProducer = DummyProducer(self.proxy)
        self.proxy.registerProducer(self.parentProducer, True)

def testStop(self):
        self.proxy.stopProducing()
        self.assertTrue(self.parentProducer.stopped)

class ConsumerProxyTests(unittest.TestCase):
    """Consumer methods on me should be relayed to the Consumer I proxy."""

proxyClass = pcp.BasicProducerConsumerProxy

def setUp(self):
        self.underlying = DummyConsumer()
        self.consumer = self.proxyClass(self.underlying)

def testWrite(self):
        # NOTE: This test only valid for streaming (Push) systems.
        self.consumer.write("some bytes")
        self.assertEqual(self.underlying.getvalue(), "some bytes")

def testFinish(self):
        self.consumer.finish()
        self.assertTrue(self.underlying.finished)

def testUnregister(self):
        self.consumer.unregisterProducer()
        self.assertTrue(self.underlying.unregistered)

class PullProducerTest:
    def setUp(self):
        self.underlying = DummyConsumer()
        self.proxy = self.proxyClass(self.underlying)
        self.parentProducer = DummyProducer(self.proxy)
        self.proxy.registerProducer(self.parentProducer, True)

def testHoldWrites(self):
        self.proxy.write("hello")
        # Consumer should get no data before it says resumeProducing.
        self.assertFalse(
            self.underlying.getvalue(), "Pulling Consumer got data before it pulled."
        )

def testPull(self):
        self.proxy.write("hello")
        self.proxy.resumeProducing()
        self.assertEqual(self.underlying.getvalue(), "hello")

def testMergeWrites(self):
        self.proxy.write("hello ")
        self.proxy.write("sunshine")
        self.proxy.resumeProducing()
        nwrites = len(self.underlying._writes)
        self.assertEqual(
            nwrites, 1, "Pull resulted in %d writes instead " "of 1." % (nwrites,)
        )
        self.assertEqual(self.underlying.getvalue(), "hello sunshine")

def testLateWrite(self):
        # consumer sends its initial pull before we have data
        self.proxy.resumeProducing()
        self.proxy.write("data")
        # This data should answer that pull request.
        self.assertEqual(self.underlying.getvalue(), "data")

class PCP_PullProducerTests(PullProducerTest, unittest.TestCase):
    class proxyClass(pcp.BasicProducerConsumerProxy):
        iAmStreaming = False

class PCPII_PullProducerTests(PullProducerTest, unittest.TestCase):
    class proxyClass(pcp.ProducerConsumerProxy):
        iAmStreaming = False

# Buffering!

class BufferedConsumerTests(unittest.TestCase):
    """As a consumer, ask the producer to pause after too much data."""

proxyClass = pcp.ProducerConsumerProxy

def setUp(self):
        self.underlying = DummyConsumer()
        self.proxy = self.proxyClass(self.underlying)
        self.proxy.bufferSize = 100

self.parentProducer = DummyProducer(self.proxy)
        self.proxy.registerProducer(self.parentProducer, True)

def testRegisterPull(self):
        self.proxy.registerProducer(self.parentProducer, False)
        ## Consumer SHOULD have called PushProducer.resumeProducing
        self.assertTrue(self.parentProducer.resumed)

def testPauseIntercept(self):
        self.proxy.pauseProducing()
        self.assertFalse(self.parentProducer.paused)

def testResumeIntercept(self):
        self.proxy.pauseProducing()
        self.proxy.resumeProducing()
        # With a streaming producer, just because the proxy was resumed is
        # not necessarily a reason to resume the parent producer.  The state
        # of the buffer should decide that.
        self.assertFalse(self.parentProducer.resumed)

def testTriggerPause(self):
        """Make sure I say \"when.\" """

# Pause the proxy so data sent to it builds up in its buffer.
        self.proxy.pauseProducing()
        self.assertFalse(self.parentProducer.paused, "don't pause yet")
        self.proxy.write("x" * 51)
        self.assertFalse(self.parentProducer.paused, "don't pause yet")
        self.proxy.write("x" * 51)
        self.assertTrue(self.parentProducer.paused)

def testTriggerResume(self):
        """Make sure I resumeProducing when my buffer empties."""
        self.proxy.pauseProducing()
        self.proxy.write("x" * 102)
        self.assertTrue(self.parentProducer.paused, "should be paused")
        self.proxy.resumeProducing()
        # Resuming should have emptied my buffer, so I should tell my
        # parent to resume too.
        self.assertFalse(self.parentProducer.paused, "Producer should have resumed.")
        self.assertFalse(self.proxy.producerPaused)

class BufferedPullTests(unittest.TestCase):
    class proxyClass(pcp.ProducerConsumerProxy):
        iAmStreaming = False

def _writeSomeData(self, data):
            pcp.ProducerConsumerProxy._writeSomeData(self, data[:100])
            return min(len(data), 100)

def setUp(self):
        self.underlying = DummyConsumer()
        self.proxy = self.proxyClass(self.underlying)
        self.proxy.bufferSize = 100

self.parentProducer = DummyProducer(self.proxy)
        self.proxy.registerProducer(self.parentProducer, False)

def testResumePull(self):
        # If proxy has no data to send on resumeProducing, it had better pull
        # some from its PullProducer.
        self.parentProducer.resumed = False
        self.proxy.resumeProducing()
        self.assertTrue(self.parentProducer.resumed)

def testLateWriteBuffering(self):
        # consumer sends its initial pull before we have data
        self.proxy.resumeProducing()
        self.proxy.write("datum" * 21)
        # This data should answer that pull request.
        self.assertEqual(self.underlying.getvalue(), "datum" * 20)
        # but there should be some left over
        self.assertEqual(self.proxy._buffer, ["datum"])

# TODO:
#  test that web request finishing bug (when we weren't proxying
#    unregisterProducer but were proxying finish, web file transfers
#    would hang on the last block.)
#  test what happens if writeSomeBytes decided to write zero bytes.

${this.title}

Code Editor : test_pcp.py