diff --git a/bridge/bridge.py b/bridge/bridge.py index 33218dc..bea79bc 100644 --- a/bridge/bridge.py +++ b/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): diff --git a/runner/harness/manifest.py b/runner/harness/manifest.py index 106bd09..ea71cb2 100644 --- a/runner/harness/manifest.py +++ b/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 diff --git a/tests/gitea/custom/test_admin_api.py b/tests/gitea/custom/test_admin_api.py index d2a7b84..3050f75 100644 --- a/tests/gitea/custom/test_admin_api.py +++ b/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", - user=adm_user, password=adm_pass, + live_app, + 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", - user=adm_user, password=adm_pass) + _api( + live_app, f"/admin/users/{test_user}", method="DELETE", user=adm_user, password=adm_pass + ) diff --git a/tests/gitea/custom/test_git_push.py b/tests/gitea/custom/test_git_push.py index 87d69b2..d11cdbe 100644 --- a/tests/gitea/custom/test_git_push.py +++ b/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", - user=user, password=password, + live_app, + 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, ( - f"Unexpected commit message: {commit_msg!r}" - ) + assert "automated push test" in commit_msg, f"Unexpected commit message: {commit_msg!r}" finally: shutil.rmtree(tmpdir, ignore_errors=True) # removes parent + repo subdir # 6. Cleanup — delete the test repo diff --git a/tests/gitea/custom/test_lfs_roundtrip.py b/tests/gitea/custom/test_lfs_roundtrip.py index 959764c..5f3aa24 100644 --- a/tests/gitea/custom/test_lfs_roundtrip.py +++ b/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, ( - f"LFS round-trip OID mismatch: expected {expected_oid}, got sha256:{fetched_sha256}" - ) + assert ( + 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, ( - "Could not read LFS_JWT_SECRET from /etc/gitea/app.ini before restart" - ) + assert ( + 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", - live_app.replace(".", "_") + "_app"], - capture_output=True, timeout=120, + ["docker", "service", "update", "--force", live_app.replace(".", "_") + "_app"], + capture_output=True, + 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, ( - "LFS object corrupted after restart — JWT secret may have changed" - ) + assert ( + 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) diff --git a/tests/gitea/ops.py b/tests/gitea/ops.py index 7d8b1ec..c65f915 100644 --- a/tests/gitea/ops.py +++ b/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", - "--username", _ADMIN_USER, - "--password", password, - "--email", _ADMIN_EMAIL, + "gitea", + "admin", + "user", + "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", - "--username", _ADMIN_USER, - "--password", password, + "gitea", + "admin", + "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", - user=user, password=password, + domain, + 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( - domain, f"/repos/{user}/{_MARKER_REPO}", user=user, password=password - ) + status, _ = _gitea_api(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( - f"No admin creds for {domain} — was ops.pre_install called for this run?" - ) + raise RuntimeError(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), ( - f"pre_restore: {_MARKER_REPO} still present after delete — divergence did not take" - ) + assert not marker_repo_exists( + 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) diff --git a/tests/gitea/recipe_meta.py b/tests/gitea/recipe_meta.py index f967205..d7b9eb8 100644 --- a/tests/gitea/recipe_meta.py +++ b/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) diff --git a/tests/gitea/test_backup.py b/tests/gitea/test_backup.py index 229ff72..57b2f16 100644 --- a/tests/gitea/test_backup.py +++ b/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), ( - f"{live_app}: ci-marker repo is not present at backup time (backup would capture empty state)" - ) + assert marker_repo_exists( + live_app, user, password + ), f"{live_app}: ci-marker repo is not present at backup time (backup would capture empty state)" diff --git a/tests/gitea/test_install.py b/tests/gitea/test_install.py index ca3559b..296ccf5 100644 --- a/tests/gitea/test_install.py +++ b/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(), ( - "Sign-in page did not render expected gitea content" - ) + assert ( + "gitea" in content.lower() or "sign in" in content.lower() + ), "Sign-in page did not render expected gitea content" finally: browser.close() diff --git a/tests/gitea/test_upgrade.py b/tests/gitea/test_upgrade.py index 5a7589c..8f2f46e 100644 --- a/tests/gitea/test_upgrade.py +++ b/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), ( - f"{live_app}: ci-marker repo did not survive the upgrade (sqlite3 data lost)" - ) + assert marker_repo_exists( + live_app, user, password + ), f"{live_app}: ci-marker repo did not survive the upgrade (sqlite3 data lost)" diff --git a/tests/unit/test_discovery.py b/tests/unit/test_discovery.py index b2baed0..4934d49 100644 --- a/tests/unit/test_discovery.py +++ b/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