slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #587 from AIOSAI/dev

Browse Source 
fix codeql #16
This commit is contained in:
AIPass
2026-05-15 14:53:41 -07:00
committed by GitHub
parent c798dc64e8 c114d4405f
commit 336ef63bf4
3 changed files with 169 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/drone/apps/handlers/git/commit_handler.py
+58
View File
@@ -18,6 +18,59 @@ from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
def _run_test_gate(repo_root: Path) -> dict | None:
"""Run pytest for changed branches. Returns error dict if tests fail, None if all pass."""
status_result = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
changed_branches: set[str] = set()
for line in status_result.stdout.splitlines():
if len(line) < 4:
continue
filepath = line[3:].split(" -> ")[-1]
parts = Path(filepath).parts
if len(parts) >= 3 and parts[0] == "src" and parts[1] == "aipass":
changed_branches.add(parts[2])
venv_python = repo_root / ".venv" / "bin" / "python"
python_bin = str(venv_python) if venv_python.exists() else "python3"
failed_branches: list[tuple[str, str]] = []
for branch_name in sorted(changed_branches):
test_dir = repo_root / "src" / "aipass" / branch_name / "tests"
if not test_dir.is_dir():
continue
try:
test_result = subprocess.run(
[python_bin, "-m", "pytest", str(test_dir), "--tb=short", "-q"],
capture_output=True,
text=True,
cwd=str(repo_root),
timeout=120,
)
except subprocess.TimeoutExpired:
logger.warning("pytest timed out for branch %s (120s limit)", branch_name)
failed_branches.append((branch_name, "pytest timed out (120s)"))
continue
if test_result.returncode != 0:
failed_branches.append((branch_name, test_result.stdout.strip()))
if not failed_branches:
return None
msg_parts = ["Test failures — fix before committing:"]
for bname, output in failed_branches:
msg_parts.append(f"\n--- {bname} ---\n{output}")
return {
"stdout": "",
"stderr": "\n".join(msg_parts),
"exit_code": 1,
}
def stage_branch_dir(branch_dir: Path, repo_root: Path | None = None) -> dict:
"""Stage all changes under branch_dir.
@@ -111,6 +164,11 @@ def commit_changes(
"stderr": f"Lint errors — fix before committing:\n{lint_check.stdout.strip()}",
"exit_code": 1,
}
test_gate_result = _run_test_gate(repo_root)
if test_gate_result is not None:
return test_gate_result
add_result = subprocess.run(
["git", "add", "-A"],
capture_output=True,
src/aipass/drone/apps/handlers/git/dev_pr_handler.py
+1
View File
@@ -90,6 +90,7 @@ def create_branch_pr(description: str, target_branch: str = "main") -> dict:
if "already exists" in stderr:
existing_url = ""
for line in stderr.splitlines():
# codeql[py/incomplete-url-substring-sanitization]
if "github.com" in line:
existing_url = line.strip()
break
src/aipass/drone/tests/test_git_access.py
+110 -1
View File
@@ -346,6 +346,7 @@ class TestCommitChanges:
mock_ruff_fix = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_format = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_gate = MagicMock(returncode=0, stdout="", stderr="")
mock_status = MagicMock(returncode=0, stdout=" M src/aipass/api/app.py\n", stderr="")
mock_add = MagicMock(returncode=0, stderr="")
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
mock_commit = MagicMock(returncode=0, stdout="[main def456] all commit", stderr="")
@@ -356,13 +357,121 @@ class TestCommitChanges:
patch("shutil.which", return_value="/usr/bin/ruff"),
patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[mock_ruff_fix, mock_ruff_format, mock_ruff_gate, mock_add, mock_diff, mock_commit],
side_effect=[
mock_ruff_fix,
mock_ruff_format,
mock_ruff_gate,
mock_status,
mock_add,
mock_diff,
mock_commit,
],
),
):
result = commit_changes("all commit", branch_dir=branch_dir, all_files=True)
assert result["exit_code"] == 0
def test_commit_all_blocks_on_test_failure(self, repo_dir: Path) -> None:
mock_ruff_fix = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_format = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_gate = MagicMock(returncode=0, stdout="", stderr="")
mock_status = MagicMock(
returncode=0,
stdout=" M src/aipass/drone/apps/handlers/git/commit_handler.py\n",
stderr="",
)
mock_pytest = MagicMock(
returncode=1,
stdout="FAILED test_foo.py::test_bar - assert 1 == 2\n1 failed",
stderr="",
)
test_dir = repo_dir / "src" / "aipass" / "drone" / "tests"
test_dir.mkdir(parents=True)
with (
patch("shutil.which", return_value="/usr/bin/ruff"),
patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[mock_ruff_fix, mock_ruff_format, mock_ruff_gate, mock_status, mock_pytest],
),
):
result = commit_changes("fail commit", all_files=True)
assert result["exit_code"] == 1
assert "Test failures" in result["stderr"]
assert "drone" in result["stderr"]
def test_commit_all_passes_with_green_tests(self, repo_dir: Path) -> None:
mock_ruff_fix = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_format = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_gate = MagicMock(returncode=0, stdout="", stderr="")
mock_status = MagicMock(
returncode=0,
stdout=" M src/aipass/drone/apps/handlers/git/commit_handler.py\n",
stderr="",
)
mock_pytest = MagicMock(returncode=0, stdout="3 passed", stderr="")
mock_add = MagicMock(returncode=0, stderr="")
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
mock_commit = MagicMock(returncode=0, stdout="[main abc999] green commit", stderr="")
test_dir = repo_dir / "src" / "aipass" / "drone" / "tests"
test_dir.mkdir(parents=True)
with (
patch("shutil.which", return_value="/usr/bin/ruff"),
patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[
mock_ruff_fix,
mock_ruff_format,
mock_ruff_gate,
mock_status,
mock_pytest,
mock_add,
mock_diff,
mock_commit,
],
),
):
result = commit_changes("green commit", all_files=True)
assert result["exit_code"] == 0
def test_commit_all_skips_branches_without_tests(self, repo_dir: Path) -> None:
mock_ruff_fix = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_format = MagicMock(returncode=0, stdout="", stderr="")
mock_ruff_gate = MagicMock(returncode=0, stdout="", stderr="")
mock_status = MagicMock(
returncode=0,
stdout=" M src/aipass/flow/apps/module.py\n",
stderr="",
)
mock_add = MagicMock(returncode=0, stderr="")
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
mock_commit = MagicMock(returncode=0, stdout="[main skip77] no tests", stderr="")
with (
patch("shutil.which", return_value="/usr/bin/ruff"),
patch(
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
side_effect=[
mock_ruff_fix,
mock_ruff_format,
mock_ruff_gate,
mock_status,
mock_add,
mock_diff,
mock_commit,
],
),
):
result = commit_changes("no tests commit", all_files=True)
assert result["exit_code"] == 0
def test_commit_os_error(self, repo_dir: Path) -> None:
mock_diff = MagicMock(returncode=1, stdout="", stderr="")