fix(gtea): ruff format + check all gtea files and bridge.py

Clears cc-ci self-test lint failures:
- ruff format: 9 files reformatted (all gtea test files + test_discovery.py)
- ruff check --fix: bridge.py UP017 (datetime.UTC alias) + 6 gtea check errors
- manifest.py B007: rename unused loop variable path → _path (no auto-fix available)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

bridge/bridge.py

@@ -37,7 +37,7 @@ import time
 import urllib.error
 import urllib.parse
 import urllib.request
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 GITEA_API = os.environ.get("GITEA_API", "https://git.autonomic.zone/api/v1")
@@ -82,7 +82,7 @@ GITEA_TOKEN = _read(os.environ["GITEA_TOKEN_FILE"])
 # Shared dedup across the poll + webhook paths: a comment id triggers at most one run.
 _PROCESSED: set = set()
 _PROCESSED_LOCK = threading.Lock()
-_PROCESS_STARTED_AT = datetime.now(timezone.utc)
+_PROCESS_STARTED_AT = datetime.now(UTC)
 
 
 def log(*a):

runner/harness/manifest.py

@@ -51,7 +51,7 @@ def _pre_ops(path: str) -> list[str]:
 
 def _custom_counts(recipe: str, repo_local: str | None) -> dict[str, dict[str, int]]:
     out: dict[str, dict[str, int]] = {}
-    for source, path in discovery.custom_tests(recipe, repo_local):
+    for source, _path in discovery.custom_tests(recipe, repo_local):
         sub = "custom"
         out.setdefault(source, {}).setdefault(sub, 0)
         out[source][sub] += 1

tests/gitea/custom/test_admin_api.py

@@ -16,10 +16,10 @@ import urllib.request
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-from ops import admin_creds  # noqa: E402
-
 import ssl
 
+from ops import admin_creds  # noqa: E402
+
 _CTX = ssl.create_default_context()
 _CTX.check_hostname = False
 _CTX.verify_mode = ssl.CERT_NONE
@@ -68,7 +68,9 @@ def test_admin_api_user_org_token_lifecycle(live_app):
 
     # 1. Create test user
     status, body = _api(
-        live_app, "/admin/users", method="POST",
+        live_app,
+        "/admin/users",
+        method="POST",
         body={
             "username": test_user,
             "email": f"{test_user}@ci.local",
@@ -77,7 +79,8 @@ def test_admin_api_user_org_token_lifecycle(live_app):
             "login_name": test_user,
             "source_id": 0,
         },
-        user=adm_user, password=adm_pass,
+        user=adm_user,
+        password=adm_pass,
     )
     assert status == 201, f"user create HTTP {status}: {body}"
     assert body.get("login") == test_user, f"unexpected login: {body}"
@@ -85,9 +88,12 @@ def test_admin_api_user_org_token_lifecycle(live_app):
     try:
         # 2. Create org (as admin, add test user as member)
         status, body = _api(
-            live_app, "/orgs", method="POST",
+            live_app,
+            "/orgs",
+            method="POST",
             body={"username": test_org, "visibility": "public"},
-            user=adm_user, password=adm_pass,
+            user=adm_user,
+            password=adm_pass,
         )
         assert status == 201, f"org create HTTP {status}: {body}"
         assert body.get("username") == test_org, f"unexpected org: {body}"
@@ -96,9 +102,12 @@ def test_admin_api_user_org_token_lifecycle(live_app):
             # 3. Create API token for test user (admin creates token on behalf of user).
             # Gitea 1.22+ requires explicit scopes; supply the minimum needed for steps 4-5.
             status, tok_body = _api(
-                live_app, f"/users/{test_user}/tokens", method="POST",
+                live_app,
+                f"/users/{test_user}/tokens",
+                method="POST",
                 body={"name": token_name, "scopes": ["read:user", "read:organization"]},
-                user=adm_user, password=adm_pass,
+                user=adm_user,
+                password=adm_pass,
             )
             assert status == 201, f"token create HTTP {status}: {tok_body}"
             token = tok_body.get("sha1")
@@ -116,8 +125,11 @@ def test_admin_api_user_org_token_lifecycle(live_app):
 
             # 6. Delete the token
             status, _ = _api(
-                live_app, f"/users/{test_user}/tokens/{token_name}", method="DELETE",
+                live_app,
-                user=adm_user, password=adm_pass,
+                f"/users/{test_user}/tokens/{token_name}",
+                method="DELETE",
+                user=adm_user,
+                password=adm_pass,
             )
             assert status in (204, 404), f"token delete HTTP {status}"
 
@@ -127,5 +139,6 @@ def test_admin_api_user_org_token_lifecycle(live_app):
 
     finally:
         # Delete test user (admin only)
-        _api(live_app, f"/admin/users/{test_user}", method="DELETE",
+        _api(
-             user=adm_user, password=adm_pass)
+            live_app, f"/admin/users/{test_user}", method="DELETE", user=adm_user, password=adm_pass
+        )

tests/gitea/custom/test_git_push.py

@@ -21,10 +21,10 @@ import urllib.request
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-from ops import admin_creds  # noqa: E402
-
 import ssl
 
+from ops import admin_creds  # noqa: E402
+
 _CTX = ssl.create_default_context()
 _CTX.check_hostname = False
 _CTX.verify_mode = ssl.CERT_NONE
@@ -72,9 +72,12 @@ def test_git_push(live_app):
 
     # 1. Create test repo (auto_init adds an initial commit so the branch exists)
     status, body = _api(
-        live_app, "/user/repos", method="POST",
+        live_app,
+        "/user/repos",
+        method="POST",
         body={"name": repo_name, "private": False, "auto_init": True, "default_branch": "main"},
-        user=user, password=password,
+        user=user,
+        password=password,
     )
     assert status == 201, f"repo create HTTP {status}: {body}"
     # Embed credentials directly in the URL (password is 32-char hex, URL-safe).
@@ -107,14 +110,14 @@ def test_git_push(live_app):
 
         # 5. Verify commit landed via API
         status, commits = _api(
-            live_app, f"/repos/{user}/{repo_name}/commits?limit=1",
+            live_app,
-            user=user, password=password,
+            f"/repos/{user}/{repo_name}/commits?limit=1",
+            user=user,
+            password=password,
         )
         assert status == 200 and commits, f"commit list HTTP {status}: {commits}"
         commit_msg = commits[0].get("commit", {}).get("message", "").strip()
-        assert "automated push test" in commit_msg, (
+        assert "automated push test" in commit_msg, f"Unexpected commit message: {commit_msg!r}"
-            f"Unexpected commit message: {commit_msg!r}"
-        )
     finally:
         shutil.rmtree(tmpdir, ignore_errors=True)  # removes parent + repo subdir
         # 6. Cleanup — delete the test repo

tests/gitea/custom/test_lfs_roundtrip.py

@@ -28,11 +28,11 @@ import urllib.request
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-from harness import abra as harness_abra, lifecycle  # noqa: E402
-from ops import admin_creds  # noqa: E402
-
 import ssl
 
+from harness import lifecycle  # noqa: E402
+from ops import admin_creds  # noqa: E402
+
 _CTX = ssl.create_default_context()
 _CTX.check_hostname = False
 _CTX.verify_mode = ssl.CERT_NONE
@@ -83,6 +83,7 @@ def test_lfs_roundtrip(live_app):
     """
     if not _lfs_available():
         import pytest
+
         pytest.skip(
             "compose.lfs.yml absent in gitea recipe checkout — LFS is not enabled on this branch. "
             "This test runs on lfs-plain-gitea (PR #1) and is EXPECTED_NA on main."
@@ -103,9 +104,12 @@ def test_lfs_roundtrip(live_app):
 
     # 1. Create LFS test repo
     status, body = _api(
-        live_app, "/user/repos", method="POST",
+        live_app,
+        "/user/repos",
+        method="POST",
         body={"name": repo_name, "private": False, "auto_init": True, "default_branch": "main"},
-        user=user, password=password,
+        user=user,
+        password=password,
     )
     assert status in (201, 409), f"repo create HTTP {status}: {body}"
 
@@ -136,9 +140,14 @@ def test_lfs_roundtrip(live_app):
         # Verify git-lfs pointer was tracked correctly
         lfs_ls = subprocess.run(
             ["git", "lfs", "ls-files"],
-            cwd=tmpdir, capture_output=True, text=True, env={**os.environ, **git_env}
+            cwd=tmpdir,
+            capture_output=True,
+            text=True,
+            env={**os.environ, **git_env},
         )
-        assert "testblob.bin" in lfs_ls.stdout, f"testblob.bin not in git-lfs ls-files: {lfs_ls.stdout}"
+        assert (
+            "testblob.bin" in lfs_ls.stdout
+        ), f"testblob.bin not in git-lfs ls-files: {lfs_ls.stdout}"
 
         # 6. Download in a FRESH clone (proves the LFS server stores and serves the object)
         fresh_dir = tempfile.mkdtemp(prefix="ccci-gitea-lfs-dl-")
@@ -149,9 +158,9 @@ def test_lfs_roundtrip(live_app):
             with open(fetched_path, "rb") as f:
                 fetched = f.read()
             fetched_sha256 = hashlib.sha256(fetched).hexdigest()
-            assert fetched_sha256 == expected_sha256, (
+            assert (
-                f"LFS round-trip OID mismatch: expected {expected_oid}, got sha256:{fetched_sha256}"
+                fetched_sha256 == expected_sha256
-            )
+            ), f"LFS round-trip OID mismatch: expected {expected_oid}, got sha256:{fetched_sha256}"
         finally:
             shutil.rmtree(fresh_dir, ignore_errors=True)
 
@@ -162,21 +171,22 @@ def test_lfs_roundtrip(live_app):
             ["sh", "-c", "grep -E '^LFS_JWT_SECRET' /etc/gitea/app.ini || echo NOT_FOUND"],
             timeout=30,
         ).strip()
-        assert current_jwt and "NOT_FOUND" not in current_jwt, (
+        assert (
-            "Could not read LFS_JWT_SECRET from /etc/gitea/app.ini before restart"
+            current_jwt and "NOT_FOUND" not in current_jwt
-        )
+        ), "Could not read LFS_JWT_SECRET from /etc/gitea/app.ini before restart"
 
         # Restart the gitea container
         lifecycle.exec_in_app(live_app, ["true"], timeout=5)  # no-op to confirm exec works
         subprocess.run(
-            ["docker", "service", "update", "--force",
+            ["docker", "service", "update", "--force", live_app.replace(".", "_") + "_app"],
-             live_app.replace(".", "_") + "_app"],
+            capture_output=True,
-            capture_output=True, timeout=120,
+            timeout=120,
         )
         # Wait for gitea to come back up
-        from harness import generic
+
         # Re-read meta from the live_app fixture (meta is not in scope here — use the stored meta)
         import time
+
         deadline = time.time() + 120
         while time.time() < deadline:
             status2, _ = _api(live_app, "/version", user=user, password=password)
@@ -203,9 +213,9 @@ def test_lfs_roundtrip(live_app):
             assert os.path.exists(pr_blob), "testblob.bin not fetched in post-restart clone"
             with open(pr_blob, "rb") as f:
                 pr_data = f.read()
-            assert hashlib.sha256(pr_data).hexdigest() == expected_sha256, (
+            assert (
-                "LFS object corrupted after restart — JWT secret may have changed"
+                hashlib.sha256(pr_data).hexdigest() == expected_sha256
-            )
+            ), "LFS object corrupted after restart — JWT secret may have changed"
         finally:
             shutil.rmtree(post_restart_dir, ignore_errors=True)
 

tests/gitea/ops.py

@@ -42,6 +42,7 @@ def _ssl_ctx():
     global _SSL_CTX
     if _SSL_CTX is None:
         import ssl
+
         ctx = ssl.create_default_context()
         ctx.check_hostname = False
         ctx.verify_mode = ssl.CERT_NONE
@@ -80,10 +81,17 @@ def _ensure_admin(domain: str) -> tuple[str, str]:
         lifecycle.exec_in_app(
             domain,
             [
-                "gitea", "admin", "user", "create", "--admin",
+                "gitea",
-                "--username", _ADMIN_USER,
+                "admin",
-                "--password", password,
+                "user",
-                "--email", _ADMIN_EMAIL,
+                "create",
+                "--admin",
+                "--username",
+                _ADMIN_USER,
+                "--password",
+                password,
+                "--email",
+                _ADMIN_EMAIL,
                 "--must-change-password=false",
             ],
             timeout=120,
@@ -93,9 +101,14 @@ def _ensure_admin(domain: str) -> tuple[str, str]:
             lifecycle.exec_in_app(
                 domain,
                 [
-                    "gitea", "admin", "user", "change-password",
+                    "gitea",
-                    "--username", _ADMIN_USER,
+                    "admin",
-                    "--password", password,
+                    "user",
+                    "change-password",
+                    "--username",
+                    _ADMIN_USER,
+                    "--password",
+                    password,
                 ],
                 timeout=60,
             )
@@ -132,9 +145,12 @@ def _gitea_api(
 def _create_marker_repo(domain: str, user: str, password: str) -> bool:
     """Create ci-marker repo with auto_init=True. Returns True if created or already exists."""
     status, _ = _gitea_api(
-        domain, "/user/repos", method="POST",
+        domain,
+        "/user/repos",
+        method="POST",
         body={"name": _MARKER_REPO, "private": False, "auto_init": True, "default_branch": "main"},
-        user=user, password=password,
+        user=user,
+        password=password,
     )
     return status in (201, 409)
 
@@ -142,17 +158,18 @@ def _create_marker_repo(domain: str, user: str, password: str) -> bool:
 def _delete_marker_repo(domain: str, user: str, password: str) -> bool:
     """Delete ci-marker repo. Returns True if deleted or already gone."""
     status, _ = _gitea_api(
-        domain, f"/repos/{user}/{_MARKER_REPO}", method="DELETE",
+        domain,
-        user=user, password=password,
+        f"/repos/{user}/{_MARKER_REPO}",
+        method="DELETE",
+        user=user,
+        password=password,
     )
     return status in (204, 404)
 
 
 def marker_repo_exists(domain: str, user: str, password: str) -> bool:
     """Check whether the ci-marker repo is present. Called by test_*.py overlays."""
-    status, _ = _gitea_api(
+    status, _ = _gitea_api(domain, f"/repos/{user}/{_MARKER_REPO}", user=user, password=password)
-        domain, f"/repos/{user}/{_MARKER_REPO}", user=user, password=password
-    )
     return status == 200
 
 
@@ -161,9 +178,7 @@ def admin_creds(domain: str) -> tuple[str, str]:
     existing = _load_creds(domain)
     if existing:
         return existing
-    raise RuntimeError(
+    raise RuntimeError(f"No admin creds for {domain} — was ops.pre_install called for this run?")
-        f"No admin creds for {domain} — was ops.pre_install called for this run?"
-    )
 
 
 def pre_install(ctx):
@@ -205,7 +220,7 @@ def pre_restore(ctx):
     generic.assert_serving(ctx.domain, ctx.meta)
     ok = _delete_marker_repo(ctx.domain, user, password)
     assert ok, f"pre_restore: could not delete {_MARKER_REPO} repo on {ctx.domain}"
-    assert not marker_repo_exists(ctx.domain, user, password), (
+    assert not marker_repo_exists(
-        f"pre_restore: {_MARKER_REPO} still present after delete — divergence did not take"
+        ctx.domain, user, password
-    )
+    ), f"pre_restore: {_MARKER_REPO} still present after delete — divergence did not take"
     print(f"  gitea ops: {_MARKER_REPO!r} deleted (diverged from backup state)", flush=True)

tests/gitea/recipe_meta.py

@@ -35,6 +35,7 @@ def READY_PROBE(ctx):
 def SCREENSHOT(page, ctx):
     # Navigate to the sign-in page — a credential-free view that shows the gitea UI.
     from playwright.sync_api import sync_playwright  # noqa: F401 — re-entry guard
+
     page.goto(f"{ctx.base_url}/user/login", wait_until="networkidle", timeout=30_000)
     page.wait_for_selector("form.ui.form", timeout=15_000)
 

tests/gitea/test_backup.py

@@ -22,6 +22,6 @@ def test_backup_captures_marker_repo(live_app, meta):
     # backupbot cycles the gitea container during backup — wait for it to be back up.
     generic.assert_serving(live_app, meta)
     user, password = admin_creds(live_app)
-    assert marker_repo_exists(live_app, user, password), (
+    assert marker_repo_exists(
-        f"{live_app}: ci-marker repo is not present at backup time (backup would capture empty state)"
+        live_app, user, password
-    )
+    ), f"{live_app}: ci-marker repo is not present at backup time (backup would capture empty state)"

tests/gitea/test_install.py

@@ -33,8 +33,9 @@ def test_install_gitea(live_app, meta):
     # 3. Admin API reachable + authenticated — GET /api/v1/users/search as ci_admin
     user, password = admin_creds(live_app)
     import base64
-    import urllib.request
     import ssl
+    import urllib.request
+
     ctx = ssl.create_default_context()
     ctx.check_hostname = False
     ctx.verify_mode = ssl.CERT_NONE
@@ -44,6 +45,7 @@ def test_install_gitea(live_app, meta):
         headers={"Authorization": f"Basic {auth}"},
     )
     import urllib.error
+
     try:
         with urllib.request.urlopen(req, timeout=15, context=ctx) as r:
             assert r.status == 200, f"Admin API /users/search returned {r.status}"
@@ -52,16 +54,19 @@ def test_install_gitea(live_app, meta):
 
     # 4. Playwright: sign-in page renders
     from playwright.sync_api import sync_playwright
+
     url = f"https://{live_app}/user/login"
     with sync_playwright() as p:
         browser = p.chromium.launch(args=["--no-sandbox"])
         try:
             page = browser.new_context(ignore_https_errors=True).new_page()
-            harness_browser.goto_with_retry(page, url, accept_statuses=(200, 302), goto_timeout_ms=30_000)
+            harness_browser.goto_with_retry(
+                page, url, accept_statuses=(200, 302), goto_timeout_ms=30_000
+            )
             page.wait_for_selector("input#user_name", timeout=20_000)
             content = page.content()
-            assert "gitea" in content.lower() or "sign in" in content.lower(), (
+            assert (
-                "Sign-in page did not render expected gitea content"
+                "gitea" in content.lower() or "sign in" in content.lower()
-            )
+            ), "Sign-in page did not render expected gitea content"
         finally:
             browser.close()

tests/gitea/test_upgrade.py

@@ -20,6 +20,6 @@ def test_upgrade_preserves_marker_repo(live_app, meta):
     """The ci-marker repo survived the upgrade to the PR head (data continuity)."""
     generic.assert_serving(live_app, meta)
     user, password = admin_creds(live_app)
-    assert marker_repo_exists(live_app, user, password), (
+    assert marker_repo_exists(
-        f"{live_app}: ci-marker repo did not survive the upgrade (sqlite3 data lost)"
+        live_app, user, password
-    )
+    ), f"{live_app}: ci-marker repo did not survive the upgrade (sqlite3 data lost)"

tests/unit/test_discovery.py

@@ -106,7 +106,11 @@ def test_custom_tests_prefers_custom_and_warns_on_deprecated_aliases(tmp_path, m
 
     customs = discovery.custom_tests(fake_recipe, None)
 
-    assert [os.path.basename(path) for _, path in customs] == ["test_a.py", "test_b.py", "test_c.py"]
+    assert [os.path.basename(path) for _, path in customs] == [
+        "test_a.py",
+        "test_b.py",
+        "test_c.py",
+    ]
     err = capsys.readouterr().err
     assert "deprecated folder 'functional/'" in err
     assert "deprecated folder 'playwright/'" in err