add tests for granian lock on windows platform using multiple workers

Author: izmmisha

tests/conftest.py

@@ -10,7 +10,7 @@
 
 
 @asynccontextmanager
-async def _server(interface, port, threading_mode, tls=False):
+async def _server(interface, port, threading_mode, tls=False, workers=1):
     certs_path = Path.cwd() / "tests" / "fixtures" / "tls"
     tls_opts = (
         f"--ssl-certificate {certs_path / 'cert.pem'} "
@@ -20,6 +20,7 @@ async def _server(interface, port, threading_mode, tls=False):
         "".join([
             f"granian --interface {interface} --port {port} ",
             f"--threads 1 --threading-mode {threading_mode} ",
+            f"--workers {workers} ",
             tls_opts,
             f"tests.apps.{interface}:app"
         ]),

tests/test_lock.py

@@ -0,0 +1,30 @@
+import httpx
+import pytest
+from concurrent.futures import ProcessPoolExecutor
+import asyncio
+
+def make_req(port):
+    return httpx.post(f"http://localhost:{port}/echo", data="test")
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("threading_mode", ["runtime", "workers"])
+@pytest.mark.parametrize("server_mode", ["wsgi", "asgi", "rsgi"])
+async def test_lock(wsgi_server, asgi_server, rsgi_server, threading_mode, server_mode):
+    server = {"wsgi": wsgi_server,
+              "asgi": asgi_server,
+              "rsgi": rsgi_server,
+              }[server_mode]
+    async with server(threading_mode, workers=2) as port:
+        with ProcessPoolExecutor() as executor:
+            for _ in range(100):
+                futures = [executor.submit(make_req, port) for _ in range(3)]
+                ok = 0
+                timeout = 0
+                for future in futures:
+                    try:
+                        res = await asyncio.wrap_future(future)
+                        assert res.status_code == 200
+                        ok += 1
+                    except httpx.ReadTimeout:
+                        timeout += 1
+                assert (timeout, ok) == (0, 3)