FM / test

/usr/lib/python3/dist-packages/twisted/test/

/usr/lib/python3/dist-packages/twisted/test

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
cert.pem.no_trailing_newline	1414	0644	edit dl rm
crash_test_dummy.py	543	0644	edit dl rm
iosim.py	17843	0644	edit dl rm
key.pem.no_trailing_newline	1707	0644	edit dl rm
mock_win32process.py	1499	0644	edit dl rm
myrebuilder1.py	158	0644	edit dl rm
myrebuilder2.py	158	0644	edit dl rm
plugin_basic.py	943	0644	edit dl rm
plugin_extra1.py	407	0644	edit dl rm
plugin_extra2.py	579	0644	edit dl rm
process_cmdline.py	162	0644	edit dl rm
process_echoer.py	214	0644	edit dl rm
process_fds.py	945	0644	edit dl rm
process_getargv.py	283	0644	edit dl rm
process_getenv.py	268	0644	edit dl rm
process_linger.py	286	0644	edit dl rm
process_reader.py	188	0644	edit dl rm
process_signal.py	214	0644	edit dl rm
process_stdinreader.py	857	0644	edit dl rm
process_tester.py	1035	0644	edit dl rm
process_tty.py	130	0644	edit dl rm
process_twisted.py	1206	0644	edit dl rm
proto_helpers.py	28687	0644	edit dl rm
raiser.cpython-38-x86_64-linux-gnu.so	23864	0644	edit dl rm
reflect_helper_IE.py	61	0644	edit dl rm
reflect_helper_VE.py	82	0644	edit dl rm
reflect_helper_ZDE.py	47	0644	edit dl rm
server.pem	4444	0644	edit dl rm
ssl_helpers.py	1032	0644	edit dl rm
stdio_test_consumer.py	1216	0644	edit dl rm
stdio_test_halfclose.py	1938	0644	edit dl rm
stdio_test_hostpeer.py	1021	0644	edit dl rm
stdio_test_lastwrite.py	1206	0644	edit dl rm
stdio_test_loseconn.py	1548	0644	edit dl rm
stdio_test_producer.py	1507	0644	edit dl rm
stdio_test_write.py	923	0644	edit dl rm
stdio_test_writeseq.py	915	0644	edit dl rm
testutils.py	5321	0644	edit dl rm
test_abstract.py	3497	0644	edit dl rm
test_adbapi.py	26148	0644	edit dl rm
test_amp.py	110553	0644	edit dl rm
test_application.py	32816	0644	edit dl rm
test_compat.py	29278	0644	edit dl rm
test_context.py	1515	0644	edit dl rm
test_cooperator.py	21467	0644	edit dl rm
test_defer.py	103407	0644	edit dl rm
test_defgen.py	10701	0644	edit dl rm
test_dict.py	1442	0644	edit dl rm
test_dirdbm.py	6695	0644	edit dl rm
test_error.py	8593	0644	edit dl rm
test_factories.py	4637	0644	edit dl rm
test_failure.py	32288	0644	edit dl rm
test_fdesc.py	7371	0644	edit dl rm
test_finger.py	1997	0644	edit dl rm
test_formmethod.py	3645	0644	edit dl rm
test_ftp.py	130321	0644	edit dl rm
test_ftp_options.py	2685	0644	edit dl rm
test_htb.py	3190	0644	edit dl rm
test_ident.py	7015	0644	edit dl rm
test_internet.py	46422	0644	edit dl rm
test_iosim.py	9060	0644	edit dl rm
test_iutils.py	13445	0644	edit dl rm
test_lockfile.py	15506	0644	edit dl rm
test_log.py	36330	0644	edit dl rm
test_logfile.py	18227	0644	edit dl rm
test_loopback.py	14486	0644	edit dl rm
test_main.py	2501	0644	edit dl rm
test_memcache.py	25143	0644	edit dl rm
test_modules.py	17888	0644	edit dl rm
test_monkey.py	5637	0644	edit dl rm
test_nooldstyle.py	5955	0644	edit dl rm
test_paths.py	74357	0644	edit dl rm
test_pcp.py	12551	0644	edit dl rm
test_persisted.py	14624	0644	edit dl rm
test_plugin.py	26113	0644	edit dl rm
test_policies.py	33499	0644	edit dl rm
test_postfix.py	4243	0644	edit dl rm
test_process.py	86120	0644	edit dl rm
test_protocols.py	7450	0644	edit dl rm
test_randbytes.py	3357	0644	edit dl rm
test_rebuild.py	8495	0644	edit dl rm
test_reflect.py	26085	0644	edit dl rm
test_roots.py	1812	0644	edit dl rm
test_shortcut.py	1991	0644	edit dl rm
test_sip.py	25284	0644	edit dl rm
test_sob.py	5632	0644	edit dl rm
test_socks.py	17738	0644	edit dl rm
test_ssl.py	23846	0644	edit dl rm
test_sslverify.py	113322	0644	edit dl rm
test_stateful.py	2021	0644	edit dl rm
test_stdio.py	13156	0644	edit dl rm
test_strerror.py	5183	0644	edit dl rm
test_stringtransport.py	13264	0644	edit dl rm
test_strports.py	1797	0644	edit dl rm
test_task.py	39326	0644	edit dl rm
test_tcp.py	65820	0644	edit dl rm
test_tcp_internals.py	13117	0644	edit dl rm
test_text.py	6456	0644	edit dl rm
test_threadable.py	3738	0644	edit dl rm
test_threadpool.py	22204	0644	edit dl rm
test_threads.py	13268	0644	edit dl rm
test_tpfile.py	1601	0644	edit dl rm
test_twistd.py	74201	0644	edit dl rm
test_twisted.py	11173	0644	edit dl rm
test_udp.py	24676	0644	edit dl rm
test_unix.py	15157	0644	edit dl rm
test_usage.py	23645	0644	edit dl rm
__init__.py	103	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/test/test_pcp.py (12551B)

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

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

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

# 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.

cmd: