Support _request_timeout in dynamic client watch()

kubernetes/base/dynamic/client.py

@@ -158,7 +158,7 @@ def server_side_apply(self, resource, body=None, name=None, namespace=None, forc
 
         return self.request('patch', path, body=body, force_conflicts=force_conflicts, **kwargs)
 
-    def watch(self, resource, namespace=None, name=None, label_selector=None, field_selector=None, resource_version=None, timeout=None, watcher=None, allow_watch_bookmarks=None):
+    def watch(self, resource, namespace=None, name=None, label_selector=None, field_selector=None, resource_version=None, timeout=None, watcher=None, allow_watch_bookmarks=None, _request_timeout=None):
         """
         Stream events for a resource from the Kubernetes API
 
@@ -172,6 +172,8 @@ def watch(self, resource, namespace=None, name=None, label_selector=None, field_
         :param timeout: The amount of time in seconds to wait before terminating the stream
         :param watcher: The Watcher object that will be used to stream the resource
         :param allow_watch_bookmarks: Ask the API server to send BOOKMARK events
+        :param _request_timeout: The socket-level timeout for a single request,
+                                 as a float or a (connect, read) tuple
 
         :return: Event object with these keys:
                    'type': The type of event such as "ADDED", "DELETED", etc.
@@ -204,6 +206,7 @@ def watch(self, resource, namespace=None, name=None, label_selector=None, field_
             serialize=False,
             timeout_seconds=timeout,
             allow_watch_bookmarks=allow_watch_bookmarks,
+            _request_timeout=_request_timeout,
         ):
             event['object'] = ResourceInstance(resource, event['object'])
             yield event

kubernetes/base/dynamic/client_test.py

@@ -208,6 +208,31 @@ def log_message(self, format, *args):
             target.server_close()
             proxy.server_close()
 
+    def test_watch_forwards_request_timeout(self):
+        class FakeWatcher:
+            def __init__(self):
+                self.kwargs = None
+
+            def stream(self, func, **kwargs):
+                self.kwargs = kwargs
+                return iter(())
+
+        class FakeResource:
+            def get(self, **kwargs):
+                pass
+
+        dynamic = DynamicClient.__new__(DynamicClient)
+        watcher = FakeWatcher()
+
+        list(dynamic.watch(
+            FakeResource(),
+            namespace='default',
+            watcher=watcher,
+            _request_timeout=30,
+        ))
+
+        self.assertEqual(30, watcher.kwargs['_request_timeout'])
+
 
 if __name__ == '__main__':
     unittest.main()