add watch testcases

Author: tg123

src/Watcher.cs

@@ -22,8 +22,10 @@ public enum WatchEventType
 
     public class Watcher<T> : IDisposable
     {
-        private readonly StreamReader _streamReader;
+        public bool Watching { get; private set; }
+
         private readonly CancellationTokenSource _cts;
+        private readonly StreamReader _streamReader;
 
         public Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, Action<Exception> onError)
         {
@@ -37,25 +39,40 @@ public Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, Act
 
             Task.Run(async () =>
             {
-                while (!streamReader.EndOfStream)
+                try
                 {
-                    if (token.IsCancellationRequested)
-                    {
-                        return;
-                    }
+                    Watching = true;
 
-                    try
+                    while (!streamReader.EndOfStream)
                     {
+                        if (token.IsCancellationRequested)
+                        {
+                            return;
+                        }
+
                         var line = await streamReader.ReadLineAsync();
-                        var @event = SafeJsonConvert.DeserializeObject<WatchEvent>(line);
 
-                        OnEvent?.Invoke(@event.Type, @event.Object);
-                    }
-                    catch (Exception e)
-                    {
-                        OnError?.Invoke(e);
+                        try
+                        {
+                            var @event = SafeJsonConvert.DeserializeObject<WatchEvent>(line);
+                            OnEvent?.Invoke(@event.Type, @event.Object);
+                        }
+                        catch (Exception e)
+                        {
+                            // error if deserialized failed or onevent throws
+                            OnError?.Invoke(e);
+                        }
                     }
                 }
+                catch (Exception e)
+                {
+                    // error when transport error, IOException ect
+                    OnError?.Invoke(e);
+                }
+                finally
+                {
+                    Watching = false;
+                }
             }, token);
         }
 

tests/AuthTests.cs

@@ -4,6 +4,7 @@
 using System.Net;
 using System.Net.Http.Headers;
 using System.Security.Cryptography.X509Certificates;
+using System.Threading.Tasks;
 using k8s.Models;
 using k8s.Tests.Mock;
 using Microsoft.AspNetCore.Hosting;
@@ -39,7 +40,7 @@ public void TestAnonymous()
             using (var server = new MockKubeApiServer(cxt =>
             {
                 cxt.Response.StatusCode = (int) HttpStatusCode.Unauthorized;
-                return false;
+                return Task.FromResult(false);
             }))
             {
                 var client = new Kubernetes(new KubernetesClientConfiguration
@@ -69,10 +70,10 @@ public void TestBasicAuth()
                 if (header != expect)
                 {
                     cxt.Response.StatusCode = (int) HttpStatusCode.Unauthorized;
-                    return false;
+                    return Task.FromResult(false);
                 }
 
-                return true;
+                return Task.FromResult(true);
             }))
             {
                 {
@@ -256,10 +257,10 @@ public void TestToken()
                 if (header != expect)
                 {
                     cxt.Response.StatusCode = (int) HttpStatusCode.Unauthorized;
-                    return false;
+                    return Task.FromResult(false);
                 }
 
-                return true;
+                return Task.FromResult(true);
             }))
             {
                 {

tests/Mock/MockKubeApiServer.cs

@@ -20,21 +20,19 @@ public class MockKubeApiServer : IDisposable
 
         private readonly IWebHost _webHost;
 
-        public MockKubeApiServer(Func<HttpContext, bool> shouldNext = null, Action<ListenOptions> listenConfigure = null,
+        public MockKubeApiServer(Func<HttpContext, Task<bool>> shouldNext = null, Action<ListenOptions> listenConfigure = null,
             string resp = MockPodResponse)
         {
-            shouldNext = shouldNext ?? (_ => true);
+            shouldNext = shouldNext ?? (_ => Task.FromResult(true));
             listenConfigure = listenConfigure ?? (_ => { });
 
             _webHost = WebHost.CreateDefaultBuilder()
-                .Configure(app => app.Run(httpContext =>
+                .Configure(app => app.Run(async httpContext =>
                 {
-                    if (shouldNext(httpContext))
+                    if (await shouldNext(httpContext))
                     {
-                        httpContext.Response.WriteAsync(resp);
+                        await httpContext.Response.WriteAsync(resp);
                     }
-
-                    return Task.Delay(0);
                 }))
                 .UseKestrel(options => { options.Listen(IPAddress.Loopback, 0, listenConfigure); })
                 .Build();