feat: add private repository support for SWE-smith instances (#1343)
* feat: add private repository support for SWE-smith instances Co-authored-by: reisepass <1474408+reisepass@users.noreply.github.com> * refactor: replace SSH key auth with GITHUB_TOKEN for private SWE-smith repos --------- Co-authored-by: reisepass <1474408+reisepass@users.noreply.github.com> Co-authored-by: Rb <rubenwolff@gmail.com>
This commit is contained in:
Muhammed Karamuk
@@ -0,0 +1,8 @@
|
||||
# GitHub Personal Access Token
|
||||
# Used for GitHub API operations: querying repository metadata,
|
||||
# cloning private repos, and opening pull requests.
|
||||
#
|
||||
# Required scopes:
|
||||
# public_repo - for public repos (read metadata, open PRs)
|
||||
# repo - for private repos (all of the above + private repo access)
|
||||
GITHUB_TOKEN=
|
||||
@@ -191,7 +191,46 @@ class GithubRepoConfig(BaseModel):
|
||||
return _get_git_reset_commands(self.base_commit)
|
||||
|
||||
|
||||
RepoConfig = LocalRepoConfig | GithubRepoConfig | PreExistingRepoConfig
|
||||
class SWESmithRepoConfig(BaseModel):
|
||||
"""Repository config for SWE-Smith instances that handles targeted fetch
|
||||
from a GitHub mirror, authenticating via GITHUB_TOKEN when needed.
|
||||
"""
|
||||
|
||||
repo_name: str
|
||||
base_commit: str = Field(default="HEAD")
|
||||
mirror_url: str = ""
|
||||
"""HTTPS URL of the GitHub mirror to fetch the bug branch from."""
|
||||
|
||||
type: Literal["swesmith_preexisting"] = "swesmith_preexisting"
|
||||
"""Discriminator for (de)serialization/CLI. Do not change."""
|
||||
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
def copy(self, deployment: AbstractDeployment):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _get_url_with_token(url: str, token: str) -> str:
|
||||
if not token or not url:
|
||||
return url
|
||||
_, _, url_no_protocol = url.partition("://")
|
||||
return f"https://{token}@{url_no_protocol}"
|
||||
|
||||
def get_reset_commands(self) -> list[str]:
|
||||
if self.mirror_url:
|
||||
github_token = os.getenv("GITHUB_TOKEN", "")
|
||||
url = self._get_url_with_token(self.mirror_url, github_token)
|
||||
return [
|
||||
"git restore .",
|
||||
"git reset --hard",
|
||||
f"git fetch {shlex.quote(url)} {shlex.quote(self.base_commit)}",
|
||||
"git checkout FETCH_HEAD",
|
||||
"git clean -fdq",
|
||||
]
|
||||
return _get_git_reset_commands(self.base_commit)
|
||||
|
||||
|
||||
RepoConfig = LocalRepoConfig | GithubRepoConfig | PreExistingRepoConfig | SWESmithRepoConfig
|
||||
|
||||
|
||||
def repo_from_simplified_input(
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
from abc import ABC, abstractmethod
|
||||
@@ -19,9 +20,10 @@ from sweagent.agent.problem_statement import (
|
||||
SWEBenchMultimodalProblemStatement,
|
||||
TextProblemStatement,
|
||||
)
|
||||
from sweagent.environment.repo import GithubRepoConfig, LocalRepoConfig, PreExistingRepoConfig
|
||||
from sweagent.environment.repo import GithubRepoConfig, LocalRepoConfig, PreExistingRepoConfig, SWESmithRepoConfig
|
||||
from sweagent.environment.swe_env import EnvironmentConfig
|
||||
from sweagent.utils.files import load_file
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
from sweagent.utils.log import get_logger
|
||||
|
||||
logger = get_logger("swea-config", emoji="🔧")
|
||||
@@ -391,22 +393,50 @@ class SWESmithInstances(BaseModel, AbstractInstanceSource):
|
||||
"""Discriminator for (de)serialization/CLI. Do not change."""
|
||||
|
||||
def get_instance_configs(self) -> list[BatchInstance]:
|
||||
def convert_instance_dict(instance_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
instance_dict["id"] = instance_dict["instance_id"]
|
||||
# todo: The base_commit is currently incorrect
|
||||
instance_dict["base_commit"] = instance_dict["id"]
|
||||
instance_dict["problem_statement"] = instance_dict.get("problem_statement", "")
|
||||
instance_dict["repo_name"] = "testbed"
|
||||
instance_dict["extra_fields"] = {"fail_to_pass": instance_dict["FAIL_TO_PASS"]}
|
||||
return instance_dict
|
||||
github_token = os.getenv("GITHUB_TOKEN", "")
|
||||
|
||||
instance_dicts = load_file(self.path)
|
||||
instances = [
|
||||
SimpleBatchInstance.model_validate(convert_instance_dict(instance_dict)).to_full_batch_instance(
|
||||
self.deployment
|
||||
instances = []
|
||||
|
||||
for instance_dict in instance_dicts:
|
||||
deployment = self.deployment.model_copy(deep=True)
|
||||
deployment.image = instance_dict["image_name"] # type: ignore
|
||||
|
||||
if isinstance(deployment, DockerDeploymentConfig) and deployment.python_standalone_dir is None:
|
||||
deployment.python_standalone_dir = "/root" # type: ignore
|
||||
|
||||
instance_id = instance_dict["instance_id"]
|
||||
repo_field = instance_dict.get("repo", "")
|
||||
|
||||
mirror_url = ""
|
||||
if repo_field and _is_repo_private(repo_field, github_token):
|
||||
if not github_token:
|
||||
msg = (
|
||||
f"Repo '{repo_field}' appears to be private but GITHUB_TOKEN is not set. "
|
||||
"Set GITHUB_TOKEN with 'repo' scope to access private repositories."
|
||||
)
|
||||
raise ValueError(msg)
|
||||
mirror_url = f"https://github.com/{repo_field}.git"
|
||||
|
||||
repo = SWESmithRepoConfig(
|
||||
repo_name="testbed",
|
||||
base_commit=instance_id,
|
||||
mirror_url=mirror_url,
|
||||
)
|
||||
for instance_dict in instance_dicts
|
||||
]
|
||||
|
||||
problem_statement = TextProblemStatement(
|
||||
text=instance_dict.get("problem_statement", ""),
|
||||
id=instance_id,
|
||||
extra_fields={"fail_to_pass": instance_dict.get("FAIL_TO_PASS", [])},
|
||||
)
|
||||
|
||||
instances.append(
|
||||
BatchInstance(
|
||||
env=EnvironmentConfig(deployment=deployment, repo=repo),
|
||||
problem_statement=problem_statement,
|
||||
)
|
||||
)
|
||||
|
||||
return _filter_batch_items(instances, filter_=self.filter, slice_=self.slice, shuffle=self.shuffle)
|
||||
|
||||
@property
|
||||
|
||||
@@ -1,7 +1,16 @@
|
||||
import json
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
from ghapi.all import GhApi
|
||||
|
||||
from sweagent.utils.log import get_logger
|
||||
|
||||
_logger = get_logger("swea-github", emoji="🔧")
|
||||
|
||||
_repo_privacy_cache: dict[str, bool] = {}
|
||||
|
||||
GITHUB_ISSUE_URL_PATTERN = re.compile(r"github\.com\/(.*?)\/(.*?)\/issues\/(\d+)")
|
||||
|
||||
|
||||
@@ -116,3 +125,31 @@ def _get_associated_commit_urls(org: str, repo: str, issue_number: str, *, token
|
||||
if f"fixes #{issue_number}" in message.lower() or f"closes #{issue_number}" in message.lower():
|
||||
commit_urls.append(commit.html_url)
|
||||
return commit_urls
|
||||
|
||||
|
||||
def _is_repo_private(owner_repo: str, token: str) -> bool:
|
||||
"""Check if a GitHub repository is private via the GitHub API.
|
||||
|
||||
Returns True if the repo is private or if a 404 is returned (GitHub returns
|
||||
404 for private repos when the token lacks access). Any other HTTP or
|
||||
network error is raised so callers can handle it explicitly.
|
||||
"""
|
||||
if owner_repo in _repo_privacy_cache:
|
||||
return _repo_privacy_cache[owner_repo]
|
||||
url = f"https://api.github.com/repos/{owner_repo}"
|
||||
headers = {"User-Agent": "sweagent"}
|
||||
if token:
|
||||
headers["Authorization"] = f"token {token}"
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
data = json.loads(resp.read())
|
||||
private = data.get("private", False)
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 404:
|
||||
_logger.warning("Repo '%s' returned 404 — assuming private", owner_repo)
|
||||
private = True
|
||||
else:
|
||||
raise
|
||||
_repo_privacy_cache[owner_repo] = private
|
||||
return private
|
||||
|
||||
@@ -0,0 +1,210 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from sweagent.environment.repo import SWESmithRepoConfig
|
||||
from sweagent.run.batch_instances import BatchInstance, SWESmithInstances
|
||||
|
||||
# ── SWESmithRepoConfig.get_reset_commands ──
|
||||
|
||||
|
||||
class TestSWESmithRepoConfigGetResetCommands:
|
||||
def test_no_mirror(self):
|
||||
"""Falls back to standard git reset commands."""
|
||||
repo = SWESmithRepoConfig(repo_name="testbed", base_commit="abc123")
|
||||
cmds = repo.get_reset_commands()
|
||||
assert any("git checkout" in c and "abc123" in c for c in cmds)
|
||||
assert any("git fetch" in c for c in cmds)
|
||||
|
||||
def test_with_mirror_and_token(self):
|
||||
"""Fetches from mirror URL with token embedded."""
|
||||
repo = SWESmithRepoConfig(
|
||||
repo_name="testbed",
|
||||
base_commit="branch-id",
|
||||
mirror_url="https://github.com/org/repo.git",
|
||||
)
|
||||
with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "ghp_test123"}):
|
||||
cmds = repo.get_reset_commands()
|
||||
assert any("git fetch" in c and "ghp_test123@github.com/org/repo.git" in c for c in cmds)
|
||||
assert any("git checkout FETCH_HEAD" in c for c in cmds)
|
||||
assert not any(c == "git fetch" for c in cmds)
|
||||
|
||||
def test_with_mirror_no_token(self):
|
||||
"""Mirror URL but no token — fetches with bare URL."""
|
||||
repo = SWESmithRepoConfig(
|
||||
repo_name="testbed",
|
||||
base_commit="branch-id",
|
||||
mirror_url="https://github.com/org/repo.git",
|
||||
)
|
||||
with mock.patch.dict("os.environ", {}, clear=True):
|
||||
cmds = repo.get_reset_commands()
|
||||
assert any("git fetch" in c and "https://github.com/org/repo.git" in c for c in cmds)
|
||||
assert not any("@" in c for c in cmds if "git fetch" in c)
|
||||
|
||||
|
||||
# ── SWESmithRepoConfig._get_url_with_token ──
|
||||
|
||||
|
||||
class TestGetUrlWithToken:
|
||||
def test_prepends_token(self):
|
||||
url = SWESmithRepoConfig._get_url_with_token("https://github.com/org/repo.git", "ghp_abc")
|
||||
assert url == "https://ghp_abc@github.com/org/repo.git"
|
||||
|
||||
def test_empty_token(self):
|
||||
url = SWESmithRepoConfig._get_url_with_token("https://github.com/org/repo.git", "")
|
||||
assert url == "https://github.com/org/repo.git"
|
||||
|
||||
def test_empty_url(self):
|
||||
url = SWESmithRepoConfig._get_url_with_token("", "ghp_abc")
|
||||
assert url == ""
|
||||
|
||||
|
||||
# ── _is_repo_private ──
|
||||
|
||||
|
||||
class TestIsRepoPrivate:
|
||||
def setup_method(self):
|
||||
from sweagent.utils.github import _repo_privacy_cache
|
||||
|
||||
_repo_privacy_cache.clear()
|
||||
|
||||
@patch("sweagent.utils.github.urllib.request.urlopen")
|
||||
def test_public_repo(self, mock_urlopen):
|
||||
mock_resp = Mock()
|
||||
mock_resp.read.return_value = json.dumps({"private": False}).encode()
|
||||
mock_resp.__enter__ = Mock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_resp
|
||||
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
|
||||
assert _is_repo_private("org/repo", "fake-token") is False
|
||||
|
||||
@patch("sweagent.utils.github.urllib.request.urlopen")
|
||||
def test_private_repo(self, mock_urlopen):
|
||||
mock_resp = Mock()
|
||||
mock_resp.read.return_value = json.dumps({"private": True}).encode()
|
||||
mock_resp.__enter__ = Mock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_resp
|
||||
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
|
||||
assert _is_repo_private("org/repo", "fake-token") is True
|
||||
|
||||
@patch("sweagent.utils.github.urllib.request.urlopen")
|
||||
def test_404_assumes_private(self, mock_urlopen):
|
||||
mock_urlopen.side_effect = urllib.error.HTTPError(
|
||||
url="",
|
||||
code=404,
|
||||
msg="Not Found",
|
||||
hdrs=None,
|
||||
fp=None, # type: ignore
|
||||
)
|
||||
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
|
||||
assert _is_repo_private("org/repo", "fake-token") is True
|
||||
|
||||
@patch("sweagent.utils.github.urllib.request.urlopen")
|
||||
def test_other_http_error_raises(self, mock_urlopen):
|
||||
mock_urlopen.side_effect = urllib.error.HTTPError(
|
||||
url="",
|
||||
code=500,
|
||||
msg="Server Error",
|
||||
hdrs=None,
|
||||
fp=None, # type: ignore
|
||||
)
|
||||
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
|
||||
with pytest.raises(urllib.error.HTTPError):
|
||||
_is_repo_private("org/repo", "fake-token")
|
||||
|
||||
@patch("sweagent.utils.github.urllib.request.urlopen")
|
||||
def test_caching(self, mock_urlopen):
|
||||
mock_resp = Mock()
|
||||
mock_resp.read.return_value = json.dumps({"private": False}).encode()
|
||||
mock_resp.__enter__ = Mock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_resp
|
||||
|
||||
from sweagent.utils.github import _is_repo_private
|
||||
|
||||
_is_repo_private("org/cached-repo", "token")
|
||||
_is_repo_private("org/cached-repo", "token")
|
||||
assert mock_urlopen.call_count == 1
|
||||
|
||||
|
||||
# ── SWESmithInstances.get_instance_configs ──
|
||||
|
||||
|
||||
class TestSWESmithInstancesGetInstanceConfigs:
|
||||
@staticmethod
|
||||
def _make_instance_file(tmp_path: Path, instances: list[dict]) -> Path:
|
||||
p = tmp_path / "instances.json"
|
||||
p.write_text(json.dumps(instances))
|
||||
return p
|
||||
|
||||
@staticmethod
|
||||
def _sample_instance(instance_id: str = "org__repo.abc123__test_1", repo: str = "org/repo") -> dict:
|
||||
return {
|
||||
"instance_id": instance_id,
|
||||
"image_name": "swebench/swesmith.x86_64.org_1776_repo.abc123",
|
||||
"repo": repo,
|
||||
"problem_statement": "Fix the bug",
|
||||
"FAIL_TO_PASS": ["test_foo.py::test_bar"],
|
||||
}
|
||||
|
||||
@patch("sweagent.run.batch_instances._is_repo_private", return_value=False)
|
||||
def test_public_repo(self, mock_private, tmp_path):
|
||||
path = self._make_instance_file(tmp_path, [self._sample_instance()])
|
||||
config = SWESmithInstances(path=path)
|
||||
instances = config.get_instance_configs()
|
||||
|
||||
assert len(instances) == 1
|
||||
inst = instances[0]
|
||||
assert isinstance(inst, BatchInstance)
|
||||
assert inst.env.repo.repo_name == "testbed"
|
||||
assert inst.env.repo.mirror_url == ""
|
||||
assert inst.env.deployment.image == "swebench/swesmith.x86_64.org_1776_repo.abc123"
|
||||
|
||||
@patch("sweagent.run.batch_instances._is_repo_private", return_value=True)
|
||||
def test_private_repo(self, mock_private, tmp_path):
|
||||
path = self._make_instance_file(tmp_path, [self._sample_instance()])
|
||||
config = SWESmithInstances(path=path)
|
||||
|
||||
with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "ghp_fake"}):
|
||||
instances = config.get_instance_configs()
|
||||
|
||||
assert len(instances) == 1
|
||||
inst = instances[0]
|
||||
assert inst.env.repo.mirror_url == "https://github.com/org/repo.git"
|
||||
|
||||
@patch("sweagent.run.batch_instances._is_repo_private", return_value=True)
|
||||
def test_private_repo_no_token_raises(self, mock_private, tmp_path):
|
||||
path = self._make_instance_file(tmp_path, [self._sample_instance()])
|
||||
config = SWESmithInstances(path=path)
|
||||
|
||||
with mock.patch.dict("os.environ", {}, clear=True):
|
||||
with pytest.raises(ValueError, match="GITHUB_TOKEN is not set"):
|
||||
config.get_instance_configs()
|
||||
|
||||
@patch("sweagent.run.batch_instances._is_repo_private", return_value=False)
|
||||
def test_filter_and_slice(self, mock_private, tmp_path):
|
||||
instances_data = [
|
||||
self._sample_instance(instance_id="org__repo.abc__test_1"),
|
||||
self._sample_instance(instance_id="org__repo.abc__test_2"),
|
||||
self._sample_instance(instance_id="org__repo.abc__test_3"),
|
||||
]
|
||||
path = self._make_instance_file(tmp_path, instances_data)
|
||||
config = SWESmithInstances(path=path, filter=".*test_[12]", slice="0:1")
|
||||
instances = config.get_instance_configs()
|
||||
|
||||
assert len(instances) == 1
|
||||
Reference in New Issue
Block a user