fix watching with a specified resource version

The watch code reset the version to the last found in the
response.
When you first list existing objects and then start watching from that
resource version the existing versions are older than the version you
wanted and the watch starts from the wrong version after the first
restart.
This leads to for example already deleted objects ending in the stream
again.

Fix this by setting the minimum resource version to reset from to the
input resource version. As long as k8s returns all objects in order in
the watch this should work.
We cannot use the integer value of the resource version to order it as
one should be treat the value as opaque.

Closes kubernetes-client/python#700

watch/watch.py

@@ -122,6 +122,8 @@ def stream(self, func, *args, **kwargs):
         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']
 
         timeouts = ('timeout_seconds' in kwargs)
         while True:

watch/watch_test.py

@@ -14,9 +14,10 @@
 
 import unittest
 
-from mock import Mock
+from mock import Mock, call
 
 from .watch import Watch
+callcount = 0
 
 
 class WatchTests(unittest.TestCase):
@@ -62,6 +63,68 @@ def test_watch_with_decode(self):
         fake_resp.close.assert_called_once()
         fake_resp.release_conn.assert_called_once()
 
+    def test_watch_resource_version_set(self):
+        # gh-700 ensure watching from a resource version does reset to resource
+        # version 0 after k8s resets the watch connection
+        fake_resp = Mock()
+        fake_resp.close = Mock()
+        fake_resp.release_conn = Mock()
+        values = [
+            '{"type": "ADDED", "object": {"metadata": {"name": "test1",'
+            '"resourceVersion": "1"}, "spec": {}, "status": {}}}\n',
+            '{"type": "ADDED", "object": {"metadata": {"name": "test2",'
+            '"resourceVersion": "2"}, "spec": {}, "sta',
+            'tus": {}}}\n'
+            '{"type": "ADDED", "object": {"metadata": {"name": "test3",'
+            '"resourceVersion": "3"}, "spec": {}, "status": {}}}\n'
+        ]
+        # return nothing on the first call and values on the second
+        # this emulates a watch from a rv that returns nothing in the first k8s
+        # watch reset and values later
+        def get_values(*args, **kwargs):
+            global callcount
+            callcount += 1
+            if callcount == 1:
+                return []
+            else:
+                return values
+
+        fake_resp.read_chunked = Mock(
+            side_effect=get_values)
+
+        fake_api = Mock()
+        fake_api.get_namespaces = Mock(return_value=fake_resp)
+        fake_api.get_namespaces.__doc__ = ':return: V1NamespaceList'
+
+        w = Watch()
+        count = 1
+        # ensure we keep our requested resource version or the version latest
+        # returned version when the existing versions are older than the
+        # requested version
+        # needed for the list existing objects, then watch from there use case
+        calls = []
+        first = True
+        for e in w.stream(fake_api.get_namespaces, resource_version="5"):
+            count += 1
+            # first call must use the passed rv
+            # following calls the last rv of the returned values
+            if count % 3 == 0:
+                if first:
+                    rv = "5"
+                    first = False
+                else:
+                    # ideally we want 5 here but as rv must be treated as an
+                    # opaque value we cannot interpret it and order it so rely
+                    # on k8s returning the events completely and in order
+                    rv = "3"
+                calls.append(call(_preload_content=False, watch=True,
+                                  resource_version=rv))
+            # returned
+            if count == len(values) * 3:
+                w.stop()
+
+        fake_api.get_namespaces.assert_has_calls(calls)
+
     def test_watch_stream_twice(self):
         w = Watch(float)
         for step in ['first', 'second']: