Raise RuntimeError if watcher.cancel() is called after client.close()

aetcd/client.py

@@ -644,12 +644,16 @@ async def response_callback(response):
         canceled = asyncio.Event()
 
         async def cancel():
+            if not self._watcher:
+                raise RuntimeError('Calling watcher.cancel() on a closed client')
             canceled.set()
             await events.put(None)
             await self._watcher.cancel(watcher_callback.watch_id)
 
         @_handle_errors
         async def iterator():
+            if not self._watcher:
+                raise RuntimeError('Using watcher as iterator on a closed client')
             while not canceled.is_set():
                 event = await events.get()
                 if event is None:

tests/integration/test_watch.py

@@ -234,3 +234,16 @@ async def test_watch_key_ignores_global_timeout(client, etcdctl_put):
             break
 
         await w.cancel()
+
+
+@pytest.mark.asyncio
+async def test_watch_for_closed_client(etcd, etcdctl_put):
+    w = await etcd.watch(b'key')
+    etcdctl_put('key', '1')
+    await etcd.close()
+    etcdctl_put('key', '2')
+    with pytest.raises(RuntimeError):
+        async for kv in w:
+            raise AssertionError('non-empty watcher iterator with closed client')
+    with pytest.raises(RuntimeError):
+        await w.cancel()