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>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
d832b353e4
commit
2d865f06cb
@@ -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
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user