Move the integration subdirectories' database gates to the shared helper (#366)

The same conversion as #363, on the 33 modules under tests/integration/scan,
admin and audit. Each module keeps its own fixture with its own scope and
position; only the body becomes a call.

Declared 33 before running, all 33 parse, and the converted files pass.

Author: haksungjang

apps/backend/tests/integration/admin/test_admin_backup_api.py

@@ -14,8 +14,6 @@
 from __future__ import annotations
 
 import json
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime
@@ -28,41 +26,22 @@
 
 from core.security import create_access_token
 from models import User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_membership,
     make_organization,
     make_team,
     make_user,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set — skip admin backup API tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(  # noqa: S603, S607
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        timeout=120,
-    )
-    if result.returncode != 0:
-        pytest.skip(
-            f"alembic upgrade head failed; admin backup API tests cannot run\n"
-            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
-        )
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/admin/test_admin_ops_api.py

@@ -16,8 +16,6 @@
 
 from __future__ import annotations
 
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime
@@ -28,6 +26,7 @@
 
 from core.security import create_access_token, hash_password
 from models import User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_membership,
     make_organization,
@@ -38,34 +37,14 @@
     principal_for,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set — skip admin ops API tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        timeout=120,
-    )
-    if result.returncode != 0:
-        pytest.skip(
-            f"alembic upgrade head failed; admin ops API tests cannot run\n"
-            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
-        )
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/admin/test_admin_teams_api.py

@@ -18,18 +18,16 @@
 
 from __future__ import annotations
 
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
-from pathlib import Path
 
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import text
 
 from core.security import create_access_token
 from models import User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_membership,
     make_organization,
@@ -40,34 +38,14 @@
     unique_suffix,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set — skip admin teams API tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        timeout=120,
-    )
-    if result.returncode != 0:
-        pytest.skip(
-            f"alembic upgrade head failed; admin teams API tests cannot run\n"
-            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
-        )
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/admin/test_admin_user_onboarding_api.py

@@ -14,51 +14,32 @@
 
 from __future__ import annotations
 
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime, timedelta
-from pathlib import Path
 
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import select
 
 from core.security import create_access_token, verify_password
 from models import AuditLog, Membership, RefreshToken, User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_organization,
     make_team,
     make_user,
     unique_suffix,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set, skipping admin onboarding tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if result.returncode != 0:
-        pytest.skip(f"alembic upgrade failed: {result.stderr[-400:]}")
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/admin/test_admin_users_api.py

@@ -17,18 +17,16 @@
 
 from __future__ import annotations
 
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
-from pathlib import Path
 
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import select, text
 
 from core.security import create_access_token
 from models import RefreshToken, User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_membership,
     make_organization,
@@ -37,34 +35,14 @@
     unique_suffix,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set — skip admin users API tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        timeout=120,
-    )
-    if result.returncode != 0:
-        pytest.skip(
-            f"alembic upgrade head failed; admin users API tests cannot run\n"
-            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
-        )
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/audit/test_audit_log_retention_report.py

@@ -27,17 +27,15 @@
 from __future__ import annotations
 
 import os
-import subprocess
 import uuid
 from collections.abc import Iterator
 from datetime import UTC, datetime, timedelta
-from pathlib import Path
 
 import pytest
 from sqlalchemy import create_engine, text
 from sqlalchemy.orm import Session, sessionmaker
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+from tests._db_required import migrate_to_head
 
 pytestmark = pytest.mark.integration
 
@@ -51,17 +49,7 @@ def _sync_url() -> str:
 
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    if not os.getenv("DATABASE_URL"):
-        pytest.skip("DATABASE_URL not set")
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if result.returncode != 0:
-        pytest.skip(f"alembic upgrade failed: {result.stderr[-400:]}")
+    migrate_to_head()
 
 
 @pytest.fixture

apps/backend/tests/integration/audit/test_team_scoped_audit.py

@@ -14,52 +14,30 @@
 
 from __future__ import annotations
 
-import os
-import subprocess
 import uuid
 from collections.abc import AsyncIterator
-from pathlib import Path
 
 import pytest
 from httpx import ASGITransport, AsyncClient
 
 from core.security import create_access_token
 from models import User
+from tests._db_required import migrate_to_head
 from tests._helpers import (
     make_membership,
     make_organization,
     make_team,
     make_user,
 )
 
-BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 PROBLEM_JSON = "application/problem+json"
 
 pytestmark = pytest.mark.integration
 
 
-def _require_database_url() -> str:
-    url = os.getenv("DATABASE_URL")
-    if not url:
-        pytest.skip("DATABASE_URL not set — skip team-scoped audit tests")
-    return url
-
-
 @pytest.fixture(scope="module", autouse=True)
 def _migrate_once() -> None:
-    _require_database_url()
-    result = subprocess.run(
-        ["alembic", "upgrade", "head"],
-        cwd=BACKEND_ROOT,
-        capture_output=True,
-        text=True,
-        timeout=120,
-    )
-    if result.returncode != 0:
-        pytest.skip(
-            f"alembic upgrade head failed; team-scoped audit tests cannot run\n"
-            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
-        )
+    migrate_to_head()
 
 
 @pytest.fixture