Handle error events in watch

Raise an ApiException for error events that indicate a watch failure
despite the HTTP response indicating success.

Fixes #57

Signed-off-by: Fabian Reinartz <freinartz@google.com>

watch/watch.py

@@ -128,7 +128,14 @@ def stream(self, func, *args, **kwargs):
             resp = func(*args, **kwargs)
             try:
                 for line in iter_resp_lines(resp):
-                    yield self.unmarshal_event(line, return_type)
+                    evt = self.unmarshal_event(line, return_type)
+                    if evt['type'] == 'ERROR':
+                        obj = evt['raw_object']
+                        reason = "%s: %s" % (obj['reason'], obj['message'])
+                        raise client.rest.ApiException(status=obj['code'],
+                                                       reason=reason)
+                    yield evt
+
                     if self._stop:
                         break
             finally:

watch/watch_test.py

@@ -16,6 +16,8 @@
 
 from mock import Mock
 
+from kubernetes import client
+
 from .watch import Watch
 
 
@@ -173,6 +175,32 @@ def test_watch_with_exception(self):
         fake_resp.close.assert_called_once()
         fake_resp.release_conn.assert_called_once()
 
+    def test_watch_with_error_event(self):
+        print("start")
+        fake_resp = Mock()
+        fake_resp.close = Mock()
+        fake_resp.release_conn = Mock()
+        fake_resp.read_chunked = Mock(
+            return_value=[
+                '{"type": "ERROR", "object": {"code": 410, '
+                '"reason": "Gone", "message": "error message"}}\n'])
+
+        fake_api = Mock()
+        fake_api.get_thing = Mock(return_value=fake_resp)
+
+        w = Watch()
+        try:
+            for _ in w.stream(fake_api.get_thing):
+                self.fail(self, "Should fail with ApiException.")
+        except client.rest.ApiException:
+            pass
+
+        fake_api.get_thing.assert_called_once_with(
+            _preload_content=False, watch=True)
+        fake_resp.read_chunked.assert_called_once_with(decode_content=False)
+        fake_resp.close.assert_called_once()
+        fake_resp.release_conn.assert_called_once()
+
 
 if __name__ == '__main__':
     unittest.main()