wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-06 05:14:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

merged test class

Browse Source
This commit is contained in:
fawce
2012-02-24 09:20:22 -05:00
parent 8e840e472e 6e1cc72900
commit 3be2264a86
10 changed files with 414 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/zipline.rst
+43
View File
@@ -0,0 +1,43 @@
zipline Package
===============
:mod:`zipline` Package
----------------------
.. automodule:: zipline.__init__
:members:
:undoc-members:
:show-inheritance:
:mod:`messaging` Module
-----------------------
.. automodule:: zipline.messaging
:members:
:undoc-members:
:show-inheritance:
:mod:`sources` Module
---------------------
.. automodule:: zipline.sources
:members:
:undoc-members:
:show-inheritance:
:mod:`util` Module
------------------
.. automodule:: zipline.util
:members:
:undoc-members:
:show-inheritance:
Subpackages
-----------
.. toctree::
zipline.test
zipline.transforms
docs/zipline.test.rst
+19
View File
@@ -0,0 +1,19 @@
test Package
============
:mod:`client` Module
--------------------
.. automodule:: zipline.test.client
:members:
:undoc-members:
:show-inheritance:
:mod:`test_messaging` Module
----------------------------
.. automodule:: zipline.test.test_messaging
:members:
:undoc-members:
:show-inheritance:
docs/zipline.transforms.rst
+19
View File
@@ -0,0 +1,19 @@
transforms Package
==================
:mod:`transforms` Package
-------------------------
.. automodule:: zipline.transforms
:members:
:undoc-members:
:show-inheritance:
:mod:`technical` Module
-----------------------
.. automodule:: zipline.transforms.technical
:members:
:undoc-members:
:show-inheritance:
pavement.py
+1
View File
@@ -185,4 +185,5 @@ def apidocs():
Recursively autogenerate the Sphinx autodoc for the module and
its submodules.
"""
call('rm docs/zipline.*.rst', shell=True)
call('sphinx-apidoc -o docs/ zipline', shell=True)
zipline/messaging.py
+276 -89
View File
@@ -5,15 +5,144 @@ import json
import uuid
import datetime
import zipline.util as qutil
import zmq
class Controller(object):
"""
A broker of sorts.
"""
polling = False
debug = False
def __init__(self, pull_socket, pub_socket, context=None, logging = None):
if not context:
self._ctx = zmq.Context()
else:
self._ctx = context
self.pull_socket = pull_socket
self.pub_socket = pub_socket
self.pull = self._ctx.socket(zmq.PULL)
self.pub = self._ctx.socket(zmq.PUB)
self.associated = [self.pull, self.pub]
if logging:
self.logging = logging
self.dologging = True
else:
self.logging = False
self.dologging = False
self.success = 0
self.failed = 0
try:
self.pull.bind(pull_socket)
except zmq.ZMQError:
raise Exception('Cannot not bind on %s' % pull_socket)
try:
self.pub.bind(pub_socket)
except zmq.ZMQError:
raise Exception('Cannot not bind on %s' % pub_socket)
def run(self, debug_step=False, stats=True):
self.polling = True
if self.debug or debug_step:
return self._poll_verbose(True, stats)
else:
return self._poll(False, stats)
def _poll(self, debug_step, stats):
while self.polling:
try:
self.logging.info('msg')
self.pub.send(self.pull.recv())
except KeyboardInterrupt:
self.polling = False
break
except Exception as e:
# Its common to wrap these in wildcard exceptions so
# that we don't loose messages, ever
self.logging.error(str(e))
self.failed += 1
continue
def _poll_verbose(self, debug_step, stats):
while self.polling:
try:
if debug_step:
msg = self.pull.recv()
if self.dologging:
self.logging.info(msg)
self.pub.send(msg)
self.success += 1
except KeyboardInterrupt:
self.polling = False
break
except Exception as e:
# Its common to wrap these in wildcard exceptions so
# that we don't loose messages, ever
self.logging.error(str(e))
self.failed += 1
continue
def qos(self):
return float(self.success) / (self.success + self.failed)
def destroy(self):
"""
Manual cleanup.
"""
self.polling = False
for asoc in self.associated:
asoc.close()
#if self._ctx:
#self._ctx.destroy()
def __del__(self):
self.destroy()
def message_sender(self):
"""
Spin off a socket used for sending messages to this
controller.
"""
s = self._ctx.socket(zmq.PUSH)
s.connect(self.pull_socket)
s.setsockopt(zmq.LINGER, -1)
self.associated.append(s)
return s
def message_listener(self):
"""
Spin off a socket used for receiving messages from this
controller.
"""
s = self._ctx.socket(zmq.SUB)
s.connect(self.pub_socket)
s.setsockopt(zmq.SUBSCRIBE, '')
self.associated.append(s)
return s
class Component(object):
def __init__(self):
"""
:addresses: a dict of name_string -> zmq port address strings. Must have the following entries::
- sync_address: socket address used for synchronizing the start of all workers, heartbeating, and exit notification
will be used in REP/REQ sockets. Bind is always on the REP side.
- control_address: socket address used for controlling and
monitoring the status of the simulation
- data_address: socket address used for data sources to stream their records.
will be used in PUSH/PULL sockets between data sources and a ParallelBuffer (aka the Feed). Bind
will always be on the PULL side (we always have N producers and 1 consumer)
@@ -24,31 +153,51 @@ class Component(object):
will always be on the PULL side (we always have N producers and 1 consumer)
- result_address: socket address used to publish merged data source feed and transforms to clients
will be used in PUB/SUB from one Merge to one or many clients. Bind is always on the PUB side.
Bind/Connect methods will return the correct socket type for each address. Any sockets on which recv is expected to be called
will also return a Poller.
"""
"""
self.zmq = None
self.context = None
self.addresses = None
self.out_socket = None
self.gevent_needed = False
self.killed = False
# TODO: could probably mkae this into a property instead of a
# method
def get_id(self):
NotImplemented
raise NotImplementedError
def open(self):
NotImplemented
raise NotImplementedError
def destroy(self):
"""
Tear down after normal operation.
"""
raise NotImplementedError
def kill(self):
"""
Tear down ( fast ) as a mode of failure in the
simulation.
"""
raise NotImplementedError
def do_work(self):
NotImplemented
raise NotImplementedError
def run(self):
fail = None
try:
#TODO: can't initialize these values in the __init__?
self.done = False
self.sockets = []
if self.gevent_needed:
qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id()))
import gevent_zeromq
@@ -56,25 +205,34 @@ class Component(object):
else:
import zmq
self.zmq = zmq
self.context = self.zmq.Context()
self.open()
self.setup_sync()
self.setup_control()
self.loop()
#close all the sockets
for sock in self.sockets:
sock.close()
except Exception as e:
qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id()))
raise e
fail = e
finally:
if(self.context != None):
self.context.destroy()
if fail:
raise fail
def loop(self):
while not self.done:
self.confirm()
self.do_work()
def signal_done(self):
#notify down stream components that we're done
if(self.out_socket != None):
@@ -84,15 +242,17 @@ class Component(object):
self.receive_sync_ack()
#notify internal work look that we're done
self.done = True
# TODO: probably don't need a method here ... or move into
# higher level framing protocol
def is_done_message(self, message):
return message == "DONE"
def confirm(self):
def confirm(self):
# send a synchronization request to the host
self.sync_socket.send(self.get_id() + ":RUN")
self.receive_sync_ack()
def receive_sync_ack(self):
# wait for synchronization reply from the host
socks = dict(self.sync_poller.poll(2000)) #timeout after 2 seconds.
@@ -100,32 +260,31 @@ class Component(object):
message = self.sync_socket.recv()
else:
raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id()))
def bind_data(self):
return self.bind_pull_socket(self.addresses['data_address'])
def connect_data(self):
return self.connect_push_socket(self.addresses['data_address'])
def bind_feed(self):
return self.bind_pub_socket(self.addresses['feed_address'])
def connect_feed(self):
return self.connect_sub_socket(self.addresses['feed_address'])
def bind_merge(self):
return self.bind_pull_socket(self.addresses['merge_address'])
def connect_merge(self):
return self.connect_push_socket(self.addresses['merge_address'])
def bind_result(self):
return self.bind_pub_socket(self.addresses['result_address'])
def connect_result(self):
return self.connect_sub_socket(self.addresses['result_address'])
def bind_pull_socket(self, addr):
pull_socket = self.context.socket(self.zmq.PULL)
pull_socket.bind(addr)
@@ -133,7 +292,7 @@ class Component(object):
poller.register(pull_socket, self.zmq.POLLIN)
self.sockets.append(pull_socket)
return pull_socket, poller
def connect_push_socket(self, addr):
push_socket = self.context.socket(self.zmq.PUSH)
push_socket.connect(addr)
@@ -141,14 +300,14 @@ class Component(object):
self.sockets.append(push_socket)
self.out_socket = push_socket
return push_socket
def bind_pub_socket(self, addr):
pub_socket = self.context.socket(self.zmq.PUB)
pub_socket.bind(addr)
#pub_socket.setsockopt(self.zmq.LINGER,0)
self.out_socket = pub_socket
return pub_socket
def connect_sub_socket(self, addr):
sub_socket = self.context.socket(self.zmq.SUB)
sub_socket.connect(addr)
@@ -157,7 +316,15 @@ class Component(object):
poller.register(sub_socket, self.zmq.POLLIN)
self.sockets.append(sub_socket)
return sub_socket, poller
def setup_control(self):
"""
Set up the control socket. Used to monitor the the
overall status of the simulation and to forcefully tear
down the simulation in case of a failure.
"""
pass
def setup_sync(self):
qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id()))
self.sync_socket = self.context.socket(self.zmq.REQ)
@@ -166,13 +333,17 @@ class Component(object):
self.sync_poller = self.zmq.Poller()
self.sync_poller.register(self.sync_socket, self.zmq.POLLIN)
self.sockets.append(self.sync_socket)
class ComponentHost(Component):
"""Component that can launch multiple sub-components, synchronize their start, and then wait for all
components to be finished."""
"""
Component that can launch multiple sub-components, synchronize their start, and then wait for all
components to be finished.
"""
def __init__(self, addresses, gevent_needed=False):
Component.__init__(self)
self.addresses = addresses
#workaround for defect in threaded use of strptime: http://bugs.python.org/issue11108
qutil.parse_date("2012/02/13-10:04:28.114")
self.components = {}
@@ -182,25 +353,37 @@ class ComponentHost(Component):
self.merge = MergedParallelBuffer()
self.passthrough = PassthroughTransform()
self.gevent_needed = gevent_needed
self.controller = None
#register the feed and the merge
self.register_components([self.feed, self.merge, self.passthrough])
def register_components(self, component_list):
for component in component_list:
def register_controller(self, controller):
self.controller = controller
for component in self.components.itervalues():
component.controller = controller
def register_components(self, components):
for component in components:
component.gevent_needed = self.gevent_needed
component.addresses = self.addresses
if self.controller:
component.controller = self.controller
self.components[component.get_id()] = component
self.sync_register[component.get_id()] = datetime.datetime.utcnow()
if(isinstance(component, DataSource)):
self.feed.add_source(component.get_id())
if(isinstance(component, BaseTransform)):
self.merge.add_source(component.get_id())
def unregister_component(self, component_id):
del(self.components[component_id])
del(self.sync_register[component_id])
def setup_sync(self):
"""Start the sync server."""
qutil.LOGGER.debug("Connecting sync server.")
@@ -209,11 +392,12 @@ class ComponentHost(Component):
self.poller = self.zmq.Poller()
self.poller.register(self.sync_socket, self.zmq.POLLIN)
self.sockets.append(self.sync_socket)
def open(self):
for component in self.components.values():
self.launch_component(component)
self.launch_controller()
def is_timed_out(self):
cur_time = datetime.datetime.utcnow()
if(len(self.components) == 0):
@@ -224,7 +408,7 @@ class ComponentHost(Component):
qutil.LOGGER.info("Time out for {source}. Current component registery: {reg}".format(source=source, reg=self.components))
return True
return False
def loop(self):
while not self.is_timed_out():
# wait for synchronization request
@@ -240,20 +424,23 @@ class ComponentHost(Component):
if(self.is_done_message(status)):
qutil.LOGGER.info("{id} is DONE".format(id=sync_id))
self.unregister_component(sync_id)
else:
else:
self.sync_register[sync_id] = datetime.datetime.utcnow()
#qutil.LOGGER.info("confirmed {id}".format(id=msg))
# send synchronization reply
self.sync_socket.send('ack', self.zmq.NOBLOCK)
def launch_controller(self, controller):
NotImplemented
def launch_component(self, component):
NotImplemented
class ParallelBuffer(Component):
"""Connects to N PULL sockets, publishing all messages received to a PUB socket.
Published messages are guaranteed to be in chronological order based on message property dt.
Expects to be instantiated in one execution context (thread, process, etc) and run in another."""
def __init__(self):
Component.__init__(self)
self.sent_count = 0
@@ -262,19 +449,19 @@ class ParallelBuffer(Component):
#data source component ID -> List of messages
self.data_buffer = {}
self.ds_finished_counter = 0
def get_id(self):
return "FEED"
def add_source(self, source_id):
self.data_buffer[source_id] = []
def open(self):
self.pull_socket, self.poller = self.bind_data()
self.feed_socket = self.bind_feed()
self.feed_socket = self.bind_feed()
def do_work(self):
def do_work(self):
# wait for synchronization reply from the host
socks = dict(self.poller.poll(2000)) #timeout after 2 seconds.
@@ -293,18 +480,18 @@ class ParallelBuffer(Component):
def __len__(self):
"""buffer's length is same as internal map holding separate sorted arrays of events keyed by source id"""
return len(self.data_buffer)
return len(self.data_buffer)
def append(self, source_id, value):
"""add an event to the buffer for the source specified by source_id"""
self.data_buffer[source_id].append(value)
self.received_count += 1
def next(self):
"""Get the next message in chronological order"""
if(not(self.is_full() or self.draining)):
return
cur = None
earliest = None
for events in self.data_buffer.values():
@@ -313,58 +500,58 @@ class ParallelBuffer(Component):
cur = events
if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']):
earliest = cur
if(earliest != None):
return earliest.pop(0)
def is_full(self):
"""indicates whether the buffer has messages in buffer for all un-DONE sources"""
for events in self.data_buffer.values():
if (len(events) == 0):
return False
return True
def pending_messages(self):
"""returns the count of all events from all sources in the buffer"""
total = 0
for events in self.data_buffer.values():
total += len(events)
return total
def drain(self):
"""send all messages in the buffer"""
self.draining = True
while(self.pending_messages() > 0):
self.send_next()
def send_next(self):
"""send the (chronologically) next message in the buffer."""
if(not(self.is_full() or self.draining)):
return
event = self.next()
if(event != None):
self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
self.sent_count += 1
self.sent_count += 1
class MergedParallelBuffer(ParallelBuffer):
"""
Merges multiple streams of events into single messages.
"""
def __init__(self):
ParallelBuffer.__init__(self)
def open(self):
self.pull_socket, self.poller = self.bind_merge()
self.feed_socket = self.bind_result()
def next(self):
"""Get the next merged message from the feed buffer."""
if(not(self.is_full() or self.draining)):
return
#get the raw event from the passthrough transform.
result = self.data_buffer["PASSTHROUGH"].pop(0)['value']
for source, events in self.data_buffer.iteritems():
@@ -374,10 +561,10 @@ class MergedParallelBuffer(ParallelBuffer):
cur = events.pop(0)
result[source] = cur['value']
return result
def get_id(self):
return "MERGE"
class BaseTransform(Component):
"""Top level execution entry point for the transform::
@@ -387,8 +574,8 @@ class BaseTransform(Component):
- processes all messages received from feed, until DONE message received
- pushes all transforms
- sends DONE to result socket, closes all sockets and context
Parent class for feed transforms. Subclass and override transform
Parent class for feed transforms. Subclass and override transform
method to create a new derived value from the combined feed."""
def __init__(self, name):
@@ -399,19 +586,19 @@ class BaseTransform(Component):
def get_id(self):
return self.state['name']
def open(self):
def open(self):
"""
Establishes zmq connections.
"""
#create the feed.
"""
#create the feed.
self.feed_socket, self.poller = self.connect_feed()
#create the result PUSH
self.result_socket = self.connect_merge()
def do_work(self):
"""
Loops until feed's DONE message is received:
- receive an event from the data feed
- receive an event from the data feed
- call transform (subclass' method) on event
- send the transformed event
"""
@@ -436,18 +623,18 @@ class BaseTransform(Component):
self.state['value'] = transformed_value
"""
NotImplemented
class PassthroughTransform(BaseTransform):
def __init__(self):
BaseTransform.__init__(self, "PASSTHROUGH")
def transform(self, event):
def transform(self, event):
return {'value':event}
class DataSource(Component):
"""
Baseclass for data sources. Subclass and implement send_all - usually this
Baseclass for data sources. Subclass and implement send_all - usually this
means looping through all records in a store, converting to a dict, and
calling send(map).
"""
@@ -459,8 +646,8 @@ class DataSource(Component):
def get_id(self):
return self.id
def open(self):
#create the data sink. Based on http://zguide.zeromq.org/py:tasksink2
def open(self):
#create the data sink. Based on http://zguide.zeromq.org/py:tasksink2
self.data_socket = self.connect_data()
def send(self, event):
@@ -469,11 +656,11 @@ class DataSource(Component):
sets id and type properties in the dict
sends to the data_socket.
"""
event['id'] = self.id
event['id'] = self.id
event['type'] = self.get_type()
self.data_socket.send(json.dumps(event))
def get_type(self):
raise NotImplemented
zipline/simulator.py
-49
View File
@@ -1,49 +0,0 @@
"""
Provides simulated data feed services...
"""
import multiprocessing
import json
import copy
import threading
import zipline.util as qutil
import zipline.messaging as qmsg
class SimulatorBase(qmsg.ComponentHost):
"""
Simulator coordinates the launch and communication of source, feed, transform, and merge components.
"""
def __init__(self, addresses, gevent_needed=False):
"""
"""
qmsg.ComponentHost.__init__(self, addresses, gevent_needed)
def simulate(self):
self.run()
def get_id(self):
return "Simulator"
class ThreadSimulator(SimulatorBase):
def __init__(self, addresses):
SimulatorBase.__init__(self, addresses)
def launch_component(self, component):
qutil.LOGGER.info("starting {name}".format(name=component.get_id()))
thread = threading.Thread(target=component.run)
thread.start()
return thread
class ProcessSimulator(SimulatorBase):
def __init__(self, addresses):
SimulatorBase.__init__(self, addresses)
def launch_component(self, component):
qutil.LOGGER.info("starting {name}".format(name=component.get_id()))
proc = multiprocessing.Process(target=component.run)
proc.start()
return proc
zipline/sources.py
+13 -15
View File
@@ -2,15 +2,14 @@
Provides data handlers that can push messages to a zipline.core.DataFeed
"""
import datetime
import json
import random
import zipline.util as qutil
import zipline.messaging as qmsg
import zipline.messaging as qmsg
class RandomEquityTrades(qmsg.DataSource):
"""Generates a random stream of trades for testing."""
def __init__(self, sid, source_id, count):
qmsg.DataSource.__init__(self, source_id)
self.count = count
@@ -19,22 +18,21 @@ class RandomEquityTrades(qmsg.DataSource):
self.trade_start = datetime.datetime.now()
self.minute = datetime.timedelta(minutes=1)
self.price = random.uniform(5.0, 50.0)
def get_type(self):
return 'equity_trade'
return 'equity_trade'
def do_work(self):
if(self.incr == self.count):
self.signal_done()
return
self.price = self.price + random.uniform(-0.05, 0.05)
event = {'sid':self.sid,
'dt':qutil.format_date(self.trade_start + (self.minute * self.incr)),
'price':self.price,
'volume':random.randrange(100,10000,100)}
event = {
'sid' : self.sid,
'dt' : qutil.format_date(self.trade_start + (self.minute * self.incr)),
'price' : self.price,
'volume' : random.randrange(100,10000,100)
}
self.send(event)
self.incr += 1
zipline/test/client.py
+6 -10
View File
@@ -3,20 +3,20 @@ import zipline.util as qutil
import zipline.messaging as qmsg
class TestClient(qmsg.Component):
def __init__(self, utest, expected_msg_count=0):
qmsg.Component.__init__(self)
self.received_count = 0
self.expected_msg_count = expected_msg_count
self.utest = utest
self.prev_dt = None
def get_id(self):
return "TEST_CLIENT"
return "TEST_CLIENT"
def open(self):
self.data_feed, self.poller = self.connect_result()
def do_work(self):
socks = dict(self.poller.poll(2000)) #timeout after 2 seconds.
if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:
@@ -28,7 +28,7 @@ class TestClient(qmsg.Component):
"The client should have received ({n}) the same number of messages as the feed sent ({m})."
.format(n=self.received_count, m=self.expected_msg_count))
return
self.received_count += 1
event = json.loads(msg)
if(self.prev_dt != None):
@@ -38,7 +38,3 @@ class TestClient(qmsg.Component):
self.prev_dt = event['dt']
if(self.received_count % 100 == 0):
qutil.LOGGER.info("received {n} messages".format(n=self.received_count))
zipline/test/test_messaging.py
+30 -42
View File
@@ -1,58 +1,47 @@
"""
Test suite for the messaging infrastructure of QSim.
"""
#don't worry about excessive public methods pylint: disable=R0904
#don't worry about excessive public methods pylint: disable=R0904
import unittest2 as unittest
import multiprocessing
import time
# TODO: make sure this can run in parallel... right now this is
# forbiddeen because we hardcode the ports but it should totally
# be possible and then we can the suite much faster.
#
# nosetests --processes=5
import zipline.messaging as qmsg
from zipline.simulator import ThreadSimulator, ProcessSimulator
from zipline.transforms.technical import MovingAverage
from zipline.sources import RandomEquityTrades
import zipline.util as qutil
import zipline.messaging as qmsg
from zipline.test.client import TestClient
qutil.configure_logging()
class SimulatorTestCase(unittest.TestCase):
"""Tests the message passing: datasources -> feed -> transforms -> merge -> client"""
# Should not inherit form TestCase since test runners will pick
# it up as a test.
class SimulatorTestCase(object):
def setUp(self):
"""generate some config objects for the datafeed, sources, and transforms."""
self.addresses = {'sync_address' : "tcp://127.0.0.1:10100",
'data_address' : "tcp://127.0.0.1:10101",
'feed_address' : "tcp://127.0.0.1:10102",
'merge_address' : "tcp://127.0.0.1:10103",
'result_address' : "tcp://127.0.0.1:10104",
'order_address' : "tcp://127.0.0.1:10105"
}
def get_simulator(self):
return ThreadSimulator(self.addresses)
"""
Return the simulator instance to be tested.
"""
raise NotImplementedError
def test_sources_only(self):
"""streams events from two data sources, no transforms."""
sim = self.get_simulator()
ret1 = RandomEquityTrades(133, "ret1", 400)
ret2 = RandomEquityTrades(134, "ret2", 400)
client = TestClient(self, expected_msg_count=800)
sim.register_components([ret1, ret2, client])
sim.simulate()
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=sim.feed.pending_messages()))
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=sim.feed.pending_messages()))
def test_transforms(self):
"""
2 datasources -> feed -> 2 moving average transforms -> transform merge -> testclient
verify message count at client.
"""
sim = self.get_simulator()
ret1 = RandomEquityTrades(133, "ret1", 5000)
ret2 = RandomEquityTrades(134, "ret2", 5000)
@@ -61,10 +50,12 @@ class SimulatorTestCase(unittest.TestCase):
client = TestClient(self, expected_msg_count=10000)
sim.register_components([ret1, ret2, mavg1, mavg2, client])
sim.simulate()
self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages.")
self.assertEqual(sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages.")
def dtest_error_in_feed(self):
ret1 = RandomEquityTrades(133, "ret1", 400)
ret2 = RandomEquityTrades(134, "ret2", 400)
sources = {"ret1":ret1, "ret2":ret2}
@@ -73,12 +64,9 @@ class SimulatorTestCase(unittest.TestCase):
transforms = {"mavg1":mavg1, "mavg2":mavg2}
client = TestClient(self, expected_msg_count=0)
sim = self.get_simulator(sources, transforms, client)
# TODO: way too long
sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator"))
sim.simulate()
class ProcessSimulatorTestCase(SimulatorTestCase):
def get_simulator(self):
return ProcessSimulator(self.addresses)
zipline/test/test_sanity.py
+7
View File
@@ -0,0 +1,7 @@
from unittest2 import TestCase
class TestEnviroment(TestCase):
def test_universe(self):
# first order logic is working today. Yay!
self.assertTrue(True != False)