From 88c37d7bd90c1fab019e6f0e16368e9b78d3e251 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Tue, 13 Oct 2026 12:24:17 -0700 Subject: [PATCH 1/5] Clone integration test databases from a migrated template Tests that need the current schema now clone a template database migrated once per migration set instead of running every migration per test. Tests about migrations keep migrating an empty database. Databases and login roles a test creates are dropped when it ends unless WEAVE_TEST_KEEP_DATABASES=1 or WEAVE_TEST_RESOURCE_MANIFEST asks to keep them. Add pytest-xdist for parallel runs and document the fast path. --- CONTRIBUTING.md | 34 ++++ pyproject.toml | 1 + tests/integration/conftest.py | 184 +++++++++++++++--- .../integration/connectors/test_postgresql.py | 4 + tests/integration/test_boot.py | 6 +- tests/integration/test_incident_migration.py | 4 + tests/integration/test_schedule_migration.py | 3 + uv.lock | 24 +++ workers/agentic/uv.lock | 1 + workers/files/uv.lock | 1 + 10 files changed, 229 insertions(+), 33 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4c3c3699..6976d246 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -134,6 +134,40 @@ and must fail when a required dependency is absent. Inspect fixture prerequisites and use distinct resources/evidence paths; do not run against production or reset retained services to make tests pass. +### Run the PostgreSQL integration suite + +The integration fixtures give every test its own database on the server named by +`WEAVE_TEST_DATABASE_URL`: + +- A test that needs the current schema clones a template database, + `weave_b1_template_`. The first test of a run migrates it once; its name + is a hash of the migration files, so changing a migration builds a new template + and drops the old one. The template stays on the server between runs and holds + no test data. +- A test about migrations themselves (the `empty_settings` and + `fresh_migration_settings` fixtures) starts from an empty database and runs the + real migrations. +- When a test ends, the databases and login roles it created are dropped. Set + `WEAVE_TEST_KEEP_DATABASES=1` to keep them for inspection. Setting + `WEAVE_TEST_RESOURCE_MANIFEST` to a file path records each one in that file and + keeps them too. + +Run the suite on four parallel workers (pytest-xdist). Workers share the +template, which an advisory lock lets only one of them build: + +```sh +uv run --locked --all-extras pytest tests/integration -n 4 --dist loadfile +``` + +`--dist loadfile` keeps each test file on one worker, so a file's module fixture +(the Kafka broker, for example) starts once. Without `-n` the suite runs serially. + +A server used only for these tests can skip durability. Start a throwaway +PostgreSQL with `-c fsync=off -c synchronous_commit=off -c full_page_writes=off` +after the image name of its `docker run` command; an unclean stop can corrupt its +data, so never use these settings on a server whose data you keep, such as the +Compose development stack. + [Local runtime setup](docs/reference/local-runtime.md) explains provisioning. End-to-end fixtures exercise installed wheels, admitted images, identities and external-effect receivers; they are acceptance tooling rather than the introductory diff --git a/pyproject.toml b/pyproject.toml index 56449261..2d3fc6d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -92,6 +92,7 @@ dev = [ "types-jsonschema>=4.26.0", "hatchling>=1.28.0", "types-regex>=2026.9.10.20260911", + "pytest-xdist>=3.8.0,<4", ] docs = [ "mkdocs==1.6.1", diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 394b1540..2709911d 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -14,10 +14,18 @@ # Author: Firefly Software Foundation # SPDX-License-Identifier: Apache-2.0 -"""Explicitly guarded, real PostgreSQL fixtures. Databases are retained, never dropped.""" +"""Explicitly guarded, real PostgreSQL fixtures. +Each test gets its own database. Tests that need the current schema clone it from a template database migrated +once per schema; tests that exercise migrations start from an empty database and run them for real. Databases +and login roles a test creates are dropped when it ends, unless WEAVE_TEST_KEEP_DATABASES=1 or +WEAVE_TEST_RESOURCE_MANIFEST asks to keep them. +""" + +import hashlib import os import runpy +import warnings from pathlib import Path from uuid import uuid4 @@ -39,7 +47,24 @@ def validate_release_receipt(release_backends): release_backends["keycloak_endpoint"]() +# Databases and roles created by the running test, dropped at teardown by drop_owned_resources. +_OWNED: list[tuple[str, str]] = [] +# Name of the migrated template database once this process has found or built it. +_TEMPLATE: list[str] = [] +TEMPLATE_LOCK = 7260912902 + + +class TemplateBuildError(Exception): + """Migrating the template database failed; reported as itself, not as an unavailable backend.""" + + +def keep_resources() -> bool: + return os.environ.get("WEAVE_TEST_KEEP_DATABASES") == "1" or bool(os.environ.get("WEAVE_TEST_RESOURCE_MANIFEST")) + + def retain_resource(kind: str, name: str) -> None: + """Record a database or role the test created; it is dropped at teardown unless resources are kept.""" + _OWNED.append((kind, name)) path = os.environ.get("WEAVE_TEST_RESOURCE_MANIFEST") if path: import json @@ -50,6 +75,124 @@ def retain_resource(kind: str, name: str) -> None: handle.write(json.dumps({"kind": kind, "name": name, "observed_at": datetime.now(UTC).isoformat()}) + "\n") +def control_url(): + value = os.environ.get("WEAVE_TEST_DATABASE_URL") + if not value: + pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL using the task-owned Compose backend") + url = make_url(value) + if ( + url.drivername != "postgresql+asyncpg" + or url.host not in {"127.0.0.1", "localhost"} + or url.username != "weave_b1_owner" + or url.database != "weave_b1_control" + or url.port in {None, 5432, 5434} + ): + pytest.fail("Refusing unguarded database: use the dedicated Weave integration backend") + return url + + +def template_name() -> str: + """Name the migrated template after everything that shapes its schema, so a changed migration rebuilds it.""" + import firefly_weave.persistence.migrations as module + from firefly_weave.contracts.operational_policy import OperationsPolicy + + source = Path(module.__file__).resolve().parents[3] / "migrations" + source = source if source.is_dir() else Path(module.__file__).with_name("alembic") + digest = hashlib.sha256(module.SCHEMA_VERSION.encode()) + for path in sorted(source.rglob("*.py")): + digest.update(path.relative_to(source).as_posix().encode() + b"\0" + path.read_bytes() + b"\0") + digest.update(OperationsPolicy().model_dump_json().encode()) + return "weave_b1_template_" + digest.hexdigest()[:16] + + +async def migrated_template(connection, url) -> str: + """Find or build the migrated template once; the advisory lock makes parallel workers build it only once.""" + if _TEMPLATE: + return _TEMPLATE[0] + from firefly_weave.persistence.migrations import migrate + from firefly_weave.settings import Settings + + name = template_name() + await connection.execute(text("SELECT pg_advisory_lock(:key)"), {"key": TEMPLATE_LOCK}) + try: + ready = await connection.scalar( + text("SELECT datistemplate FROM pg_database WHERE datname=:name"), {"name": name} + ) + if not ready: + # A template is marked only after a complete migration; anything else is a failed build. + await connection.execute(text(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')) + stale = await connection.scalars( + text( + "SELECT datname FROM pg_database WHERE datname LIKE 'weave\\_b1\\_template\\_%' AND datname<>:name" + ), + {"name": name}, + ) + for old in stale.all(): + await connection.execute(text(f'ALTER DATABASE "{old}" IS_TEMPLATE false')) + await connection.execute(text(f'DROP DATABASE IF EXISTS "{old}" WITH (FORCE)')) + await connection.execute(text(f'CREATE DATABASE "{name}"')) + settings = Settings( + scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False) + ) + try: + await migrate(settings) + except Exception as error: + raise TemplateBuildError("Migrating the template database failed") from error + # Clones need a template nobody is connected to. + await connection.execute(text(f'ALTER DATABASE "{name}" IS_TEMPLATE true ALLOW_CONNECTIONS false')) + finally: + await connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": TEMPLATE_LOCK}) + _TEMPLATE.append(name) + return name + + +async def create_database(*, migrated: bool): + from firefly_weave.settings import Settings + + url = control_url() + admin = create_async_engine(url, isolation_level="AUTOCOMMIT") + name = f"weave_b1_test_{uuid4().hex}" + try: + async with admin.connect() as connection: + marker = await connection.scalar(text("SELECT identity FROM weave_test_backend_guard")) + if marker != "firefly-weave-b1-local-integration": + pytest.fail("Refusing database without the task-owned backend marker") + template = f' TEMPLATE "{await migrated_template(connection, url)}"' if migrated else "" + await connection.execute(text(f'CREATE DATABASE "{name}"{template}')) + retain_resource("database", name) + except TemplateBuildError: + raise + except Exception: + pytest.fail("Real task-owned PostgreSQL is unavailable or its backend marker is missing", pytrace=False) + finally: + await admin.dispose() + return Settings(scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False)) + + +@pytest.fixture(autouse=True) +async def drop_owned_resources(): + """Drop what the test created: its databases first, so its roles hold no privileges, then the roles.""" + start = len(_OWNED) + yield + owned = _OWNED[start:] + del _OWNED[start:] + if not owned or keep_resources(): + return + admin = create_async_engine(control_url(), isolation_level="AUTOCOMMIT") + try: + async with admin.connect() as connection: + for kind in ("database", "role"): + for item, name in owned: + if item == kind: + await connection.execute( + text(f'DROP {kind.upper()} IF EXISTS "{name}"' + " WITH (FORCE)" * (kind == "database")) + ) + except Exception as error: + warnings.warn(f"Test resources were not dropped: {type(error).__name__}", stacklevel=1) + finally: + await admin.dispose() + + @pytest.fixture def services(): """Resolve transaction tests through the production service scan without owning a pool.""" @@ -98,44 +241,25 @@ def snapshot(self): @pytest.fixture async def empty_settings(): - value = os.environ.get("WEAVE_TEST_DATABASE_URL") - if not value: - pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL using the task-owned Compose backend") - url = make_url(value) - if ( - url.drivername != "postgresql+asyncpg" - or url.host not in {"127.0.0.1", "localhost"} - or url.username != "weave_b1_owner" - or url.database != "weave_b1_control" - or url.port in {None, 5432, 5434} - ): - pytest.fail("Refusing unguarded database: use the dedicated Weave integration backend") - admin = create_async_engine(url, isolation_level="AUTOCOMMIT") - name = f"weave_b1_test_{uuid4().hex}" - try: - async with admin.connect() as connection: - marker = await connection.scalar(text("SELECT identity FROM weave_test_backend_guard")) - if marker != "firefly-weave-b1-local-integration": - pytest.fail("Refusing database without the task-owned backend marker") - await connection.execute(text(f'CREATE DATABASE "{name}"')) - retain_resource("database", name) - except Exception: - pytest.fail("Real task-owned PostgreSQL is unavailable or its backend marker is missing", pytrace=False) - finally: - await admin.dispose() - from firefly_weave.settings import Settings - - return Settings(scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False)) + """An empty database, for tests that run migrations themselves.""" + return await create_database(migrated=False) @pytest.fixture -async def migration_settings(empty_settings): +async def fresh_migration_settings(empty_settings): + """An empty database migrated by this test, for tests about migrate() itself.""" from firefly_weave.persistence.migrations import migrate await migrate(empty_settings) return empty_settings +@pytest.fixture +async def migration_settings(): + """A database at the current schema, cloned from the migrated template, with owner credentials.""" + return await create_database(migrated=True) + + @pytest.fixture async def db(settings): engine = create_async_engine(settings.database_url.get_secret_value()) diff --git a/tests/integration/connectors/test_postgresql.py b/tests/integration/connectors/test_postgresql.py index 006f3552..14524d01 100644 --- a/tests/integration/connectors/test_postgresql.py +++ b/tests/integration/connectors/test_postgresql.py @@ -36,6 +36,8 @@ @pytest.fixture async def external_database(release_backends): + from conftest import retain_resource + value = os.environ.get("WEAVE_TEST_DATABASE_URL") if not value: pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL for colima-weave-tests", pytrace=False) @@ -63,11 +65,13 @@ async def external_database(release_backends): roles = {kind: f"d1_{kind}_{suffix}" for kind in ("read", "write")} try: await admin.execute(f'CREATE DATABASE "{name}"') + retain_resource("database", name) for kind, role in roles.items(): await admin.execute( f"CREATE ROLE {role} LOGIN PASSWORD '{password}' NOSUPERUSER NOCREATEDB " f"NOCREATEROLE NOINHERIT NOBYPASSRLS" ) + retain_resource("role", role) if kind == "read": await admin.execute(f"ALTER ROLE {role} SET default_transaction_read_only=on") finally: diff --git a/tests/integration/test_boot.py b/tests/integration/test_boot.py index 2482756e..ec24b237 100644 --- a/tests/integration/test_boot.py +++ b/tests/integration/test_boot.py @@ -268,13 +268,13 @@ async def test_readiness_real_connection_failure_is_redacted(settings, migration await admin.dispose() -async def test_explicit_migration_is_idempotent(migration_settings): +async def test_explicit_migration_is_idempotent(fresh_migration_settings): from sqlalchemy.ext.asyncio import create_async_engine from firefly_weave.persistence.migrations import migrate - await migrate(migration_settings) - engine = create_async_engine(migration_settings.database_url.get_secret_value()) + await migrate(fresh_migration_settings) + engine = create_async_engine(fresh_migration_settings.database_url.get_secret_value()) try: async with engine.connect() as connection: assert await connection.scalar(text("SELECT count(*) FROM weave_schema_version")) == 1 diff --git a/tests/integration/test_incident_migration.py b/tests/integration/test_incident_migration.py index eb98c792..82493597 100644 --- a/tests/integration/test_incident_migration.py +++ b/tests/integration/test_incident_migration.py @@ -40,6 +40,8 @@ def apply(sync): async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settings): + from conftest import retain_resource + engine = create_async_engine(empty_settings.database_url.get_secret_value()) role = "weave_c2_migration_" + uuid4().hex app = "weave_c2_app_" + uuid4().hex @@ -49,6 +51,8 @@ async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settin await upgrade(connection, "0009_triggers") await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS")) await connection.execute(text(f"CREATE ROLE {app} NOLOGIN NOSUPERUSER NOBYPASSRLS IN ROLE weave_app")) + retain_resource("role", role) + retain_resource("role", app) await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}")) await connection.execute(text(f"GRANT SELECT ON runs,principals TO {role}")) await connection.execute(text(f"GRANT REFERENCES ON runs,principals TO {role}")) diff --git a/tests/integration/test_schedule_migration.py b/tests/integration/test_schedule_migration.py index ea24d60f..cdde79ca 100644 --- a/tests/integration/test_schedule_migration.py +++ b/tests/integration/test_schedule_migration.py @@ -27,12 +27,15 @@ async def test_schedule_migration_requires_more_than_catalog_execute(empty_settings): + from conftest import retain_resource + engine = create_async_engine(empty_settings.database_url.get_secret_value()) role = "weave_c3_migration_" + uuid4().hex try: async with engine.begin() as connection: await upgrade(connection, "0010_incidents") await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS")) + retain_resource("role", role) await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}")) await connection.execute(text(f"GRANT SELECT,UPDATE ON weave_schema_version,alembic_version TO {role}")) await connection.execute(text(f"GRANT EXECUTE ON FUNCTION weave_tenant_ids() TO {role}")) diff --git a/uv.lock b/uv.lock index 8b0f5556..a35c9646 100644 --- a/uv.lock +++ b/uv.lock @@ -704,6 +704,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/99/89/87ef49ffe383ef4e147d27b7bf2088fb0b54ea409dd87b5a89442e5828a5/cryptography-50.0.1-cp39-abi3-win_amd64.whl", hash = "sha256:55d16b1ef3ee0958d893a977b19777887e546c9954ea81b200c3301a864013f2", size = 3875429, upload-time = "2026-08-25T19:45:24.418Z" }, ] +[[package]] +name = "execnet" +version = "2.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" }, +] + [[package]] name = "firefly-weave" version = "0.1.0a15" @@ -768,6 +777,7 @@ dev = [ { name = "mypy" }, { name = "pytest" }, { name = "pytest-asyncio" }, + { name = "pytest-xdist" }, { name = "ruff" }, { name = "types-jsonschema" }, { name = "types-pyyaml" }, @@ -822,6 +832,7 @@ dev = [ { name = "mypy", specifier = ">=1.20.2,<2" }, { name = "pytest", specifier = ">=9.0.3,<10" }, { name = "pytest-asyncio", specifier = ">=1.3.0,<2" }, + { name = "pytest-xdist", specifier = ">=3.8.0,<4" }, { name = "ruff", specifier = ">=0.15.12,<1" }, { name = "types-jsonschema", specifier = ">=4.26.0" }, { name = "types-pyyaml", specifier = ">=6.0.12" }, @@ -2538,6 +2549,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, ] +[[package]] +name = "pytest-xdist" +version = "3.8.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "execnet" }, + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" diff --git a/workers/agentic/uv.lock b/workers/agentic/uv.lock index e6b089c7..4442df69 100644 --- a/workers/agentic/uv.lock +++ b/workers/agentic/uv.lock @@ -731,6 +731,7 @@ dev = [ { name = "mypy", specifier = ">=1.20.2,<2" }, { name = "pytest", specifier = ">=9.0.3,<10" }, { name = "pytest-asyncio", specifier = ">=1.3.0,<2" }, + { name = "pytest-xdist", specifier = ">=3.8.0,<4" }, { name = "ruff", specifier = ">=0.15.12,<1" }, { name = "types-jsonschema", specifier = ">=4.26.0" }, { name = "types-pyyaml", specifier = ">=6.0.12" }, diff --git a/workers/files/uv.lock b/workers/files/uv.lock index c4334fca..b1d6d5f6 100644 --- a/workers/files/uv.lock +++ b/workers/files/uv.lock @@ -291,6 +291,7 @@ dev = [ { name = "mypy", specifier = ">=1.20.2,<2" }, { name = "pytest", specifier = ">=9.0.3,<10" }, { name = "pytest-asyncio", specifier = ">=1.3.0,<2" }, + { name = "pytest-xdist", specifier = ">=3.8.0,<4" }, { name = "ruff", specifier = ">=0.15.12,<1" }, { name = "types-jsonschema", specifier = ">=4.26.0" }, { name = "types-pyyaml", specifier = ">=6.0.12" }, From 7aabcfa3f1a8d2419cb7587b6fa011fc92a01e6f Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Tue, 13 Oct 2026 12:39:20 -0700 Subject: [PATCH 2/5] Record test-created databases and roles through a retain fixture Importing retain_resource from conftest resolved to whichever conftest pytest imported last, so a bare pytest run picked the live AI conftest and the four tests that create their own roles failed with ImportError. --- tests/integration/conftest.py | 10 ++++++++++ tests/integration/connectors/test_postgresql.py | 8 +++----- tests/integration/test_incident_migration.py | 8 +++----- tests/integration/test_operations.py | 5 ++--- tests/integration/test_schedule_migration.py | 6 ++---- 5 files changed, 20 insertions(+), 17 deletions(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 2709911d..ed2e8b9f 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -75,6 +75,16 @@ def retain_resource(kind: str, name: str) -> None: handle.write(json.dumps({"kind": kind, "name": name, "observed_at": datetime.now(UTC).isoformat()}) + "\n") +@pytest.fixture +def retain(): + """Record a database or role the test created itself, so it is dropped at teardown like the fixtures' own. + + Tests take this fixture rather than importing ``retain_resource``: ``from conftest import`` resolves to + whichever conftest pytest imported last, which is not this one when ``pytest`` or ``pytest tests`` runs. + """ + return retain_resource + + def control_url(): value = os.environ.get("WEAVE_TEST_DATABASE_URL") if not value: diff --git a/tests/integration/connectors/test_postgresql.py b/tests/integration/connectors/test_postgresql.py index 14524d01..49d263b8 100644 --- a/tests/integration/connectors/test_postgresql.py +++ b/tests/integration/connectors/test_postgresql.py @@ -35,9 +35,7 @@ @pytest.fixture -async def external_database(release_backends): - from conftest import retain_resource - +async def external_database(release_backends, retain): value = os.environ.get("WEAVE_TEST_DATABASE_URL") if not value: pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL for colima-weave-tests", pytrace=False) @@ -65,13 +63,13 @@ async def external_database(release_backends): roles = {kind: f"d1_{kind}_{suffix}" for kind in ("read", "write")} try: await admin.execute(f'CREATE DATABASE "{name}"') - retain_resource("database", name) + retain("database", name) for kind, role in roles.items(): await admin.execute( f"CREATE ROLE {role} LOGIN PASSWORD '{password}' NOSUPERUSER NOCREATEDB " f"NOCREATEROLE NOINHERIT NOBYPASSRLS" ) - retain_resource("role", role) + retain("role", role) if kind == "read": await admin.execute(f"ALTER ROLE {role} SET default_transaction_read_only=on") finally: diff --git a/tests/integration/test_incident_migration.py b/tests/integration/test_incident_migration.py index 82493597..c31e99af 100644 --- a/tests/integration/test_incident_migration.py +++ b/tests/integration/test_incident_migration.py @@ -39,9 +39,7 @@ def apply(sync): await connection.run_sync(apply) -async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settings): - from conftest import retain_resource - +async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settings, retain): engine = create_async_engine(empty_settings.database_url.get_secret_value()) role = "weave_c2_migration_" + uuid4().hex app = "weave_c2_app_" + uuid4().hex @@ -51,8 +49,8 @@ async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settin await upgrade(connection, "0009_triggers") await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS")) await connection.execute(text(f"CREATE ROLE {app} NOLOGIN NOSUPERUSER NOBYPASSRLS IN ROLE weave_app")) - retain_resource("role", role) - retain_resource("role", app) + retain("role", role) + retain("role", app) await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}")) await connection.execute(text(f"GRANT SELECT ON runs,principals TO {role}")) await connection.execute(text(f"GRANT REFERENCES ON runs,principals TO {role}")) diff --git a/tests/integration/test_operations.py b/tests/integration/test_operations.py index 7ecf7825..e438d448 100644 --- a/tests/integration/test_operations.py +++ b/tests/integration/test_operations.py @@ -729,11 +729,10 @@ async def test_inventory_checks_its_own_role_and_endpoint_with_scheduler_disable await service.close() -async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access_db, scheduler_url): +async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access_db, scheduler_url, retain): import secrets from types import SimpleNamespace - from conftest import retain_resource from sqlalchemy import make_url from firefly_weave.connections.registry import ConnectorRegistry @@ -743,7 +742,7 @@ async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access username, password = "weave_d4_column_" + uuid4().hex, secrets.token_urlsafe(32) async with access_db[1].begin() as owner: await owner.execute(text(f"CREATE ROLE {username} LOGIN PASSWORD '{password}' IN ROLE weave_scheduler")) - retain_resource("role", username) + retain("role", username) await owner.execute(text(f"GRANT SELECT(state) ON runs TO {username}")) assert not await owner.scalar(text("SELECT has_table_privilege(:role,'runs','SELECT')"), {"role": username}) assert await owner.scalar(text("SELECT has_any_column_privilege(:role,'runs','SELECT')"), {"role": username}) diff --git a/tests/integration/test_schedule_migration.py b/tests/integration/test_schedule_migration.py index cdde79ca..84901998 100644 --- a/tests/integration/test_schedule_migration.py +++ b/tests/integration/test_schedule_migration.py @@ -26,16 +26,14 @@ pytestmark = pytest.mark.integration -async def test_schedule_migration_requires_more_than_catalog_execute(empty_settings): - from conftest import retain_resource - +async def test_schedule_migration_requires_more_than_catalog_execute(empty_settings, retain): engine = create_async_engine(empty_settings.database_url.get_secret_value()) role = "weave_c3_migration_" + uuid4().hex try: async with engine.begin() as connection: await upgrade(connection, "0010_incidents") await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS")) - retain_resource("role", role) + retain("role", role) await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}")) await connection.execute(text(f"GRANT SELECT,UPDATE ON weave_schema_version,alembic_version TO {role}")) await connection.execute(text(f"GRANT EXECUTE ON FUNCTION weave_tenant_ids() TO {role}")) From 58c9db7bb7baf674ab6b2af7e109c841e39ddeda Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Tue, 13 Oct 2026 12:39:54 -0700 Subject: [PATCH 3/5] Attempt every test resource drop and report all failures together One failing DROP used to skip the remaining drops for that test. Each drop now runs on its own, and the teardown warning names every database and role left behind. --- tests/integration/conftest.py | 38 ++++++++++++---- tests/integration/test_resource_cleanup.py | 51 ++++++++++++++++++++++ 2 files changed, 81 insertions(+), 8 deletions(-) create mode 100644 tests/integration/test_resource_cleanup.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index ed2e8b9f..a38f62ae 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -179,9 +179,34 @@ async def create_database(*, migrated: bool): return Settings(scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False)) +async def drop_resources(connection, owned: list[tuple[str, str]]) -> list[str]: + """Drop the databases, then the roles; a failed drop is reported, never a reason to skip the rest. + + Databases go first so their roles no longer hold privileges in them. Returns one line per failed drop. + """ + failures = [] + for kind in ("database", "role"): + for item, name in owned: + if item != kind: + continue + try: + await connection.execute( + text(f'DROP {kind.upper()} IF EXISTS "{name}"' + " WITH (FORCE)" * (kind == "database")) + ) + except Exception as error: + failures.append(f"{kind} {name}: {type(error).__name__}") + return failures + + +@pytest.fixture +def resource_dropper(): + """The teardown drop step, for tests of the cleanup itself.""" + return drop_resources + + @pytest.fixture(autouse=True) async def drop_owned_resources(): - """Drop what the test created: its databases first, so its roles hold no privileges, then the roles.""" + """Drop what the test created and warn about anything left; a leak is reported, never turned into a failure.""" start = len(_OWNED) yield owned = _OWNED[start:] @@ -191,16 +216,13 @@ async def drop_owned_resources(): admin = create_async_engine(control_url(), isolation_level="AUTOCOMMIT") try: async with admin.connect() as connection: - for kind in ("database", "role"): - for item, name in owned: - if item == kind: - await connection.execute( - text(f'DROP {kind.upper()} IF EXISTS "{name}"' + " WITH (FORCE)" * (kind == "database")) - ) + failures = await drop_resources(connection, owned) except Exception as error: - warnings.warn(f"Test resources were not dropped: {type(error).__name__}", stacklevel=1) + failures = [f"{kind} {name}: {type(error).__name__}" for kind, name in owned] finally: await admin.dispose() + if failures: + warnings.warn("Test resources were not dropped: " + "; ".join(failures), stacklevel=1) @pytest.fixture diff --git a/tests/integration/test_resource_cleanup.py b/tests/integration/test_resource_cleanup.py new file mode 100644 index 00000000..a126c115 --- /dev/null +++ b/tests/integration/test_resource_cleanup.py @@ -0,0 +1,51 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + +"""The integration fixtures' teardown drops everything a test created, even when one drop fails.""" + + +class Connection: + def __init__(self, failing: set[str]): + self.failing = failing + self.statements: list[str] = [] + + async def execute(self, statement): + sql = str(statement) + self.statements.append(sql) + if any(name in sql for name in self.failing): + raise RuntimeError("dependent objects still exist") + + +async def test_every_drop_is_attempted_and_every_failure_reported(resource_dropper): + owned = [("role", "role_a"), ("database", "db_a"), ("database", "db_b"), ("role", "role_b")] + connection = Connection({"db_a", "role_a"}) + + failures = await resource_dropper(connection, owned) + + assert connection.statements == [ + 'DROP DATABASE IF EXISTS "db_a" WITH (FORCE)', + 'DROP DATABASE IF EXISTS "db_b" WITH (FORCE)', + 'DROP ROLE IF EXISTS "role_a"', + 'DROP ROLE IF EXISTS "role_b"', + ] + assert failures == ["database db_a: RuntimeError", "role role_a: RuntimeError"] + + +async def test_no_failure_reports_nothing(resource_dropper): + connection = Connection(set()) + + assert await resource_dropper(connection, [("database", "db_a"), ("role", "role_a")]) == [] + assert len(connection.statements) == 2 From 68f440e2cbadc59bc197a12f82b3d97e781589cd Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Tue, 13 Oct 2026 12:44:42 -0700 Subject: [PATCH 4/5] Keep other checkouts' templates while they are in use A template build dropped every other weave_b1_template_* database, so two checkouts with different schemas on one server broke each other's clones. A process now records in the template's comment when it last used it, and a build drops only templates nobody is connected to and nobody has used for 6 hours. The template name also hashes the module that defines migrate() and the Alembic version, so a change there that alters the schema builds a new template. --- tests/integration/conftest.py | 67 ++++++++++++++++++---- tests/integration/test_template_cleanup.py | 58 +++++++++++++++++++ 2 files changed, 113 insertions(+), 12 deletions(-) create mode 100644 tests/integration/test_template_cleanup.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index a38f62ae..df0d90c5 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -25,6 +25,7 @@ import hashlib import os import runpy +import time import warnings from pathlib import Path from uuid import uuid4 @@ -49,9 +50,19 @@ def validate_release_receipt(release_backends): # Databases and roles created by the running test, dropped at teardown by drop_owned_resources. _OWNED: list[tuple[str, str]] = [] -# Name of the migrated template database once this process has found or built it. +# Name of the migrated template database once this process has found or built it, and when it last marked it used. _TEMPLATE: list[str] = [] +_TOUCHED: list[float] = [] TEMPLATE_LOCK = 7260912902 +# Ruling: a build drops another checkout's template only when no session is connected to it and its last +# recorded use is older than 6 hours — a running process marks its template used when it starts and again at +# most every 10 minutes while it clones, and the longest serial integration run is about 35 minutes, so a +# template in use by another checkout is never dropped — cost if wrong: a process idle for over 6 hours while +# another checkout builds a different schema sees its next clone fail as an unavailable backend (a rerun +# rebuilds it), and templates from successive migration edits stay on the server for up to 6 hours. +TEMPLATE_STALE_AFTER = 6 * 60 * 60 +TEMPLATE_TOUCH_EVERY = 10 * 60 +TEMPLATE_USED = "weave-template-used-at=" class TemplateBuildError(Exception): @@ -102,7 +113,13 @@ def control_url(): def template_name() -> str: - """Name the migrated template after everything that shapes its schema, so a changed migration rebuilds it.""" + """Name the migrated template after everything that shapes its schema, so any change to it builds a new one. + + The hash covers the migration files, the schema version, the default operations policy the migrations record, + the module that defines ``migrate()`` and ``check_schema()``, and the installed Alembic version. + """ + import alembic + import firefly_weave.persistence.migrations as module from firefly_weave.contracts.operational_policy import OperationsPolicy @@ -111,13 +128,46 @@ def template_name() -> str: digest = hashlib.sha256(module.SCHEMA_VERSION.encode()) for path in sorted(source.rglob("*.py")): digest.update(path.relative_to(source).as_posix().encode() + b"\0" + path.read_bytes() + b"\0") - digest.update(OperationsPolicy().model_dump_json().encode()) + digest.update(OperationsPolicy().model_dump_json().encode() + b"\0") + digest.update(Path(module.__file__).read_bytes() + b"\0") + digest.update(alembic.__version__.encode()) return "weave_b1_template_" + digest.hexdigest()[:16] +async def mark_template_used(connection, name: str) -> None: + """Record, in the template's comment, the server time a process last used it.""" + now = int(await connection.scalar(text("SELECT floor(extract(epoch FROM now()))"))) + await connection.execute(text(f"COMMENT ON DATABASE \"{name}\" IS '{TEMPLATE_USED}{now}'")) + _TOUCHED[:] = [time.monotonic()] + + +async def drop_stale_templates(connection, keep: str) -> None: + """Drop other templates nobody is connected to and nobody has used recently; see TEMPLATE_STALE_AFTER.""" + stale = await connection.scalars( + text( + "SELECT d.datname FROM pg_database d WHERE d.datname LIKE 'weave\\_b1\\_template\\_%' AND d.datname<>:keep" + " AND NOT EXISTS (SELECT 1 FROM pg_stat_activity a WHERE a.datname=d.datname)" + " AND coalesce(substring(shobj_description(d.oid,'pg_database') FROM :pattern)::bigint, 0)" + " < extract(epoch FROM now()) - :age" + ), + {"keep": keep, "pattern": f"^{TEMPLATE_USED}([0-9]+)$", "age": TEMPLATE_STALE_AFTER}, + ) + for old in stale.all(): + await connection.execute(text(f'ALTER DATABASE "{old}" IS_TEMPLATE false')) + await connection.execute(text(f'DROP DATABASE IF EXISTS "{old}" WITH (FORCE)')) + + +@pytest.fixture +def template_sweep(): + """The stale-template sweep a template build runs, with the comment prefix that records a template's last use.""" + return drop_stale_templates, TEMPLATE_USED, TEMPLATE_STALE_AFTER + + async def migrated_template(connection, url) -> str: """Find or build the migrated template once; the advisory lock makes parallel workers build it only once.""" if _TEMPLATE: + if time.monotonic() - _TOUCHED[0] > TEMPLATE_TOUCH_EVERY: + await mark_template_used(connection, _TEMPLATE[0]) return _TEMPLATE[0] from firefly_weave.persistence.migrations import migrate from firefly_weave.settings import Settings @@ -131,15 +181,7 @@ async def migrated_template(connection, url) -> str: if not ready: # A template is marked only after a complete migration; anything else is a failed build. await connection.execute(text(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')) - stale = await connection.scalars( - text( - "SELECT datname FROM pg_database WHERE datname LIKE 'weave\\_b1\\_template\\_%' AND datname<>:name" - ), - {"name": name}, - ) - for old in stale.all(): - await connection.execute(text(f'ALTER DATABASE "{old}" IS_TEMPLATE false')) - await connection.execute(text(f'DROP DATABASE IF EXISTS "{old}" WITH (FORCE)')) + await drop_stale_templates(connection, name) await connection.execute(text(f'CREATE DATABASE "{name}"')) settings = Settings( scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False) @@ -150,6 +192,7 @@ async def migrated_template(connection, url) -> str: raise TemplateBuildError("Migrating the template database failed") from error # Clones need a template nobody is connected to. await connection.execute(text(f'ALTER DATABASE "{name}" IS_TEMPLATE true ALLOW_CONNECTIONS false')) + await mark_template_used(connection, name) finally: await connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": TEMPLATE_LOCK}) _TEMPLATE.append(name) diff --git a/tests/integration/test_template_cleanup.py b/tests/integration/test_template_cleanup.py new file mode 100644 index 00000000..9935a656 --- /dev/null +++ b/tests/integration/test_template_cleanup.py @@ -0,0 +1,58 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + +"""A template build drops only the other templates nobody is using.""" + +import os +from uuid import uuid4 + +import pytest +from sqlalchemy import make_url, text +from sqlalchemy.ext.asyncio import create_async_engine + +pytestmark = pytest.mark.integration + + +async def test_build_keeps_templates_in_use_or_recently_used(empty_settings, retain, template_sweep): + sweep, used, stale_after = template_sweep + control = make_url(os.environ["WEAVE_TEST_DATABASE_URL"]) + prefix = "weave_b1_template_" + uuid4().hex[:8] + names = {kind: f"{prefix}_{kind}" for kind in ("recent", "connected", "old", "unmarked")} + admin = create_async_engine(control, isolation_level="AUTOCOMMIT") + try: + async with admin.connect() as connection: + now = int(await connection.scalar(text("SELECT floor(extract(epoch FROM now()))"))) + for kind, name in names.items(): + await connection.execute(text(f'CREATE DATABASE "{name}"')) + retain("database", name) + if kind != "unmarked": + seen = now - 60 if kind == "recent" else now - stale_after - 60 + await connection.execute(text(f"COMMENT ON DATABASE \"{name}\" IS '{used}{seen}'")) + holder = create_async_engine(control.set(database=names["connected"])) + try: + async with holder.connect() as held: + await held.execute(text("SELECT 1")) + await sweep(connection, "weave_b1_template_none") + finally: + await holder.dispose() + left = set( + await connection.scalars( + text("SELECT datname FROM pg_database WHERE datname LIKE :prefix"), {"prefix": prefix + "%"} + ) + ) + finally: + await admin.dispose() + assert left == {names["recent"], names["connected"]} From ecf43dca09f84e21131f01b82ad525eb1a871097 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Tue, 13 Oct 2026 12:44:57 -0700 Subject: [PATCH 5/5] Describe when the integration template is built and that durability settings are optional The first test of a run builds the template only when it is missing, and turning off durability made no measurable difference. Also document which templates a build drops. --- CONTRIBUTING.md | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6976d246..1dde59c3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -140,10 +140,14 @@ The integration fixtures give every test its own database on the server named by `WEAVE_TEST_DATABASE_URL`: - A test that needs the current schema clones a template database, - `weave_b1_template_`. The first test of a run migrates it once; its name - is a hash of the migration files, so changing a migration builds a new template - and drops the old one. The template stays on the server between runs and holds - no test data. + `weave_b1_template_`. The first test of a run finds that template, or + builds it by running the migrations when it is missing. Its name is a hash of + the migration files, the code that runs them and the Alembic version, so + changing a migration builds a new template. The template stays on the server + between runs and holds no test data. +- A build drops other templates only when nobody is connected to them and no + run has used them for 6 hours, so checkouts with different migrations can + share one server. - A test about migrations themselves (the `empty_settings` and `fresh_migration_settings` fixtures) starts from an empty database and runs the real migrations. @@ -162,11 +166,11 @@ uv run --locked --all-extras pytest tests/integration -n 4 --dist loadfile `--dist loadfile` keeps each test file on one worker, so a file's module fixture (the Kafka broker, for example) starts once. Without `-n` the suite runs serially. -A server used only for these tests can skip durability. Start a throwaway -PostgreSQL with `-c fsync=off -c synchronous_commit=off -c full_page_writes=off` -after the image name of its `docker run` command; an unclean stop can corrupt its -data, so never use these settings on a server whose data you keep, such as the -Compose development stack. +Turning off durability on the test server is optional: with cloned and dropped +databases it made no measurable difference. If you try it, pass +`-c fsync=off -c synchronous_commit=off -c full_page_writes=off` only to a +throwaway server; an unclean stop can corrupt its data, so never to a server +whose data you keep, such as the Compose development stack. [Local runtime setup](docs/reference/local-runtime.md) explains provisioning. End-to-end fixtures exercise installed wheels, admitted images, identities and