Create the shared test fixtures once per pytest run

docs/testing.md

@@ -107,6 +107,14 @@ The test session hooks upload fixture data to S3 and create and remove Athena/Gl
 Selecting a single test under `tests/pyathena/` still invokes the session setup, even when the test itself uses only mocks or pure Python logic.
 Do not assume that `-k` or a single test path makes this suite offline.
 
+The tables and views from `tests/pyathena/tables.py` and the data files they read are in a fixture schema, `ENV.fixture_schema`, which tests only read.
+With pytest-xdist, the controller creates it once before the workers start and drops it after they finish.
+A run without workers creates and drops the fixture schema itself.
+The session hooks run only when a path given to pytest is `tests/pyathena/` or inside it, as with `just test pyathena`.
+Each test process also creates its own schema, `ENV.schema`, where tests create their tables and views.
+The fixture schema is the default schema of the cursor and engine fixtures, so a test that creates an object qualifies its name with `ENV.schema`.
+At the end of the run, the process that created the fixture schema lists its tables and fails the run if they differ from the ones it created.
+
 Run the suites relevant to the change:
 
 ```bash

scripts/sweep_databases.py

@@ -71,7 +71,7 @@ def _eligible(database: dict[str, Any], cutoff: datetime) -> bool:
 def sweep_databases(client: Any, catalog_id: str, *, dry_run: bool = True) -> dict[str, int]:
     """Preview or delete test databases older than seven days.
 
-    Fixtures generate fresh database names for each session or worker.
+    Fixtures generate fresh database names for each pytest run and each test process.
     Databases younger than seven days are retained, including concurrent CI runs.
     Only Glue metadata is deleted; S3 objects and child catalogs are untouched.
     """

tests/__init__.py

@@ -12,6 +12,15 @@
 )
 
 
+def _random_schema():
+    """Return a new test schema name, the shape scripts/sweep_databases.py sweeps.
+
+    Returns:
+        ``pyathena_test_`` followed by 10 random lowercase letters and digits.
+    """
+    return "pyathena_test_" + "".join(random.choices(string.ascii_lowercase + string.digits, k=10))
+
+
 class Env:
     def __init__(self):
         self.region_name = os.getenv("AWS_DEFAULT_REGION")
@@ -31,19 +40,29 @@ def __init__(self):
         )
         self.default_work_group = os.getenv("AWS_ATHENA_DEFAULT_WORKGROUP", "primary")
         self.managed_work_group = os.getenv("AWS_ATHENA_MANAGED_WORKGROUP")
-        self.schema = "pyathena_test_" + "".join(
-            random.choices(string.ascii_lowercase + string.digits, k=10)
-        )
+        # Each test process creates the objects of its tests in its own schema.
+        self.schema = _random_schema()
+        # The read-only tables and views from tests/pyathena/tables.py, and the
+        # data files they read, are in their own schema. With pytest-xdist, the
+        # controller creates it once and passes its name to the workers, which
+        # replace this value in pytest_configure (tests/pyathena/conftest.py).
+        self.fixture_schema = _random_schema()
         # Optional Amazon S3 Tables configuration. `s3tables_catalog` is the
         # registered table-bucket catalog, e.g. "s3tablescatalog/<table-bucket>".
-        # The test session creates its own namespace in it, named like the schema,
+        # Each test process creates its own namespace in it, named like the schema,
         # because a table dropped during a listing of a shared namespace fails
         # the listing. The S3 Tables tests skip when the catalog is unset.
         self.s3tables_catalog = os.getenv("AWS_ATHENA_S3_TABLES_CATALOG")
         self.s3tables_namespace = self.schema if self.s3tables_catalog else None
-        self.s3_filesystem_test_file_key = (
-            f"{self.s3_staging_key}{self.schema}/filesystem/test_read/test.dat"
-        )
+
+    @property
+    def s3_filesystem_test_file_key(self):
+        """The S3 key of the read-only file the filesystem tests read.
+
+        Returns:
+            The key, under the fixture schema's prefix in the staging directory.
+        """
+        return f"{self.s3_staging_key}{self.fixture_schema}/filesystem/test_read/test.dat"
 
 
 ENV = Env()

tests/pyathena/aio/arrow/test_cursor.py

@@ -80,7 +80,7 @@ async def test_no_result_set_raises(self, aio_arrow_cursor):
     async def test_context_manager(self):
         from pyathena.aio.arrow.cursor import AioArrowCursor
 
-        conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioArrowCursor)
+        conn = await _aio_connect(schema_name=ENV.fixture_schema, cursor_class=AioArrowCursor)
         try:
             async with conn.cursor() as cursor:
                 await cursor.execute("SELECT * FROM one_row")

tests/pyathena/aio/conftest.py

@@ -20,11 +20,21 @@ async def _aio_connect(schema_name="default", **kwargs):
 
 @pytest.fixture
 async def aio_cursor(request):
+    """Yield an ``AioCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.cursor import AioCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -34,11 +44,21 @@ async def aio_cursor(request):
 
 @pytest.fixture
 async def aio_dict_cursor(request):
+    """Yield an ``AioDictCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.cursor import AioDictCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioDictCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioDictCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -48,11 +68,21 @@ async def aio_dict_cursor(request):
 
 @pytest.fixture
 async def aio_pandas_cursor(request):
+    """Yield an ``AioPandasCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.pandas.cursor import AioPandasCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioPandasCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioPandasCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -62,11 +92,21 @@ async def aio_pandas_cursor(request):
 
 @pytest.fixture
 async def aio_arrow_cursor(request):
+    """Yield an ``AioArrowCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.arrow.cursor import AioArrowCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioArrowCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioArrowCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -76,11 +116,21 @@ async def aio_arrow_cursor(request):
 
 @pytest.fixture
 async def aio_polars_cursor(request):
+    """Yield an ``AioPolarsCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.polars.cursor import AioPolarsCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioPolarsCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioPolarsCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -90,11 +140,21 @@ async def aio_polars_cursor(request):
 
 @pytest.fixture
 async def aio_s3fs_cursor(request):
+    """Yield an ``AioS3FSCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     from pyathena.aio.s3fs.cursor import AioS3FSCursor
 
     if not hasattr(request, "param"):
         request.param = {}
-    conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioS3FSCursor, **request.param)
+    conn = await _aio_connect(
+        schema_name=ENV.fixture_schema, cursor_class=AioS3FSCursor, **request.param
+    )
     try:
         async with conn.cursor() as cursor:
             yield cursor
@@ -104,14 +164,22 @@ async def aio_s3fs_cursor(request):
 
 @pytest.fixture
 async def aio_spark_cursor(request):
+    """Yield an ``AioSparkCursor`` whose default schema is the fixture schema.
+
+    Args:
+        request: The fixture request; its optional ``param`` holds connection options.
+
+    Yields:
+        The cursor.
+    """
     import asyncio
 
     from pyathena.aio.spark.cursor import AioSparkCursor
 
     if not hasattr(request, "param"):
         request.param = {}
     conn = await _aio_connect(
-        schema_name=ENV.schema,
+        schema_name=ENV.fixture_schema,
         cursor_class=AioSparkCursor,
         work_group=ENV.spark_work_group,
         **request.param,

tests/pyathena/aio/pandas/test_cursor.py

@@ -71,7 +71,7 @@ async def test_no_result_set_raises(self, aio_pandas_cursor):
     async def test_context_manager(self):
         from pyathena.aio.pandas.cursor import AioPandasCursor
 
-        conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioPandasCursor)
+        conn = await _aio_connect(schema_name=ENV.fixture_schema, cursor_class=AioPandasCursor)
         try:
             async with conn.cursor() as cursor:
                 await cursor.execute("SELECT * FROM one_row")

tests/pyathena/aio/polars/test_cursor.py

@@ -63,7 +63,7 @@ async def test_no_result_set_raises(self, aio_polars_cursor):
     async def test_context_manager(self):
         from pyathena.aio.polars.cursor import AioPolarsCursor
 
-        conn = await _aio_connect(schema_name=ENV.schema, cursor_class=AioPolarsCursor)
+        conn = await _aio_connect(schema_name=ENV.fixture_schema, cursor_class=AioPolarsCursor)
         try:
             async with conn.cursor() as cursor:
                 await cursor.execute("SELECT * FROM one_row")

tests/pyathena/aio/s3fs/test_cursor.py

@@ -71,7 +71,7 @@ async def test_async_iterator(self, aio_s3fs_cursor):
         assert rows == [(1,)]
 
     async def test_context_manager(self):
-        conn = await _aio_connect(schema_name=ENV.schema)
+        conn = await _aio_connect(schema_name=ENV.fixture_schema)
         try:
             async with conn.cursor(AioS3FSCursor) as cursor:
                 await cursor.execute("SELECT * FROM one_row")

tests/pyathena/aio/spark/test_cursor.py

@@ -112,7 +112,7 @@ async def test_spark_dataframe(self, aio_spark_cursor):
                 df = spark.read.format("csv") \\
                     .option("header", "true") \\
                     .option("inferSchema", "true") \\
-                    .load("{ENV.s3_staging_dir}{ENV.schema}/spark_group_by/spark_group_by.csv")
+                    .load("{ENV.s3_staging_dir}{ENV.fixture_schema}/spark_group_by/spark_group_by.csv")
                 """
             ),
             description="test description",
@@ -172,7 +172,7 @@ async def test_spark_sql(self, aio_spark_cursor):
         await aio_spark_cursor.execute(
             textwrap.dedent(
                 f"""
-                spark.sql("SELECT * FROM {ENV.schema}.one_row").show()
+                spark.sql("SELECT * FROM {ENV.fixture_schema}.one_row").show()
                 """
             )
         )
@@ -484,7 +484,7 @@ async def test_executemany(self, aio_spark_cursor):
 
     async def test_context_manager(self):
         conn = await _aio_connect(
-            schema_name=ENV.schema,
+            schema_name=ENV.fixture_schema,
             cursor_class=AioSparkCursor,
             work_group=ENV.spark_work_group,
         )

tests/pyathena/aio/sqlalchemy/test_base.py

@@ -105,7 +105,9 @@ async def test_unicode(self, async_engine):
     async def test_reflect_table(self, async_engine):
         _, conn = async_engine
         one_row = await conn.run_sync(
-            lambda sync_conn: Table("one_row", MetaData(schema=ENV.schema), autoload_with=sync_conn)
+            lambda sync_conn: Table(
+                "one_row", MetaData(schema=ENV.fixture_schema), autoload_with=sync_conn
+            )
         )
         assert len(one_row.c) == 1
         assert one_row.c.number_of_rows is not None
@@ -119,15 +121,15 @@ def _inspect(sync_conn):
             return insp.get_schema_names()
 
         schemas = await conn.run_sync(_inspect)
-        assert ENV.schema in schemas
+        assert ENV.fixture_schema in schemas
         assert "default" in schemas
 
     async def test_get_table_names(self, async_engine):
         _, conn = async_engine
 
         def _inspect(sync_conn):
             insp = sqlalchemy.inspect(sync_conn)
-            return insp.get_table_names(schema=ENV.schema)
+            return insp.get_table_names(schema=ENV.fixture_schema)
 
         table_names = await conn.run_sync(_inspect)
         assert "many_rows" in table_names
@@ -138,9 +140,9 @@ async def test_throttled_reflection_reads_glue(self, async_engine, monkeypatch):
         def reflect(sync_conn):
             insp = sqlalchemy.inspect(sync_conn)
             return (
-                insp.get_table_comment("one_row", schema=ENV.schema),
-                insp.get_table_options("one_row", schema=ENV.schema),
-                insp.get_table_names(schema=ENV.schema),
+                insp.get_table_comment("one_row", schema=ENV.fixture_schema),
+                insp.get_table_options("one_row", schema=ENV.fixture_schema),
+                insp.get_table_names(schema=ENV.fixture_schema),
             )
 
         expected = await conn.run_sync(reflect)
@@ -160,8 +162,8 @@ async def test_has_table(self, async_engine):
         def _inspect(sync_conn):
             insp = sqlalchemy.inspect(sync_conn)
             return (
-                insp.has_table("one_row", schema=ENV.schema),
-                insp.has_table("this_table_does_not_exist", schema=ENV.schema),
+                insp.has_table("one_row", schema=ENV.fixture_schema),
+                insp.has_table("this_table_does_not_exist", schema=ENV.fixture_schema),
             )
 
         exists, not_exists = await conn.run_sync(_inspect)
@@ -173,7 +175,7 @@ async def test_get_columns(self, async_engine):
 
         def _inspect(sync_conn):
             insp = sqlalchemy.inspect(sync_conn)
-            return insp.get_columns(table_name="one_row", schema=ENV.schema)
+            return insp.get_columns(table_name="one_row", schema=ENV.fixture_schema)
 
         columns = await conn.run_sync(_inspect)
         actual = columns[0]

tests/pyathena/aio/test_cursor.py

@@ -44,7 +44,7 @@ async def test_fetchone(self, aio_cursor):
         assert await aio_cursor.fetchone() == (1,)
         assert aio_cursor.rownumber == 1
         assert await aio_cursor.fetchone() is None
-        assert aio_cursor.database == ENV.schema
+        assert aio_cursor.database == ENV.fixture_schema
         assert aio_cursor.catalog
         assert aio_cursor.query_id
         assert aio_cursor.query
@@ -440,7 +440,7 @@ async def test_executemany_fetch(self, aio_cursor):
             await aio_cursor.fetchone()
 
     async def test_context_manager(self):
-        conn = await _aio_connect(schema_name=ENV.schema)
+        conn = await _aio_connect(schema_name=ENV.fixture_schema)
         try:
             async with conn.cursor() as cursor:
                 await cursor.execute("SELECT * FROM one_row")
@@ -525,7 +525,8 @@ async def read():
             return (
                 view(await aio_cursor.get_table_metadata("one_row")),
                 sorted(view(m) for m in await aio_cursor.list_table_metadata()),
-                ENV.schema in [d.name for d in await aio_cursor.list_databases("AwsDataCatalog")],
+                ENV.fixture_schema
+                in [d.name for d in await aio_cursor.list_databases("AwsDataCatalog")],
             )
 
         expected = await read()