Repository: liampauling/flumine
Pull request #335 (2020-11-25 12:02:26)

Flaky flaky integration tests fix
Missing 'on_process' function (now subclassed)


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index c4db406..0251eca 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,17 @@
 Release History
 ---------------
 
+1.14.11 (2020-11-25)
++++++++++++++++++++
+
+**Improvements**
+
+- Flaky flaky integration tests
+
+**Bug Fixes**
+
+- Missing 'on_process' function (now subclassed)
+
 1.14.10 (2020-11-25)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 7b439a1..cac8f74 100644
--- a/flumine/__version__.py
+++ b/flumine/__version__.py
@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.14.10"
+__version__ = "1.14.11"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index b5a394c..8706def 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -1,7 +1,7 @@
 import logging
 import datetime
 from betfairlightweight.streaming import StreamListener, HistoricalGeneratorStream
-from betfairlightweight.streaming.stream import BaseStream as BFLWBaseStream
+from betfairlightweight.streaming.stream import MarketStream
 from betfairlightweight.resources.baseresource import BaseResource
 
 from .basestream import BaseStream
@@ -9,15 +9,13 @@ from .basestream import BaseStream
 logger = logging.getLogger(__name__)
 
 
-class Stream(BFLWBaseStream):
+class FlumineMarketStream(MarketStream):
     """
     Custom bflw stream to speed up processing
     by limiting to inplay/not inplay or limited
     seconds to start.
     """
 
-    _lookup = "mc"
-
     def snap(self, market_ids: list = None) -> list:
         market_books = []
         for cache in list(self._caches.values()):
@@ -58,7 +56,7 @@ class HistoricListener(StreamListener):
 
     def _add_stream(self, unique_id, stream_type):
         if stream_type == "marketSubscription":
-            return Stream(self)
+            return FlumineMarketStream(self)
 
 
 class HistoricalStream(BaseStream):


Test patch:
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 36d5601..63034a6 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -96,6 +96,8 @@ class IntegrationTest(unittest.TestCase):
         framework.add_strategy(market_strategy)
         framework.run()
 
+        self.assertEqual(len(framework.markets), 1)
+
         for market in framework.markets:
             limit_orders = [
                 o for o in market.blotter if o.trade.strategy == limit_strategy
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 7a0adf3..a8aa313 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -448,10 +448,10 @@ class TestHistoricalStream(unittest.TestCase):
         self.assertEqual(generator, mock_generator().get_generator())
 
 
-class TestStream(unittest.TestCase):
+class TestFlumineMarketStream(unittest.TestCase):
     def setUp(self) -> None:
         self.listener = mock.Mock()
-        self.stream = historicalstream.Stream(self.listener)
+        self.stream = historicalstream.FlumineMarketStream(self.listener)
 
     def test_init(self):
         self.assertEqual(self.stream._listener, self.listener)
@@ -459,7 +459,7 @@ class TestStream(unittest.TestCase):
 
     def test_snap_inplay(self):
         # inPlay
-        self.stream = historicalstream.Stream(
+        self.stream = historicalstream.FlumineMarketStream(
             mock.Mock(inplay=True, seconds_to_start=None)
         )
         self.stream._caches = {
@@ -475,7 +475,7 @@ class TestStream(unittest.TestCase):
         }
         self.assertEqual(len(self.stream.snap()), 1)
 
-        self.stream = historicalstream.Stream(
+        self.stream = historicalstream.FlumineMarketStream(
             mock.Mock(inplay=False, seconds_to_start=None)
         )
         self.stream._caches = {
@@ -489,7 +489,7 @@ class TestStream(unittest.TestCase):
 
     def test_snap_seconds_to_start(self):
         # secondsToStart
-        self.stream = historicalstream.Stream(
+        self.stream = historicalstream.FlumineMarketStream(
             mock.Mock(inplay=None, seconds_to_start=600)
         )
         self.stream._caches = {
@@ -527,7 +527,7 @@ class TestHistoricListener(unittest.TestCase):
         self.assertTrue(self.listener.inplay)
         self.assertEqual(self.listener.seconds_to_start, 123)
 
-    @mock.patch("flumine.streams.historicalstream.Stream")
+    @mock.patch("flumine.streams.historicalstream.FlumineMarketStream")
     def test__add_stream(self, mock_stream):
         self.assertEqual(
             self.listener._add_stream(123, "marketSubscription"), mock_stream()