Merge pull request #89 from tomplus/feat/stream_res_version

fix: watch.stream stores resource_version for the next call

Author: tomplus

kubernetes_asyncio/watch/watch.py

@@ -181,6 +181,8 @@ def stream(self, func, *args, **kwargs):
         self.return_type = self.get_return_type(func)
         kwargs['watch'] = True
         kwargs['_preload_content'] = False
+        if 'resource_version' in kwargs:
+            self.resource_version = kwargs['resource_version']
 
         self.func = partial(func, *args, **kwargs)
 

kubernetes_asyncio/watch/watch_test.py

@@ -67,6 +67,9 @@ async def test_watch_with_decode(self):
             _preload_content=False, watch=True, resource_version='123')
         fake_resp.release.assert_called_once_with()
 
+        # last resource_version has to be stored in the object
+        self.assertEqual(watch.resource_version, '2')
+
     async def test_watch_k8s_empty_response(self):
         """Stop the iterator when the response is empty.
 
@@ -190,6 +193,32 @@ async def test_watch_timeout(self):
              call(_preload_content=False, watch=True, resource_version='1555')])
         fake_resp.release.assert_called_once_with()
 
+    async def test_watch_timeout_with_resource_version(self):
+        fake_resp = CoroutineMock()
+        fake_resp.content.readline = CoroutineMock()
+        fake_resp.release = Mock()
+
+        fake_resp.content.readline.side_effect = [asyncio.TimeoutError(),
+                                                  b""]
+
+        fake_api = Mock()
+        fake_api.get_namespaces = CoroutineMock(return_value=fake_resp)
+        fake_api.get_namespaces.__doc__ = ':return: V1NamespaceList'
+
+        watch = kubernetes_asyncio.watch.Watch()
+        async with watch.stream(fake_api.get_namespaces, resource_version='10') as stream:
+            async for e in stream: # noqa
+                pass
+
+        # all calls use the passed resource version
+        fake_api.get_namespaces.assert_has_calls(
+            [call(_preload_content=False, watch=True, resource_version='10'),
+             call(_preload_content=False, watch=True, resource_version='10')])
+
+        fake_resp.release.assert_called_once_with()
+        self.assertEqual(watch.resource_version, '10')
+
+
 if __name__ == '__main__':
     import asynctest
     asynctest.main()