Fix ClusterMetadata request_update() (#1056)

* Add tests for cluster request_update()

* Removed @run_until_complete.

* Wrap async test in class

* See if this makes loop available.

* Just block on Python future.

* 🤦

* Explicitly update.

* Improved request_update tests.

* Fixed ClusterMetadata future usage.

* Simplified test.

* black fml.

aiokafka/cluster.py

@@ -209,7 +209,7 @@ def failed_update(self, exception):
                 f = self._future
                 self._future = None
         if f:
-            f.failure(exception)
+            f.set_exception(exception)
         self._last_refresh_ms = time.time() * 1000
 
     def update_metadata(self, metadata):
@@ -307,7 +307,7 @@ def update_metadata(self, metadata):
         self._last_successful_refresh_ms = now
 
         if f:
-            f.success(self)
+            f.set_result(self)
         log.debug("Updated cluster metadata to %s", self)
 
         for listener in self._listeners:

tests/test_cluster.py

@@ -19,3 +19,23 @@ def test_empty_broker_list():
         )
     )
     assert len(cluster.brokers()) == 2
+
+
+def test_request_update_expecting_success():
+    cluster = ClusterMetadata()
+    updated_cluster = cluster.request_update()
+    cluster.update_metadata(
+        MetadataResponse[0]([(0, "foo", 12), (1, "bar", 34)], []),
+    )
+    assert updated_cluster.result() == cluster
+
+
+def test_request_update_expecting_failure():
+    cluster = ClusterMetadata()
+    updated_cluster = cluster.request_update()
+    test_metadata = MetadataResponse[0](
+        [],  # empty brokers
+        [(17, "foo", []), (17, "bar", [])],  # topics w/ error
+    )
+    cluster.update_metadata(test_metadata)
+    assert updated_cluster.exception() is not None