Source code for panoramisk.testing

from __future__ import unicode_literals
import asyncio
from unittest import mock

from . import manager
from . import utils


MagicMock = mock.MagicMock
patch = mock.patch
call = mock.call


class AMIProtocol(manager.AMIProtocol):

    debug_count = [0]

    def connection_made(self, transport):
        super(AMIProtocol, self).connection_made(transport)
        self.transport = MagicMock()

    def send(self, data, as_list=False):
        utils.IdGenerator.reset(uid='transaction_uid')
        future = super(AMIProtocol, self).send(data, as_list=as_list)
        if getattr(self.factory, 'stream', None) is not None:
            with open(self.factory.stream, 'rb') as fd:
                for resp in fd.read().split(b'\n\n'):
                    self.data_received(resp + b'\n\n')
                    if future.done():
                        break
            if not future.done():  # pragma: no cover
                print(self.responses)
                raise AssertionError("Future's result was never set")
        return future


[docs]class Manager(manager.Manager): fixtures_dir = None defaults = manager.Manager.defaults.copy() def __init__(self, **config): self.defaults.update( protocol_factory=AMIProtocol, stream=None) super(Manager, self).__init__(**config) self.stream = self.config.get('stream') if self.loop is None: self.loop = asyncio.get_event_loop() protocol = AMIProtocol() protocol.factory = manager protocol.connection_made(mock.MagicMock()) future = asyncio.Future() future.set_result((mock.MagicMock(), protocol)) self.protocol = protocol self.connection_made(future) utils.IdGenerator.reset(uid='transaction_uid') utils.EOL = '\n'