diff --git a/pygitweb/src/pygitweb/action_runs.py b/pygitweb/src/pygitweb/action_runs.py new file mode 100644 index 0000000..0d00d9b --- /dev/null +++ b/pygitweb/src/pygitweb/action_runs.py @@ -0,0 +1,236 @@ +"""In-process async runs for plugin Action.on_mr with progress fan-out.""" + +from __future__ import annotations + +import asyncio +import logging +import secrets +import threading +from collections.abc import AsyncIterator, Coroutine +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +from pygittools.merge import MergeRequest + +from pygitweb.api.action import Action, ActionProgress, MrEvent, MrTrigger, ProgressSink + +logger = logging.getLogger(__name__) + +_MAX_HISTORY = 64 + + +@dataclass +class ActionRun: + run_id: str + action_name: str + project: str + trigger: MrTrigger + mr_ref: str + mr_oid: str + created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + history: list[ActionProgress] = field(default_factory=list) + current: ActionProgress = field( + default_factory=lambda: ActionProgress(message="queued", fraction=0.0), + ) + _condition: threading.Condition = field(default_factory=threading.Condition) + + def snapshot(self) -> dict[str, object]: + with self._condition: + cur = self.current + return { + "run_id": self.run_id, + "action": self.action_name, + "project": self.project, + "trigger": self.trigger.value, + "mr_ref": self.mr_ref, + "mr_oid": self.mr_oid, + "created_at": self.created_at.isoformat(), + "message": cur.message, + "fraction": cur.fraction, + "done": cur.done, + "ok": cur.ok, + } + + +class _RunProgressSink: + def __init__(self, registry: ActionRunRegistry, run: ActionRun) -> None: + self._registry = registry + self._run = run + + async def report(self, progress: ActionProgress) -> None: + self._registry.publish(self._run, progress) + + +class ActionRunRegistry: + """Tracks MR-triggered action runs and wakes SSE/poll waiters on progress.""" + + def __init__(self) -> None: + self._runs: dict[str, ActionRun] = {} + self._lock = threading.Lock() + + def get(self, run_id: str) -> ActionRun | None: + with self._lock: + return self._runs.get(run_id) + + def get_many(self, run_ids: list[str]) -> list[ActionRun]: + with self._lock: + return [self._runs[rid] for rid in run_ids if rid in self._runs] + + def start_mr_triggers( + self, + project: str, + trigger: MrTrigger, + mr: MergeRequest, + *, + actions: dict[str, Action] | None = None, + ) -> list[str]: + """Start matching plugin ``on_mr`` runners; return new run ids.""" + from pygitweb.plugin_loader import PLUGIN_ACTIONS + + plugins = actions if actions is not None else PLUGIN_ACTIONS + mr_oid = "" + try: + from pygitweb.git_helpers import open_repo + + repo = open_repo(project) + if mr.name in repo.references: + mr_oid = str(repo.references[mr.name].resolve().target) + except Exception: + logger.exception("pygitweb: failed to resolve MR oid for triggers") + + event = MrEvent( + project=project, + trigger=trigger, + mr_ref=mr.name, + mr_oid=mr_oid, + title=mr.title, + ours=mr.ours, + theirs=mr.theirs, + head=str(mr.target), + ) + run_ids: list[str] = [] + for name, action in plugins.items(): + try: + wanted = action.mr_triggers() + except Exception: + logger.exception("pygitweb: mr_triggers failed for action %r", name) + continue + if trigger not in wanted: + continue + run_id = secrets.token_hex(8) + run = ActionRun( + run_id=run_id, + action_name=name, + project=project, + trigger=trigger, + mr_ref=mr.name, + mr_oid=mr_oid, + ) + with self._lock: + self._runs[run_id] = run + run_ids.append(run_id) + self._spawn(_execute_run(self, run, action, event)) + return run_ids + + def _spawn(self, coro: Coroutine[Any, Any, None]) -> None: + try: + loop = asyncio.get_running_loop() + except RuntimeError: + thread = threading.Thread( + target=_run_coro_in_thread, + args=(coro,), + daemon=True, + name="pygitweb-action-run", + ) + thread.start() + return + loop.create_task(coro) + + def publish(self, run: ActionRun, progress: ActionProgress) -> None: + """Record progress and wake waiters (safe from any thread / event loop).""" + with run._condition: + run.current = progress + run.history.append(progress) + if len(run.history) > _MAX_HISTORY: + run.history = run.history[-_MAX_HISTORY:] + run._condition.notify_all() + + async def wait_for_update( + self, + run: ActionRun, + after: ActionProgress, + timeout: float = 30.0, + ) -> ActionProgress: + """Park until progress advances past ``after``, the run finishes, or ``timeout``.""" + + def _wait() -> ActionProgress: + with run._condition: + if run.current is not after or run.current.done: + return run.current + run._condition.wait(timeout=timeout) + return run.current + + return await asyncio.to_thread(_wait) + + async def stream(self, run: ActionRun) -> AsyncIterator[ActionProgress]: + """Yield the current snapshot, then each subsequent progress until done.""" + with run._condition: + current = run.current + yield current + while not current.done: + nxt = await self.wait_for_update(run, after=current) + if nxt is current and not nxt.done: + continue + current = nxt + yield current + + +def _run_coro_in_thread(coro: Coroutine[Any, Any, None]) -> None: + try: + asyncio.run(coro) + except Exception: + logger.exception("pygitweb: background action run failed") + + +async def _execute_run( + registry: ActionRunRegistry, + run: ActionRun, + action: Action, + event: MrEvent, +) -> None: + sink: ProgressSink = _RunProgressSink(registry, run) + registry.publish(run, ActionProgress(message="running", fraction=0.0)) + try: + await action.on_mr(event, sink) + if not run.current.done: + registry.publish( + run, + ActionProgress(message="completed", fraction=1.0, done=True, ok=True), + ) + except Exception as exc: + logger.exception("pygitweb: action %r on_mr failed", run.action_name) + registry.publish( + run, + ActionProgress(message=str(exc) or "failed", fraction=1.0, done=True, ok=False), + ) + + +ACTION_RUNS: ActionRunRegistry = ActionRunRegistry() + + +def start_mr_action_runs( + project: str, + trigger: MrTrigger, + mr: MergeRequest, + *, + actions: dict[str, Action] | None = None, +) -> list[str]: + """Convenience wrapper around ``ACTION_RUNS.start_mr_triggers``.""" + return ACTION_RUNS.start_mr_triggers(project, trigger, mr, actions=actions) + + +def parse_run_ids(raw: str | None) -> list[str]: + if not raw or not raw.strip(): + return [] + return [part.strip() for part in raw.split(",") if part.strip()] diff --git a/pygitweb/src/pygitweb/action_runs_test.py b/pygitweb/src/pygitweb/action_runs_test.py new file mode 100644 index 0000000..badc327 --- /dev/null +++ b/pygitweb/src/pygitweb/action_runs_test.py @@ -0,0 +1,270 @@ +"""Tests for MR-triggered plugin action runs and progress fan-out.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi import Request +from fastapi.testclient import TestClient +from pygittools.merge import MR_REF_PREFIX, MergeRequest +from starlette.responses import PlainTextResponse, Response + +from pygitweb.action_runs import ACTION_RUNS, ActionRunRegistry, start_mr_action_runs +from pygitweb.api.action import Action, ActionProgress, MrEvent, MrTrigger, ProgressSink +from pygitweb.config import settings +from pygitweb.main import app + + +class _ProgressAction(Action): + def __init__(self) -> None: + self.events: list[MrEvent] = [] + + def action_name(self) -> str: + return "progress_demo" + + def action(self, project: str, req: Request | None = None) -> Response | None: + return PlainTextResponse(project) + + def mr_triggers(self) -> frozenset[MrTrigger]: + return frozenset({MrTrigger.OPENED, MrTrigger.UPDATED}) + + async def on_mr(self, event: MrEvent, progress: ProgressSink) -> None: + self.events.append(event) + await progress.report(ActionProgress(message="step-1", fraction=0.25)) + await asyncio.sleep(0.01) + await progress.report(ActionProgress(message="step-2", fraction=0.75)) + await progress.report(ActionProgress(message="finished", fraction=1.0, done=True, ok=True)) + + +def _commit_chain(repo: pygit2.Repository) -> tuple[pygit2.Oid, pygit2.Oid]: + sig = pygit2.Signature("tester", "tester@example.com") + tb = repo.TreeBuilder() + tree = tb.write() + a = repo.create_commit(None, sig, sig, "a", tree, []) + repo.create_reference("refs/heads/main", a) + b = repo.create_commit(None, sig, sig, "b", tree, [a]) + repo.create_reference("refs/heads/feature", b) + return a, b + + +@pytest.fixture +def mr_project(tmp_path: Path) -> Generator[dict[str, object], None, None]: + root = tmp_path / "action-runs" + root.mkdir() + repo_dir = root / "demo" + repo_dir.mkdir() + repo = pygit2.init_repository(str(repo_dir), bare=False) + _a, b = _commit_chain(repo) + projects_list = root / "projects.list" + projects_list.write_text("demo tester\n", encoding="utf-8") + plugin = _ProgressAction() + with ( + patch.object(settings, "PROJECTROOT", str(root)), + patch.object(settings, "PROJECTS_LIST", str(projects_list)), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "AUTH", False), + patch("pygitweb.plugin_loader.PLUGIN_ACTIONS", {"progress_demo": plugin}), + ): + client = TestClient(app) + yield { + "client": client, + "project": "demo", + "repo": repo, + "plugin": plugin, + "tip_b": str(b), + } + + +def test_start_mr_triggers_reports_progress(mr_project: dict[str, object]) -> None: + repo = mr_project["repo"] + assert isinstance(repo, pygit2.Repository) + plugin = mr_project["plugin"] + assert isinstance(plugin, _ProgressAction) + mr = MergeRequest( + pygit2.Oid(hex=str(mr_project["tip_b"])), + pygit2.Signature("tester", "tester@example.com"), + ours="refs/heads/main", + theirs="refs/heads/feature", + title="Feature", + name=f"{MR_REF_PREFIX}abcdabcdabcdabcd", + ) + mr.write(repo) + registry = ActionRunRegistry() + run_ids = registry.start_mr_triggers( + "demo", + MrTrigger.OPENED, + mr, + actions={"progress_demo": plugin}, + ) + assert len(run_ids) == 1 + run = registry.get(run_ids[0]) + assert run is not None + + async def wait_done() -> ActionProgress: + for _ in range(100): + if run.current.done: + return run.current + await asyncio.sleep(0.02) + return run.current + + final = asyncio.run(wait_done()) + assert final.done is True + assert final.ok is True + assert final.message == "finished" + assert plugin.events and plugin.events[0].trigger is MrTrigger.OPENED + assert any(p.message == "step-1" for p in run.history) + + +def test_create_mr_starts_opened_runs(mr_project: dict[str, object]) -> None: + client = mr_project["client"] + assert isinstance(client, TestClient) + plugin = mr_project["plugin"] + assert isinstance(plugin, _ProgressAction) + post = client.post( + "/mr/create", + data={ + "project": "demo", + "theirs": "refs/heads/feature", + "ours": "refs/heads/main", + "title": "From form", + }, + follow_redirects=False, + ) + assert post.status_code == 303 + location = post.headers["location"] + assert "runs=" in location + assert "a=tag" in location + + async def wait_plugin() -> None: + for _ in range(100): + if plugin.events: + return + await asyncio.sleep(0.02) + + asyncio.run(wait_plugin()) + assert plugin.events + assert plugin.events[0].trigger is MrTrigger.OPENED + assert plugin.events[0].title == "From form" + + run_ids = location.split("runs=")[1].split("&")[0] + resp = client.get("/mr/runs", params={"project": "demo", "ids": run_ids}) + assert resp.status_code == 200 + body = resp.json() + assert body["project"] == "demo" + assert body["runs"] + assert body["runs"][0]["action"] == "progress_demo" + + +def test_runs_sse_streams_until_done(mr_project: dict[str, object]) -> None: + repo = mr_project["repo"] + assert isinstance(repo, pygit2.Repository) + plugin = mr_project["plugin"] + assert isinstance(plugin, _ProgressAction) + client = mr_project["client"] + assert isinstance(client, TestClient) + mr = MergeRequest( + pygit2.Oid(hex=str(mr_project["tip_b"])), + pygit2.Signature("tester", "tester@example.com"), + ours="refs/heads/main", + title="SSE", + name=f"{MR_REF_PREFIX}sse0sse0sse0sse0", + ) + mr.write(repo) + run_ids = start_mr_action_runs("demo", MrTrigger.OPENED, mr, actions={"progress_demo": plugin}) + assert run_ids + with client.stream( + "GET", + f"/mr/runs/{run_ids[0]}/events", + params={"project": "demo"}, + ) as resp: + assert resp.status_code == 200 + chunks: list[str] = [] + for line in resp.iter_lines(): + if line.startswith("data: "): + chunks.append(line.removeprefix("data: ")) + if '"done": true' in line or '"done":true' in line: + break + assert chunks + assert any("finished" in chunk or "completed" in chunk or "step" in chunk for chunk in chunks) + + +def test_action_defaults_skip_triggers() -> None: + class _Plain(Action): + def action_name(self) -> str: + return "plain" + + def action(self, project: str, req: Request | None = None) -> Response | None: + return None + + plain = _Plain() + assert plain.mr_triggers() == frozenset() + mr = MergeRequest( + "0" * 40, + "x ", + ours="refs/heads/main", + title="t", + name=f"{MR_REF_PREFIX}ffffffffffffffff", + ) + # Invalid head oid — start_mr_triggers should still skip because no triggers. + ids = ActionRunRegistry().start_mr_triggers( + "demo", + MrTrigger.OPENED, + mr, + actions={"plain": plain}, + ) + assert ids == [] + + +def test_parse_and_registry_helpers() -> None: + from pygitweb.action_runs import parse_run_ids + + assert parse_run_ids(None) == [] + assert parse_run_ids(" a, b , ") == ["a", "b"] + assert ACTION_RUNS.get("missing") is None + + +def test_refresh_from_theirs_starts_updated_runs(mr_project: dict[str, object]) -> None: + repo = mr_project["repo"] + assert isinstance(repo, pygit2.Repository) + plugin = mr_project["plugin"] + assert isinstance(plugin, _ProgressAction) + client = mr_project["client"] + assert isinstance(client, TestClient) + sig = pygit2.Signature("tester", "tester@example.com") + main = repo.references["refs/heads/main"].target + feature = repo.references["refs/heads/feature"].target + mr = MergeRequest( + main, + sig, + ours="refs/heads/main", + theirs="refs/heads/feature", + title="Track feature", + name=f"{MR_REF_PREFIX}upd0upd0upd0upd0", + ) + mr_oid = mr.write(repo) + assert str(mr.target) == str(main) + assert str(feature) != str(main) + + resp = client.get( + f"/project/{mr_project['project']}", + params={"a": "tag", "h": str(mr_oid)}, + follow_redirects=False, + ) + assert resp.status_code == 302 + location = resp.headers["location"] + assert "runs=" in location + + async def wait_updated() -> None: + for _ in range(100): + if any(e.trigger is MrTrigger.UPDATED for e in plugin.events): + return + await asyncio.sleep(0.02) + + asyncio.run(wait_updated()) + assert any(e.trigger is MrTrigger.UPDATED for e in plugin.events) diff --git a/pygitweb/src/pygitweb/actions.py b/pygitweb/src/pygitweb/actions.py index 20a56bf..4996223 100644 --- a/pygitweb/src/pygitweb/actions.py +++ b/pygitweb/src/pygitweb/actions.py @@ -1132,8 +1132,15 @@ def git_tags(project: str, request: Request, page: int, pagecount: int) -> HTMLR ) -def git_tag(project: str, h: str | None, notice: str | None = None) -> HTMLResponse | RedirectResponse: +def git_tag( + project: str, + h: str | None, + notice: str | None = None, + runs: str | None = None, +) -> HTMLResponse | RedirectResponse: """Single tag view. Port of git_tag.""" + from pygitweb.action_runs import parse_run_ids + if not h: raise HTTPException(status_code=400, detail="Tag ref or hash (h) required") if not is_valid_ref_format(h): @@ -1177,6 +1184,7 @@ def git_tag(project: str, h: str | None, notice: str | None = None) -> HTMLRespo tagger=tagger, tagger_date=_format_date(tagger_epoch, tagger_tz), notice=notice, + run_ids=parse_run_ids(runs), ) target_short = target_oid[:7] if target_oid else "" if target_type == "commit": diff --git a/pygitweb/src/pygitweb/api/action.py b/pygitweb/src/pygitweb/api/action.py index 73e8cb1..468f0c0 100644 --- a/pygitweb/src/pygitweb/api/action.py +++ b/pygitweb/src/pygitweb/api/action.py @@ -1,21 +1,105 @@ from __future__ import annotations +from dataclasses import dataclass +from enum import StrEnum from typing import Protocol, runtime_checkable from abcreg import AnnotatedProtocolMeta # type: ignore[import-untyped] from fastapi import Request from starlette.responses import Response +_DEFAULT_METHODS = ("mr_triggers", "on_mr") + + +class MrTrigger(StrEnum): + OPENED = "opened" + UPDATED = "updated" + + +@dataclass(frozen=True, slots=True) +class MrEvent: + """Context passed to plugin actions when an MR is opened or updated.""" + + project: str + trigger: MrTrigger + mr_ref: str + mr_oid: str + title: str + ours: str + theirs: str | None + head: str + + +@dataclass(frozen=True, slots=True) +class ActionProgress: + """One progress update from an async MR action runner.""" + + message: str + fraction: float | None = None + done: bool = False + ok: bool | None = None + @runtime_checkable -class Action(Protocol, metaclass=AnnotatedProtocolMeta): +class ProgressSink(Protocol): + """Receives progress from ``Action.on_mr`` so the caller can stream updates.""" + + async def report(self, progress: ActionProgress) -> None: ... + + +def _install_default_methods(interface: type, impl: type) -> None: + """Give implementations the protocol's optional methods before signature checks.""" + if getattr(impl, "_is_protocol", False): + return + skip = set(interface.__mro__) + for name in _DEFAULT_METHODS: + if any(name in base.__dict__ for base in impl.__mro__ if base not in skip): + continue + for base in interface.__mro__: + default = base.__dict__.get(name) + if callable(default): + setattr(impl, name, default) + break + + +class _ActionMeta(AnnotatedProtocolMeta): + def __init__( + cls, + name: str, + bases: tuple[type, ...], + namespace: dict[str, object], + **kwargs: object, + ) -> None: + if not getattr(cls, "_is_protocol", False): + for base in bases: + if isinstance(base, _ActionMeta) and getattr(base, "_is_protocol", False): + _install_default_methods(base, cls) + super().__init__(name, bases, namespace, **kwargs) + + def register(cls, subclass: type) -> type: + _install_default_methods(cls, subclass) + super().register(subclass) + return subclass + + +@runtime_checkable +class Action(Protocol, metaclass=_ActionMeta): """ Action: A per-project action provided by a plugin. ``project`` is the project path/slug as used in URLs (under PROJECTROOT). Subclass this protocol, or register a virtual subclass with ``@Action.register``. + + Optional MR hooks: declare interest with ``mr_triggers`` and implement ``on_mr`` + as an async runner that reports progress through ``progress``. """ def action_name(self) -> str: ... def action(self, project: str, req: Request | None = None) -> Response | None: ... + + def mr_triggers(self) -> frozenset[MrTrigger]: + return frozenset() + + async def on_mr(self, event: MrEvent, progress: ProgressSink) -> None: + return None diff --git a/pygitweb/src/pygitweb/main.py b/pygitweb/src/pygitweb/main.py index 93110dd..770ab24 100644 --- a/pygitweb/src/pygitweb/main.py +++ b/pygitweb/src/pygitweb/main.py @@ -878,6 +878,7 @@ async def dispatch( search_sort: Annotated[SearchSortQuery | None, Query(alias="sort")] = None, search_max_count: Annotated[str | None, Query(alias="max_count")] = None, notice: Annotated[str | None, Query(alias="notice")] = None, + runs: Annotated[str | None, Query(alias="runs")] = None, ) -> Response: """ Dispatch by path: /project/{project} -> summary; /project/{project}/action/... -> action. @@ -963,7 +964,7 @@ async def dispatch( p, pc = parse_pagination(page, pagecount) return git_tags(project, request, p, pc) if action == "tag": - return git_tag(project, hash_param, notice=notice) + return git_tag(project, hash_param, notice=notice, runs=runs) if action == "commit": return git_commit(project, hash_param) if action == "patch": diff --git a/pygitweb/src/pygitweb/merge_requests.py b/pygitweb/src/pygitweb/merge_requests.py index de02916..23730c3 100644 --- a/pygitweb/src/pygitweb/merge_requests.py +++ b/pygitweb/src/pygitweb/merge_requests.py @@ -2,17 +2,23 @@ from __future__ import annotations +import asyncio import json from typing import Annotated, Any from urllib.parse import quote, urlencode import pygit2 -from fastapi import APIRouter, Depends, Form, HTTPException, Query -from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request +from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse from pygittools.merge import MergeRequest, MergeRequestStatus, get_merge_request_by_oid +from pygitweb.action_runs import ACTION_RUNS, parse_run_ids, start_mr_action_runs +from pygitweb.api.action import MrTrigger from pygitweb.auth import access_token_from_request, principal_from_session, require_permission -from pygitweb.dependencies import ValidatedReadableFormProject, ValidatedReadableQueryProject +from pygitweb.dependencies import ( + ValidatedReadableFormProject, + ValidatedReadableQueryProject, +) from pygitweb.git_helpers import open_repo, resolve_mr_opener from pygitweb.permissions import Permission from pygitweb.templates_env import render_page @@ -105,11 +111,19 @@ def merge_request_tag_response( tagger: str, tagger_date: str, notice: str | None = None, + run_ids: list[str] | None = None, ) -> HTMLResponse | RedirectResponse: new_oid = mr.refresh_from_theirs(repo) if new_oid is not None: + updated_runs = start_mr_action_runs(project, MrTrigger.UPDATED, mr) + params: dict[str, str] = {"a": "tag", "h": str(new_oid)} + all_runs = [*(run_ids or []), *updated_runs] + if all_runs: + params["runs"] = ",".join(all_runs) + if notice: + params["notice"] = notice return RedirectResponse( - url=f"/project/{project}?a=tag&h={quote(str(new_oid), safe='')}", + url=f"/project/{project}?{urlencode(params)}", status_code=302, ) @@ -126,6 +140,27 @@ def merge_request_tag_response( comments_rows.append((ch[:7], f"/project/{project}?a=tag&h={quote(ch, safe='')}")) current_tag_oid = str(repo.references[mr.name].resolve().target) + requested_ids = run_ids or [] + known = {run.run_id: run.snapshot() for run in ACTION_RUNS.get_many(requested_ids)} + action_runs = [ + known.get( + rid, + { + "run_id": rid, + "action": rid, + "project": project, + "trigger": "", + "mr_ref": mr.name, + "mr_oid": current_tag_oid, + "created_at": "", + "message": "queued", + "fraction": 0.0, + "done": False, + "ok": None, + }, + ) + for rid in requested_ids + ] ctx = { "project": project, @@ -150,6 +185,8 @@ def merge_request_tag_response( "comments": comments_rows, "merge_action_url": f"/mr/ff?project={quote(project, safe='')}&h={quote(current_tag_oid, safe='')}", "notice": notice or "", + "action_runs": action_runs, + "action_run_ids": ",".join(requested_ids), **status_fields, } title = f"Merge request: {mr.title or ''} — {project or ''}" @@ -195,7 +232,10 @@ def merge_request_create( description="", ) oid = mr.write(repo) + run_ids = start_mr_action_runs(project, MrTrigger.OPENED, mr) params: dict[str, str] = {"a": "tag", "h": str(oid)} + if run_ids: + params["runs"] = ",".join(run_ids) if opener_warning: params["notice"] = opener_warning dest = f"/project/{project}?{urlencode(params)}" @@ -231,3 +271,50 @@ def merge_fast_forward( tag_oid = str(repo.references[mr.name].resolve().target) dest = f"/project/{project}?a=tag&h={quote(tag_oid, safe='')}" return RedirectResponse(url=dest, status_code=303) + + +@merge_router.get("/runs") +def merge_request_runs( + project: ValidatedReadableQueryProject, + ids: Annotated[str, Query(description="Comma-separated action run ids")] = "", +) -> dict[str, object]: + """Snapshot progress for one or more MR-triggered action runs.""" + requested = parse_run_ids(ids) + runs = ACTION_RUNS.get_many(requested) + missing = [rid for rid in requested if ACTION_RUNS.get(rid) is None] + return { + "project": project, + "runs": [run.snapshot() for run in runs], + "missing": missing, + } + + +@merge_router.get("/runs/{run_id}/events") +async def merge_request_run_events( + run_id: str, + request: Request, + project: ValidatedReadableQueryProject, +) -> StreamingResponse: + """Server-sent events stream of progress for a single action run.""" + run = ACTION_RUNS.get(run_id) + if run is None or run.project != project: + raise HTTPException(status_code=404, detail="Action run not found") + + async def event_gen() -> Any: + async for progress in ACTION_RUNS.stream(run): + if await request.is_disconnected(): + return + payload = { + "run_id": run.run_id, + "action": run.action_name, + "message": progress.message, + "fraction": progress.fraction, + "done": progress.done, + "ok": progress.ok, + } + yield f"data: {json.dumps(payload)}\n\n" + if progress.done: + return + await asyncio.sleep(0) + + return StreamingResponse(event_gen(), media_type="text/event-stream") diff --git a/pygitweb/src/pygitweb/plugin_loader_test.py b/pygitweb/src/pygitweb/plugin_loader_test.py index a7e5bd8..f64a08d 100644 --- a/pygitweb/src/pygitweb/plugin_loader_test.py +++ b/pygitweb/src/pygitweb/plugin_loader_test.py @@ -73,6 +73,7 @@ def test_load_skips_disabled_plugins() -> None: assert load_subpages(frozenset()) == [] actions = load_plugin_actions(frozenset({"pygitweb_demo"})) assert set(actions) == {"demo_action"} + assert actions["demo_action"].mr_triggers() == frozenset() subpages = load_subpages(frozenset({"pygitweb_demo"})) assert [s.subpage_name() for s in subpages] == ["demo_subpage"] assert subpages[0].link_name("proj") is None diff --git a/pygitweb/src/pygitweb/static/mr-action-runs.js b/pygitweb/src/pygitweb/static/mr-action-runs.js new file mode 100644 index 0000000..d9f2703 --- /dev/null +++ b/pygitweb/src/pygitweb/static/mr-action-runs.js @@ -0,0 +1,101 @@ +(() => { + const root = document.getElementById("mr-action-runs"); + if (!(root instanceof HTMLElement)) { + return; + } + const project = root.dataset.project || ""; + const runIds = (root.dataset.runIds || "") + .split(",") + .map((id) => id.trim()) + .filter(Boolean); + if (!project || runIds.length === 0) { + return; + } + + const applyProgress = (payload) => { + const item = root.querySelector(`[data-run-id="${payload.run_id}"]`); + if (!(item instanceof HTMLElement)) { + return; + } + const status = item.querySelector("[data-run-status]"); + const bar = item.querySelector("[data-run-bar]"); + if (status) { + status.textContent = payload.message || ""; + } + if (bar instanceof HTMLElement) { + const pct = Math.max(0, Math.min(100, Math.round((payload.fraction ?? 0) * 100))); + bar.style.width = `${pct}%`; + bar.setAttribute("aria-valuenow", String(pct)); + bar.classList.remove("bg-success", "bg-danger"); + if (payload.done && payload.ok === true) { + bar.classList.add("bg-success"); + } else if (payload.done && payload.ok === false) { + bar.classList.add("bg-danger"); + } + } + return Boolean(payload.done); + }; + + const pollOnce = async () => { + const params = new URLSearchParams({ project, ids: runIds.join(",") }); + const resp = await fetch(`/mr/runs?${params.toString()}`, { + headers: { Accept: "application/json" }, + }); + if (!resp.ok) { + return true; + } + const body = await resp.json(); + const runs = Array.isArray(body.runs) ? body.runs : []; + let allDone = runs.length > 0; + for (const run of runs) { + if (!applyProgress(run)) { + allDone = false; + } + } + return allDone; + }; + + const preferSSE = typeof EventSource === "function"; + if (preferSSE) { + const sources = runIds.map((runId) => { + const params = new URLSearchParams({ project }); + const source = new EventSource(`/mr/runs/${encodeURIComponent(runId)}/events?${params}`); + source.onmessage = (event) => { + try { + const payload = JSON.parse(event.data); + if (applyProgress(payload)) { + source.close(); + } + } catch { + source.close(); + } + }; + source.onerror = () => { + source.close(); + }; + return source; + }); + window.addEventListener( + "beforeunload", + () => { + for (const source of sources) { + source.close(); + } + }, + { once: true }, + ); + return; + } + + const poll = async () => { + try { + const done = await pollOnce(); + if (!done) { + window.setTimeout(poll, 1000); + } + } catch { + window.setTimeout(poll, 2000); + } + }; + poll(); +})(); diff --git a/pygitweb/src/pygitweb/templates/merge_request_tag.html b/pygitweb/src/pygitweb/templates/merge_request_tag.html index 4a70db3..3882185 100644 --- a/pygitweb/src/pygitweb/templates/merge_request_tag.html +++ b/pygitweb/src/pygitweb/templates/merge_request_tag.html @@ -1,6 +1,6 @@ {% extends "base.html" %} {% block content %} -{# Merge request view for annotated MR tags. Expects: project, project_url, tag_name, tag_oid, tag_view_url, mr_title, mr_description, ours_ref, ours_url, theirs_ref, theirs_url, head_short, head_url, head_full, tagger, tagger_date, opener, created_at, updated_at, comments [ (short, url) ], merge_action_url, status_code, status_label, status_detail, show_merge_button #} +{# Merge request view for annotated MR tags. Expects: project, project_url, tag_name, tag_oid, tag_view_url, mr_title, mr_description, ours_ref, ours_url, theirs_ref, theirs_url, head_short, head_url, head_full, tagger, tagger_date, opener, created_at, updated_at, comments [ (short, url) ], merge_action_url, status_code, status_label, status_detail, show_merge_button, action_runs, action_run_ids #}

Merge request

@@ -8,6 +8,38 @@ {% endif %} +{% if action_run_ids %} +
+
+

Plugin actions

+
+
+
    + {% for run in action_runs %} +
  • +
    + {{ run.action }} + {{ run.message }} +
    +
    +
    +
    +
    • + {% endfor %} +
+
+
+ +{% endif %} +

Project: {{ project }} · Tag: {{ tag_name }}