fix: set missing loop param for service inits (#436)

Author: Daniel Gellert

faust/app/base.py

@@ -466,7 +466,7 @@ def __init__(
         self._default_options = (id, options)
 
         # The agent manager manages all agents.
-        self.agents = AgentManager(self)
+        self.agents = AgentManager(self, loop=loop)
 
         # Sensors monitor Faust using a standard sensor API.
         self.sensors = SensorDelegate(self)
@@ -1791,7 +1791,7 @@ def _update_assignment(self, assigned: Set[TP]) -> Tuple[Set[TP], Set[TP]]:
         return revoked, newly_assigned
 
     def _new_producer(self) -> ProducerT:
-        return self.transport.create_producer(beacon=self.beacon)
+        return self.transport.create_producer(loop=self.loop, beacon=self.beacon)
 
     def _new_consumer(self) -> ConsumerT:
         return self.transport.create_consumer(

faust/transport/consumer.py

@@ -485,7 +485,7 @@ def __init__(
         self.not_waiting_next_records = Event()
         self.not_waiting_next_records.set()
         self._reset_state()
-        super().__init__(**kwargs)
+        super().__init__(loop=loop, **kwargs)
         self.transactions = self.transport.create_transaction_manager(
             consumer=self,
             producer=self.app.producer,

faust/transport/drivers/aiokafka.py

@@ -1482,7 +1482,7 @@ async def _create_topic(
                 topic,
                 partitions,
                 replication,
-                loop=asyncio.get_event_loop_policy().get_event_loop(),
+                loop=self.loop,
                 **kwargs,
             )
         try:

faust/transport/producer.py

@@ -144,7 +144,7 @@ def __init__(
         self.partitioner = conf.producer_partitioner
         api_version = self._api_version = conf.producer_api_version
         assert api_version is not None
-        super().__init__(**kwargs)
+        super().__init__(loop=loop, **kwargs)
         self.buffer = ProducerBuffer(loop=self.loop, beacon=self.beacon)
         if conf.producer_threaded:
             self.threaded_producer = self.create_threaded_producer()

tests/unit/app/test_base.py

@@ -119,7 +119,7 @@ def test_new_producer(self, *, app):
             autospec=Transport,
         )
         assert app._new_producer() is transport.create_producer.return_value
-        transport.create_producer.assert_called_with(beacon=ANY)
+        transport.create_producer.assert_called_with(loop=app.loop, beacon=ANY)
         assert app.producer is transport.create_producer.return_value
 
     @pytest.mark.parametrize(

tests/unit/stores/test_aerospike.py

@@ -45,8 +45,9 @@ async def test_get_aerospike_client_error(self, aero):
         client_mock.connect = MagicMock(side_effect=Exception)
         faust.stores.aerospike.aerospike_client = None
         config = {"k": "v"}
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             AeroSpikeStore.get_aerospike_client(config)
+            assert exc_info.type == Exception
 
     @pytest.mark.asyncio
     async def test_get_aerospike_client_instantiated(self, aero):
@@ -90,8 +91,9 @@ def test_get_none_value(self, store):
 
     def test_get_exception(self, store):
         store.client.get = MagicMock(side_effect=Exception)
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             store._get(b"test_get")
+            assert exc_info.type == Exception
 
     def test_set_success(
         self,
@@ -119,8 +121,9 @@ def test_set_exception(self, store):
         store.client.put = MagicMock(side_effect=Exception)
         key = b"key"
         value = b"value"
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             store._set(key, value)
+            assert exc_info.type == Exception
 
     def test_persisted_offset(self, store):
         return_value = store.persisted_offset(MagicMock())
@@ -136,15 +139,17 @@ def test_del_success(self, store):
     def test_del_exception(self, store):
         key = b"key"
         store.client.remove = MagicMock(side_effect=Exception)
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             store._del(key)
+            assert exc_info.type == Exception
 
     def test_iterkeys_error(self, store):
         scan = MagicMock()
         store.client.scan = MagicMock(side_effect=Exception)
         scan.results = MagicMock(side_effect=Exception)
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             list(store._iterkeys())
+            assert exc_info.type == Exception
 
     def test_iterkeys_success(self, store):
         scan = MagicMock()
@@ -181,13 +186,15 @@ def test_itervalues_success(self, store):
 
     def test_itervalues_error(self, store):
         store.client.scan = MagicMock(side_effect=Exception)
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             set(store._itervalues())
+            assert exc_info.type == Exception
 
     def test_iteritems_error(self, store):
         store.client.scan = MagicMock(side_effect=Exception)
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             set(store._iteritems())
+            assert exc_info.type == Exception
 
     def test_iteritems_success(self, store):
         with patch("faust.stores.aerospike.aerospike", MagicMock()):
@@ -218,8 +225,9 @@ def test_iteritems_success(self, store):
     def test_contains_error(self, store):
         store.client.exists = MagicMock(side_effect=Exception)
         key = b"key"
-        with pytest.raises(Exception):
+        with pytest.raises(Exception) as exc_info:
             store._contains(key)
+            assert exc_info.type == Exception
 
     def test_contains_does_not_exist(self, store):
         store.client.exists = MagicMock(return_value=(None, None))