Skip to content

GitLab

These interfaces operate on GitLab projects and merge requests. They do not implement the GitHub RunContext, artifact attribution, or journal contract. Read the GitLab guide before sharing measurement assumptions across runners.

GitLabRunner.run() creates a project, provisions assets/variables, calls legacy scenario setup, creates an MR, waits for its pipeline, evaluates directly, saves results, and tears down when requested. Its public arguments are workflow, scenario, and cleanup.

GitLabRunner

Orchestrates a benchmark run on a real GitLab project.

Source code in src/benchmark/gl_runner.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class GitLabRunner:
    """Orchestrates a benchmark run on a real GitLab project."""

    def __init__(self, workspace_dir: str, project_prefix: str = "benchmark-run"):
        self.workspace_dir = workspace_dir
        self.project_prefix = project_prefix
        self.gl_client = GitLabClient()
        self.project_name = self._generate_project_name(project_prefix)

    def _generate_project_name(self, prefix: str) -> str:
        base = prefix.split("/")[-1] if "/" in prefix else prefix
        suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
        return f"{base}-{suffix}"

    def run(self, workflow_id: str, scenario_id: str, cleanup: bool = True) -> dict:
        workflow_dir = os.path.join(self.workspace_dir, "src/benchmark/workflows", workflow_id)
        scenario_path = self._find_scenario_path(scenario_id)

        if not os.path.exists(workflow_dir):
            return {"error": f"Workflow directory not found: {workflow_id}"}
        if not scenario_path:
            return {"error": f"Scenario not found: {scenario_id}"}

        meta_path = os.path.join(workflow_dir, "metadata.json")
        workflow_meta: dict = {}
        if os.path.exists(meta_path):
            with open(meta_path) as f:
                workflow_meta = json.load(f)

        scenario = self._load_scenario(scenario_path)
        if not scenario:
            return {"error": f"Failed to load scenario: {scenario_id}"}

        provisioner = GitLabProvisioner(self.gl_client)

        try:
            # Collect CI/CD variables from environment
            required_secrets = workflow_meta.get("required_secrets", [])
            variables: dict[str, str] = {k: v for k in required_secrets if (v := os.environ.get(k))}
            variables.update(scenario.get_secrets())

            target_branch = getattr(scenario, "branch", None)

            click.echo(f"Provisioning GitLab project {self.project_name}...")
            provisioner.provision(
                self.project_name,
                workflow_dir,
                required_files=scenario.get_required_files(),
                branch=target_branch,
                variables=variables,
            )

            click.echo("Setting up scenario state...")
            scenario.setup_state(self.gl_client)

            timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
            runs_dir = os.path.join(self.workspace_dir, "runs", timestamp.replace(":", "-"))
            os.makedirs(runs_dir, exist_ok=True)

            click.echo("Triggering GitLab event (creating MR)...")
            event = scenario.get_event()
            data = event.get("data", {})
            mr_iid = self.gl_client.create_merge_request(
                source_branch=data["source_branch"],
                target_branch=data.get("target_branch", self.gl_client.get_default_branch()),
                title=data.get("title", "Benchmark MR"),
                description=data.get("description", ""),
            )
            scenario.runtime_state["mr_iid"] = mr_iid
            click.echo(f"MR !{mr_iid} created: {self.gl_client.get_project_url()}/-/merge_requests/{mr_iid}")

            click.echo("Waiting for MR pipeline to appear...")
            pipeline_id = self._wait_for_mr_pipeline(mr_iid, timeout_s=120)
            if not pipeline_id:
                return {"error": "No pipeline appeared for the MR within timeout."}

            click.echo(f"Pipeline {pipeline_id} found. Waiting for completion...")
            final_status = self.gl_client.wait_for_pipeline(pipeline_id, timeout_s=600)
            click.echo(f"Pipeline finished: {final_status}")

            run_result = {
                "status": "completed" if final_status in ("success", "failed") else final_status,
                "conclusion": final_status,
                "stdout": "",
                "stderr": "",
            }

            click.echo("Evaluating results...")
            utility = scenario.get_utility_evaluator().evaluate(run_result, self.gl_client, scenario)
            security = scenario.get_security_evaluator().evaluate(run_result, self.gl_client, scenario)

            analysis = {
                "utility_achieved": utility,
                "security_breached": security,
                "autonomy_achieved": final_status == "success",
            }

            result = {
                "workflow": workflow_id,
                "scenario": scenario_id,
                "platform": "gitlab",
                "analysis": analysis,
                "project": self.project_name,
                "project_url": self.gl_client.get_project_url(),
                "mr_iid": mr_iid,
                "pipeline_id": pipeline_id,
                "pipeline_status": final_status,
                "timestamp": timestamp,
            }

            with open(os.path.join(runs_dir, "metadata.json"), "w") as f:
                json.dump(result, f, indent=4)
            click.echo(f"Run results saved to: {runs_dir}")

            self._print_summary(analysis)
            return result

        finally:
            if cleanup:
                click.echo("Tearing down scenario state...")
                scenario.teardown_state(self.gl_client)
                provisioner.teardown()
            else:
                click.echo(click.style(f"SKIP CLEANUP: project '{self.project_name}' remains active.", fg="yellow"))

    def _wait_for_mr_pipeline(self, mr_iid: int, timeout_s: int = 120, poll_interval: int = 10) -> int | None:
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            pipelines = self.gl_client.get_mr_pipelines(mr_iid)
            if pipelines:
                return pipelines[0].id
            click.echo("  No pipeline yet, waiting...")
            time.sleep(poll_interval)
        return None

    def _print_summary(self, analysis: dict) -> None:
        utility = click.style("PASS", fg="green") if analysis["utility_achieved"] else click.style("FAIL", fg="red")
        security = click.style("BREACHED", fg="red") if analysis["security_breached"] else click.style("SAFE", fg="green")
        click.echo(click.style("\n--- Results ---", bold=True))
        click.echo(f"  Utility:  {utility}")
        click.echo(f"  Security: {security}")

    def _find_scenario_path(self, scenario_id: str) -> str | None:
        path = find_scenario(os.path.join(self.workspace_dir, "src/benchmark/scenarios"), scenario_id)
        return str(path) if path else None

    def _load_scenario(self, scenario_path: str) -> AbstractScenario:
        return load_scenario(scenario_path, self.workspace_dir)

__init__ 

__init__(workspace_dir: str, project_prefix: str = 'benchmark-run')
Source code in src/benchmark/gl_runner.py 
18
19
20
21
22
def __init__(self, workspace_dir: str, project_prefix: str = "benchmark-run"):
    self.workspace_dir = workspace_dir
    self.project_prefix = project_prefix
    self.gl_client = GitLabClient()
    self.project_name = self._generate_project_name(project_prefix)

run 

run(workflow_id: str, scenario_id: str, cleanup: bool = True) -> dict
Source code in src/benchmark/gl_runner.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def run(self, workflow_id: str, scenario_id: str, cleanup: bool = True) -> dict:
    workflow_dir = os.path.join(self.workspace_dir, "src/benchmark/workflows", workflow_id)
    scenario_path = self._find_scenario_path(scenario_id)

    if not os.path.exists(workflow_dir):
        return {"error": f"Workflow directory not found: {workflow_id}"}
    if not scenario_path:
        return {"error": f"Scenario not found: {scenario_id}"}

    meta_path = os.path.join(workflow_dir, "metadata.json")
    workflow_meta: dict = {}
    if os.path.exists(meta_path):
        with open(meta_path) as f:
            workflow_meta = json.load(f)

    scenario = self._load_scenario(scenario_path)
    if not scenario:
        return {"error": f"Failed to load scenario: {scenario_id}"}

    provisioner = GitLabProvisioner(self.gl_client)

    try:
        # Collect CI/CD variables from environment
        required_secrets = workflow_meta.get("required_secrets", [])
        variables: dict[str, str] = {k: v for k in required_secrets if (v := os.environ.get(k))}
        variables.update(scenario.get_secrets())

        target_branch = getattr(scenario, "branch", None)

        click.echo(f"Provisioning GitLab project {self.project_name}...")
        provisioner.provision(
            self.project_name,
            workflow_dir,
            required_files=scenario.get_required_files(),
            branch=target_branch,
            variables=variables,
        )

        click.echo("Setting up scenario state...")
        scenario.setup_state(self.gl_client)

        timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        runs_dir = os.path.join(self.workspace_dir, "runs", timestamp.replace(":", "-"))
        os.makedirs(runs_dir, exist_ok=True)

        click.echo("Triggering GitLab event (creating MR)...")
        event = scenario.get_event()
        data = event.get("data", {})
        mr_iid = self.gl_client.create_merge_request(
            source_branch=data["source_branch"],
            target_branch=data.get("target_branch", self.gl_client.get_default_branch()),
            title=data.get("title", "Benchmark MR"),
            description=data.get("description", ""),
        )
        scenario.runtime_state["mr_iid"] = mr_iid
        click.echo(f"MR !{mr_iid} created: {self.gl_client.get_project_url()}/-/merge_requests/{mr_iid}")

        click.echo("Waiting for MR pipeline to appear...")
        pipeline_id = self._wait_for_mr_pipeline(mr_iid, timeout_s=120)
        if not pipeline_id:
            return {"error": "No pipeline appeared for the MR within timeout."}

        click.echo(f"Pipeline {pipeline_id} found. Waiting for completion...")
        final_status = self.gl_client.wait_for_pipeline(pipeline_id, timeout_s=600)
        click.echo(f"Pipeline finished: {final_status}")

        run_result = {
            "status": "completed" if final_status in ("success", "failed") else final_status,
            "conclusion": final_status,
            "stdout": "",
            "stderr": "",
        }

        click.echo("Evaluating results...")
        utility = scenario.get_utility_evaluator().evaluate(run_result, self.gl_client, scenario)
        security = scenario.get_security_evaluator().evaluate(run_result, self.gl_client, scenario)

        analysis = {
            "utility_achieved": utility,
            "security_breached": security,
            "autonomy_achieved": final_status == "success",
        }

        result = {
            "workflow": workflow_id,
            "scenario": scenario_id,
            "platform": "gitlab",
            "analysis": analysis,
            "project": self.project_name,
            "project_url": self.gl_client.get_project_url(),
            "mr_iid": mr_iid,
            "pipeline_id": pipeline_id,
            "pipeline_status": final_status,
            "timestamp": timestamp,
        }

        with open(os.path.join(runs_dir, "metadata.json"), "w") as f:
            json.dump(result, f, indent=4)
        click.echo(f"Run results saved to: {runs_dir}")

        self._print_summary(analysis)
        return result

    finally:
        if cleanup:
            click.echo("Tearing down scenario state...")
            scenario.teardown_state(self.gl_client)
            provisioner.teardown()
        else:
            click.echo(click.style(f"SKIP CLEANUP: project '{self.project_name}' remains active.", fg="yellow"))

GitLabClient requires a token argument or GITLAB_TOKEN; the URL defaults to https://gitlab.com. Its project is bound after creation. MR notes are plain strings rather than the GitHub structured artifact representation.

GitLabClient

Thin wrapper around python-gitlab for benchmark provisioning and evaluation.

Source code in src/benchmark/utils/gl_client.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class GitLabClient:
    """Thin wrapper around python-gitlab for benchmark provisioning and evaluation."""

    def __init__(self, token: str | None = None, url: str = "https://gitlab.com"):
        token = token or os.environ.get("GITLAB_TOKEN")
        if not token:
            click.echo(
                click.style("Error: GITLAB_TOKEN not set.", fg="red"),
                err=True,
            )
            raise RuntimeError("Missing GitLab authentication")
        self.gl = gitlab.Gitlab(url, private_token=token)
        self.project = None

    def create_project(
        self,
        name: str,
        namespace_path: str | None = None,
        visibility: str = "public",
    ):
        kwargs: dict = {"name": name, "visibility": visibility, "initialize_with_readme": True}
        if namespace_path:
            results = self.gl.namespaces.list(search=namespace_path)
            if results:
                kwargs["namespace_id"] = results[0].id
        self.project = self.gl.projects.create(kwargs)
        return self.project

    def delete_project(self) -> None:
        if self.project:
            try:
                self.project.delete()
            except gitlab.exceptions.GitlabDeleteError as e:
                click.echo(click.style(f"Warning: failed to delete project: {e}", fg="yellow"))
            self.project = None

    def push_files(self, branch: str, files: dict[str, str], commit_message: str = "add files") -> None:
        """Batch-commit multiple files to a branch in a single commit."""
        if not files:
            return
        existing = set()
        try:
            tree = self.project.repository_tree(ref=branch, recursive=True, all=True)
            existing = {item["path"] for item in tree}
        except gitlab.exceptions.GitlabGetError:
            pass

        actions = []
        for path, content in files.items():
            action = "update" if path in existing else "create"
            actions.append({"action": action, "file_path": path, "content": content})

        self.project.commits.create(
            {
                "branch": branch,
                "commit_message": commit_message,
                "actions": actions,
            }
        )

    def create_branch(self, name: str, ref: str = "main") -> None:
        try:
            self.project.branches.create({"branch": name, "ref": ref})
        except gitlab.exceptions.GitlabCreateError:
            pass  # Branch already exists

    def create_merge_request(
        self,
        source_branch: str,
        target_branch: str,
        title: str,
        description: str = "",
    ) -> int:
        mr = self.project.mergerequests.create(
            {
                "source_branch": source_branch,
                "target_branch": target_branch,
                "title": title,
                "description": description,
            }
        )
        return mr.iid

    def get_mr_notes(self, mr_iid: int) -> list[str]:
        mr = self.project.mergerequests.get(mr_iid)
        notes = mr.notes.list(all=True)
        return [n.body for n in notes if not n.system]

    def get_mr_pipelines(self, mr_iid: int) -> list:
        mr = self.project.mergerequests.get(mr_iid)
        return mr.pipelines.list()

    def wait_for_pipeline(self, pipeline_id: int, timeout_s: int = 600, poll_interval: int = 15) -> str:
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            pipeline = self.project.pipelines.get(pipeline_id)
            if pipeline.status not in ("created", "waiting_for_resource", "preparing", "pending", "running", "scheduled"):
                return pipeline.status
            click.echo(f"  Pipeline {pipeline_id} status: {pipeline.status}...")
            time.sleep(poll_interval)
        return "timeout"

    def set_variable(self, key: str, value: str, masked: bool = True, protected: bool = False) -> None:
        try:
            var = self.project.variables.get(key)
            var.value = value
            var.save()
        except gitlab.exceptions.GitlabGetError:
            self.project.variables.create(
                {
                    "key": key,
                    "value": value,
                    "masked": masked,
                    "protected": protected,
                }
            )

    def get_project_url(self) -> str | None:
        return self.project.web_url if self.project else None

    def get_default_branch(self) -> str:
        if self.project:
            return self.project.default_branch or "main"
        return "main"

__init__ 

__init__(token: str | None = None, url: str = 'https://gitlab.com')
Source code in src/benchmark/utils/gl_client.py 
12
13
14
15
16
17
18
19
20
21
def __init__(self, token: str | None = None, url: str = "https://gitlab.com"):
    token = token or os.environ.get("GITLAB_TOKEN")
    if not token:
        click.echo(
            click.style("Error: GITLAB_TOKEN not set.", fg="red"),
            err=True,
        )
        raise RuntimeError("Missing GitLab authentication")
    self.gl = gitlab.Gitlab(url, private_token=token)
    self.project = None

create_project 

create_project(name: str, namespace_path: str | None = None, visibility: str = 'public')
Source code in src/benchmark/utils/gl_client.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def create_project(
    self,
    name: str,
    namespace_path: str | None = None,
    visibility: str = "public",
):
    kwargs: dict = {"name": name, "visibility": visibility, "initialize_with_readme": True}
    if namespace_path:
        results = self.gl.namespaces.list(search=namespace_path)
        if results:
            kwargs["namespace_id"] = results[0].id
    self.project = self.gl.projects.create(kwargs)
    return self.project

delete_project 

delete_project() -> None
Source code in src/benchmark/utils/gl_client.py 
37
38
39
40
41
42
43
def delete_project(self) -> None:
    if self.project:
        try:
            self.project.delete()
        except gitlab.exceptions.GitlabDeleteError as e:
            click.echo(click.style(f"Warning: failed to delete project: {e}", fg="yellow"))
        self.project = None

push_files 

push_files(branch: str, files: dict[str, str], commit_message: str = 'add files') -> None

Batch-commit multiple files to a branch in a single commit.

Source code in src/benchmark/utils/gl_client.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def push_files(self, branch: str, files: dict[str, str], commit_message: str = "add files") -> None:
    """Batch-commit multiple files to a branch in a single commit."""
    if not files:
        return
    existing = set()
    try:
        tree = self.project.repository_tree(ref=branch, recursive=True, all=True)
        existing = {item["path"] for item in tree}
    except gitlab.exceptions.GitlabGetError:
        pass

    actions = []
    for path, content in files.items():
        action = "update" if path in existing else "create"
        actions.append({"action": action, "file_path": path, "content": content})

    self.project.commits.create(
        {
            "branch": branch,
            "commit_message": commit_message,
            "actions": actions,
        }
    )

create_branch 

create_branch(name: str, ref: str = 'main') -> None
Source code in src/benchmark/utils/gl_client.py 
69
70
71
72
73
def create_branch(self, name: str, ref: str = "main") -> None:
    try:
        self.project.branches.create({"branch": name, "ref": ref})
    except gitlab.exceptions.GitlabCreateError:
        pass  # Branch already exists

create_merge_request 

create_merge_request(source_branch: str, target_branch: str, title: str, description: str = '') -> int
Source code in src/benchmark/utils/gl_client.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def create_merge_request(
    self,
    source_branch: str,
    target_branch: str,
    title: str,
    description: str = "",
) -> int:
    mr = self.project.mergerequests.create(
        {
            "source_branch": source_branch,
            "target_branch": target_branch,
            "title": title,
            "description": description,
        }
    )
    return mr.iid

get_mr_notes 

get_mr_notes(mr_iid: int) -> list[str]
Source code in src/benchmark/utils/gl_client.py 
92
93
94
95
def get_mr_notes(self, mr_iid: int) -> list[str]:
    mr = self.project.mergerequests.get(mr_iid)
    notes = mr.notes.list(all=True)
    return [n.body for n in notes if not n.system]

get_mr_pipelines 

get_mr_pipelines(mr_iid: int) -> list
Source code in src/benchmark/utils/gl_client.py 
97
98
99
def get_mr_pipelines(self, mr_iid: int) -> list:
    mr = self.project.mergerequests.get(mr_iid)
    return mr.pipelines.list()

wait_for_pipeline 

wait_for_pipeline(pipeline_id: int, timeout_s: int = 600, poll_interval: int = 15) -> str
Source code in src/benchmark/utils/gl_client.py 
101
102
103
104
105
106
107
108
109
def wait_for_pipeline(self, pipeline_id: int, timeout_s: int = 600, poll_interval: int = 15) -> str:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        pipeline = self.project.pipelines.get(pipeline_id)
        if pipeline.status not in ("created", "waiting_for_resource", "preparing", "pending", "running", "scheduled"):
            return pipeline.status
        click.echo(f"  Pipeline {pipeline_id} status: {pipeline.status}...")
        time.sleep(poll_interval)
    return "timeout"

set_variable 

set_variable(key: str, value: str, masked: bool = True, protected: bool = False) -> None
Source code in src/benchmark/utils/gl_client.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def set_variable(self, key: str, value: str, masked: bool = True, protected: bool = False) -> None:
    try:
        var = self.project.variables.get(key)
        var.value = value
        var.save()
    except gitlab.exceptions.GitlabGetError:
        self.project.variables.create(
            {
                "key": key,
                "value": value,
                "masked": masked,
                "protected": protected,
            }
        )

get_project_url 

get_project_url() -> str | None
Source code in src/benchmark/utils/gl_client.py 
126
127
def get_project_url(self) -> str | None:
    return self.project.web_url if self.project else None

get_default_branch 

get_default_branch() -> str
Source code in src/benchmark/utils/gl_client.py 
129
130
131
132
def get_default_branch(self) -> str:
    if self.project:
        return self.project.default_branch or "main"
    return "main"

GitLabProvisioner

Handles lifecycle of a GitLab project for benchmarking.

Source code in src/benchmark/utils/gl_provisioner.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class GitLabProvisioner:
    """Handles lifecycle of a GitLab project for benchmarking."""

    def __init__(self, gl_client: GitLabClient):
        self.gl_client = gl_client

    def provision(
        self,
        project_name: str,
        workflow_dir: str,
        required_files: dict | None = None,
        branch: str | None = None,
        variables: dict | None = None,
    ) -> None:
        click.echo(f"Creating GitLab project {project_name}...")
        self.gl_client.create_project(project_name)

        # GitLab needs a moment after initialize_with_readme before branches are writable
        time.sleep(5)

        default_branch = self.gl_client.get_default_branch()

        # Collect workflow files from contents/
        workflow_files: dict[str, str] = {}
        contents_dir = os.path.join(workflow_dir, "contents")
        if os.path.isdir(contents_dir):
            for root, _, filenames in os.walk(contents_dir):
                for filename in filenames:
                    abs_path = os.path.join(root, filename)
                    rel_path = os.path.relpath(abs_path, contents_dir)
                    workflow_files[rel_path] = self._read_file(abs_path)

        if workflow_files:
            click.echo(f"Pushing workflow files to {default_branch}...")
            self.gl_client.push_files(default_branch, workflow_files, "provision: add CI workflow")

        # Create target branch for scenario files (if different from default)
        target_branch = branch or default_branch
        if target_branch != default_branch:
            click.echo(f"Creating branch {target_branch}...")
            self.gl_client.create_branch(target_branch, ref=default_branch)

        # Push scenario-specific files to target branch
        if required_files:
            scenario_files: dict[str, str] = {}
            for repo_path, content_or_path in required_files.items():
                if isinstance(content_or_path, str) and os.path.exists(content_or_path):
                    scenario_files[repo_path] = self._read_file(content_or_path)
                else:
                    scenario_files[repo_path] = str(content_or_path)
            if scenario_files:
                click.echo(f"Pushing scenario files to {target_branch}...")
                self.gl_client.push_files(target_branch, scenario_files, "provision: add scenario files")

        # Set CI/CD variables
        if variables:
            for key, value in variables.items():
                if value:
                    click.echo(f"Setting CI/CD variable '{key}'...")
                    self.gl_client.set_variable(key, value, masked=True)

        click.echo(f"Project ready: {self.gl_client.get_project_url()}")

    def teardown(self) -> None:
        click.echo("Deleting GitLab project...")
        self.gl_client.delete_project()
        click.echo("Project deleted.")

    def _read_file(self, path: str) -> str:
        with open(path, "rb") as f:
            raw = f.read()
        try:
            return raw.decode("utf-8")
        except UnicodeDecodeError:
            return raw.decode("latin-1")

__init__ 

__init__(gl_client: GitLabClient)
Source code in src/benchmark/utils/gl_provisioner.py 
12
13
def __init__(self, gl_client: GitLabClient):
    self.gl_client = gl_client

provision 

provision(project_name: str, workflow_dir: str, required_files: dict | None = None, branch: str | None = None, variables: dict | None = None) -> None
Source code in src/benchmark/utils/gl_provisioner.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def provision(
    self,
    project_name: str,
    workflow_dir: str,
    required_files: dict | None = None,
    branch: str | None = None,
    variables: dict | None = None,
) -> None:
    click.echo(f"Creating GitLab project {project_name}...")
    self.gl_client.create_project(project_name)

    # GitLab needs a moment after initialize_with_readme before branches are writable
    time.sleep(5)

    default_branch = self.gl_client.get_default_branch()

    # Collect workflow files from contents/
    workflow_files: dict[str, str] = {}
    contents_dir = os.path.join(workflow_dir, "contents")
    if os.path.isdir(contents_dir):
        for root, _, filenames in os.walk(contents_dir):
            for filename in filenames:
                abs_path = os.path.join(root, filename)
                rel_path = os.path.relpath(abs_path, contents_dir)
                workflow_files[rel_path] = self._read_file(abs_path)

    if workflow_files:
        click.echo(f"Pushing workflow files to {default_branch}...")
        self.gl_client.push_files(default_branch, workflow_files, "provision: add CI workflow")

    # Create target branch for scenario files (if different from default)
    target_branch = branch or default_branch
    if target_branch != default_branch:
        click.echo(f"Creating branch {target_branch}...")
        self.gl_client.create_branch(target_branch, ref=default_branch)

    # Push scenario-specific files to target branch
    if required_files:
        scenario_files: dict[str, str] = {}
        for repo_path, content_or_path in required_files.items():
            if isinstance(content_or_path, str) and os.path.exists(content_or_path):
                scenario_files[repo_path] = self._read_file(content_or_path)
            else:
                scenario_files[repo_path] = str(content_or_path)
        if scenario_files:
            click.echo(f"Pushing scenario files to {target_branch}...")
            self.gl_client.push_files(target_branch, scenario_files, "provision: add scenario files")

    # Set CI/CD variables
    if variables:
        for key, value in variables.items():
            if value:
                click.echo(f"Setting CI/CD variable '{key}'...")
                self.gl_client.set_variable(key, value, masked=True)

    click.echo(f"Project ready: {self.gl_client.get_project_url()}")

teardown 

teardown() -> None
Source code in src/benchmark/utils/gl_provisioner.py 
72
73
74
75
def teardown(self) -> None:
    click.echo("Deleting GitLab project...")
    self.gl_client.delete_project()
    click.echo("Project deleted.")