test: add unit test on watch request

kr8s/tests/test_api.py

@@ -2,8 +2,10 @@
 # SPDX-License-Identifier: BSD 3-Clause License
 import queue
 import threading
-
+import unittest.mock
+import contextlib
 import anyio
+import asyncio
 import pytest
 
 import kr8s
@@ -314,6 +316,55 @@ async def test_api_timeout() -> None:
         await api.version()
 
 
+async def test_api_gone(spec):
+    """Ensure that the 410 Gone response is handled by restarting a watch request."""
+
+    async def watch_pod(pod):
+        """Start a watch request on the pod."""
+        async for _, obj in pod.async_watch():
+            assert obj.raw['kind'] == Pod.kind
+
+    async def counter(mock_async_get_kind):
+        """Check every second how many call are made on the `mock_async_get_kind` method."""
+        while True:
+            if mock_async_get_kind.call_count >= 2:
+                # If two calls have been made, this means that the watch request has been successfully restarted.
+                return
+            await asyncio.sleep(1)
+
+    class MockResponse:
+        async def aiter_lines(self):
+            for _ in range(3):
+                yield '{"type":"ERROR","object":{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"too old resource version: 1 (585682157)","reason":"Expired","code":410}}'
+
+    @contextlib.asynccontextmanager
+    async def mock_async_get_kind(*args, **kwargs):
+        await asyncio.sleep(3)
+        yield Pod, MockResponse()
+
+    pod = await Pod(spec)
+    await pod.create()
+
+    try:
+        with unittest.mock.patch.object(pod.api, "async_get_kind", wraps=mock_async_get_kind) as mock:
+            async with asyncio.TaskGroup() as tg:
+                watch_task = tg.create_task(watch_pod(pod))
+                counter_task = tg.create_task(counter(mock))
+                counter_task.add_done_callback(lambda task: watch_task.cancel())
+
+            try:
+                watch_task.result()
+                assert False, "A problem occurs during the watch, because it has been stopped."
+            except asyncio.CancelledError:
+                # This is the normal behavior if the counter task has been ended,
+                # all other exceptions must be raised
+                pass
+            finally:
+                counter_task.cancel()
+    finally:
+        await pod.delete()
+
+
 async def test_lookup_kind():
     api = await kr8s.asyncio.api()