add more unittests

app/tests/test_async_stream.py

@@ -129,6 +129,7 @@ async def test_read():
         return 0.01  # async wait of 0.01 
     cnt = 0
     ifc =  AsyncStreamClient(reader, writer, None, closed)
+    ifc.proc_max = 0
     ifc.prot_set_timeout_cb(timeout)
     ifc.rx_set_cb(app_read)
     await ifc.client_loop('')
@@ -160,14 +161,21 @@ async def test_write():
         nonlocal ifc
         ifc.close()  # clears the closed callback
         cnt += 1
+    def app_read():
+        nonlocal ifc
+        ifc.proc_start -= 3
+        return 0.01  # async wait of 0.01 
     
     cnt = 0
     ifc =  AsyncStreamClient(reader, writer, None, closed)
+    ifc.proc_max = 10
     ifc.prot_set_timeout_cb(timeout)
+    ifc.rx_set_cb(app_read)
     ifc.tx_add(b'test-data-resp')
     assert 14 == ifc.tx_len()
     await ifc.client_loop('')
     print('End loop')
+    assert ifc.proc_max >= 3
     assert 13 == ifc.rx_len()
     assert 0 == ifc.tx_len()
     assert cnt == 1
@@ -307,8 +315,11 @@ class TestType():
     FWD_SW_EXCPT = 2
     FWD_TIMEOUT = 3
     FWD_OS_ERROR = 4
+    FWD_OS_ERROR_NO_STREAM = 5
+    FWD_RUNTIME_ERROR = 6
+    FWD_RUNTIME_ERROR_NO_STREAM = 7
 
-def create_remote(remote, test_type):
+def create_remote(remote, test_type, with_close_hdr:bool = False):
     def update_hdr(buf):
         return
     def callback():
@@ -318,9 +329,24 @@ def create_remote(remote, test_type):
             raise TimeoutError
         elif test_type == TestType.FWD_OS_ERROR:
             raise ConnectionRefusedError
+        elif test_type == TestType.FWD_OS_ERROR_NO_STREAM:
+            remote.stream = None
+            raise ConnectionRefusedError
+        elif test_type == TestType.FWD_RUNTIME_ERROR:
+            raise RuntimeError("Peer closed")
+        elif test_type == TestType.FWD_RUNTIME_ERROR_NO_STREAM:
+            remote.stream = None
+            raise RuntimeError("Peer closed")
 
+    def close():
+        return
+    if with_close_hdr:
+        close_hndl = close
+    else:
+        close_hndl = None
+    
     remote.ifc = AsyncStreamClient(
-        FakeReader(), FakeWriter(), StreamPtr(None), None)
+        FakeReader(), FakeWriter(), StreamPtr(None), close_hndl)
     remote.ifc.prot_set_update_header_cb(update_hdr)
     remote.ifc.prot_set_init_new_client_conn_cb(callback)
     remote.stream = FakeProto(False)
@@ -332,8 +358,44 @@ async def test_forward():
     cnt = 0
 
     async def _create_remote():
-        nonlocal cnt, remote
+        nonlocal cnt, remote, ifc
         create_remote(remote, TestType.FWD_NO_EXCPT)
+        ifc.fwd_add(b'test-forward_msg2 ')
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_with_conn():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote, ifc
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    create_remote(remote, TestType.FWD_NO_EXCPT)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 0
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_no_conn():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt
         cnt += 1
     
     cnt = 0
@@ -378,3 +440,93 @@ async def test_forward_os_error():
     await ifc.server_loop()
     assert cnt == 1
     del ifc
+
+@pytest.mark.asyncio
+async def test_forward_os_error2():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote
+        create_remote(remote, TestType.FWD_OS_ERROR, True)
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_os_error3():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote
+        create_remote(remote, TestType.FWD_OS_ERROR_NO_STREAM)
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_runtime_error():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote
+        create_remote(remote, TestType.FWD_RUNTIME_ERROR)
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_runtime_error2():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote
+        create_remote(remote, TestType.FWD_RUNTIME_ERROR, True)
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc
+
+@pytest.mark.asyncio
+async def test_forward_runtime_error3():
+    assert asyncio.get_running_loop()
+    remote = StreamPtr(None)
+    cnt = 0
+
+    async def _create_remote():
+        nonlocal cnt, remote
+        create_remote(remote, TestType.FWD_RUNTIME_ERROR_NO_STREAM, True)
+        cnt += 1
+    
+    cnt = 0
+    ifc =  AsyncStreamServer(fake_reader_fwd(), FakeWriter(), None, _create_remote, remote)
+    ifc.fwd_add(b'test-forward_msg')
+    await ifc.server_loop()
+    assert cnt == 1
+    del ifc