tests - cleanups

tests/fixtures/schema-discriminated-union-warning.yaml

@@ -12,6 +12,7 @@ components:
       properties:
         object_type:
           type: string
+          enum: ["a"]
         a:
           type: string
 
@@ -22,7 +23,7 @@ components:
       properties:
         object_type:
           type: string
-          enum: ["f"]
+          enum: ["b"]
         b:
           type: string
 
@@ -33,7 +34,7 @@ components:
       properties:
         object_type:
           type: string
-#          const: "f"
+          enum: ["c"]
         c:
           type: string
 

tests/schema_test.py

@@ -369,24 +369,30 @@ def test_schema_discriminated_union_invalid_array(with_schema_discriminated_unio
 def test_schema_discriminated_union_warnings(with_schema_discriminated_union_warning, openapi_version):
     from aiopenapi3.errors import DiscriminatorWarning
 
-    with (
-        pytest.warns(
-            DiscriminatorWarning,
-            match=r"Discriminated Union member key property enum mismatches property mapping \S+ \!= \S+",
-        ),
-        pytest.warns(DiscriminatorWarning, match=r"Discriminated Union member \S+ without const/enum key property \S+"),
+    s = copy.deepcopy(with_schema_discriminated_union_warning)
+    api = OpenAPI("/", s)
+
+    with pytest.warns(
+        DiscriminatorWarning,
+        match=r"Discriminated Union member key property enum mismatches property mapping \S+ \!= \S+",
     ):
-        api = OpenAPI("/", with_schema_discriminated_union_warning)
+        s = copy.deepcopy(with_schema_discriminated_union_warning)
+        s["components"]["schemas"]["B"]["properties"]["object_type"]["enum"] = ["f"]
+        api = OpenAPI("/", s)
 
-    if (openapi_version.major, openapi_version.minor, openapi_version.patch) >= (3, 1, 0):
+    with pytest.warns(
+        DiscriminatorWarning, match=r"Discriminated Union member \S+ without const/enum key property \S+"
+    ):
         s = copy.deepcopy(with_schema_discriminated_union_warning)
         del s["components"]["schemas"]["B"]["properties"]["object_type"]["enum"]
-        s["components"]["schemas"]["B"]["properties"]["object_type"]["enum"] = ["f"]
-        s["components"]["schemas"]["A"]["properties"]["object_type"]["enum"] = ["a"]
-        s["components"]["schemas"]["C"]["properties"]["object_type"]["const"] = "c"
+        api = OpenAPI("/", s)
+
+    if (openapi_version.major, openapi_version.minor, openapi_version.patch) >= (3, 1, 0):
+        s = copy.deepcopy(with_schema_discriminated_union_warning)
+        s["components"]["schemas"]["C"]["properties"]["object_type"]["const"] = "f"
         with pytest.warns(
             DiscriminatorWarning,
-            match=r"Discriminated Union member key property enum mismatches property mapping \S+ \!= \S+",
+            match=r"Discriminated Union member key property const mismatches property mapping \S+ \!= \S+",
         ):
             api = OpenAPI("/", s)
 

tests/tls_test.py

@@ -82,13 +82,28 @@ async def server(config):
     try:
         sd = asyncio.Event()
         task = event_loop.create_task(serve(app, config, shutdown_trigger=sd.wait))
-        await asyncio.sleep(1)
         yield config
     finally:
         sd.set()
         await task
 
 
+@pytest_asyncio.fixture(loop_scope="session")
+async def wait_for_server(server):
+    for i in range(10):
+        try:
+            host, _, port = server.bind[0].rpartition(":")
+            r, w = await asyncio.open_connection(host=host, port=port)
+        except Exception as e:
+            await asyncio.sleep(0.1)
+        else:
+            await w.drain()
+            w.close()
+            await w.wait_closed()
+            break
+    return server
+
+
 @pytest.fixture(scope="session")
 def event_loop_policy():
     return uvloop.EventLoopPolicy()
@@ -121,7 +136,7 @@ def parsed(self, ctx: "Document.Context") -> "Document.Context":
 
 
 @pytest_asyncio.fixture(loop_scope="session")
-async def client(server, certs):
+async def client(server, certs, wait_for_server):
     def self_signed(*args, **kwargs) -> httpx.AsyncClient:
         ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certs["org"]["issuer"])
         if (cert := kwargs.get("cert", None)) is not None:
@@ -197,7 +212,7 @@ async def test_tls_optional(server, client, certs):
 
 
 @pytest.mark.asyncio(loop_scope="session")
-async def test_sync(server, certs):
+async def test_sync(server, certs, wait_for_server):
     def self_signed_(*args, **kwargs) -> httpx.Client:
         ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certs["org"]["issuer"])
         if (cert := kwargs.get("cert", None)) is not None: