Da

✏️ Editing: 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.

📁 Sid Gifari File Manager