diff --git a/README.md b/README.md index 1e1a5fb..2f8a603 100644 --- a/README.md +++ b/README.md @@ -49,3 +49,30 @@ work_item/scripts/python -m pytest tests/e2e tests/integration -q - `tests/story_traceability.json` maps every delivered story to its tests; `tests/e2e/test_traceability.py` fails if a Phase A story loses coverage or a test file is left unexercised. + +### Phase B acceptance gate + +Phase B (Epic E02: durable jobs, workflow shell, safety/analysis views) is proven +through the real process and browser boundaries. One command runs the Phase B API, +worker-recovery, and browser (Playwright) suites: + +```bash +work_item/scripts/python -m pytest tests/e2e -m phase_b -q +``` + +- `tests/e2e/test_phase_b_pipeline.py` launches the real server and durable worker as + child processes and drives them only over HTTP/SSE: analysis start/progress, + resumable SSE reconnect, the polling fallback, cancellation, per-asset error + inspection, one-mutating-job rejection during read-only browsing, the NSFW→vision + privacy gate, and durability across a full restart. +- `tests/e2e/test_worker_kill.py` kills a worker mid-item and proves a fresh worker + resumes the fenced job (the "resume" journey). +- `tests/e2e/test_analysis_browser.py` starts a job from the Analyze view and watches + live progress arrive over the browser's real SSE adapter. + +The full Phase B regression, including the unchanged Phase A gate, is the whole +end-to-end suite: + +```bash +work_item/scripts/python -m pytest tests/e2e -q +``` diff --git a/photo_pipeline/__main__.py b/photo_pipeline/__main__.py index 06d4a41..db68a53 100644 --- a/photo_pipeline/__main__.py +++ b/photo_pipeline/__main__.py @@ -27,6 +27,11 @@ def main(argv: Sequence[str] | None = None) -> int: if args.command == "worker": from photo_pipeline.db import create_db_engine, create_session_factory + + # Importing this registers the safety/analysis job handlers into the shared + # REGISTRY the Worker defaults to; without it a standalone worker process + # claims nothing because it knows no job types. + import photo_pipeline.jobs.domain_handlers # noqa: F401 from photo_pipeline.jobs.worker import Worker run_migrations(config.database_url) diff --git a/photo_pipeline/services/analysis.py b/photo_pipeline/services/analysis.py index 8222f2b..454fdbd 100644 --- a/photo_pipeline/services/analysis.py +++ b/photo_pipeline/services/analysis.py @@ -17,6 +17,7 @@ ponytail: port the donor's retry+RPD throttling when analysis runs at real volum from __future__ import annotations import json +import os import uuid from datetime import datetime, timezone from typing import Protocol @@ -216,9 +217,45 @@ def _result_dict(row: AnalysisResult) -> dict: def _default_provider() -> VisionProvider: + # Test seam (concept §18: deterministic fakes replace the vision edge, enabled + # only by test configuration). When this env var names a writable log file, the + # worker/API use a fake that records every analyzed path — so the SFW-only gate + # can be asserted end-to-end through the real integration layer — instead of the + # real OpenAI-compatible call. Never set in production. + log_path = os.environ.get("PHOTO_PIPELINE_FAKE_VISION_LOG") + if log_path: + return _RecordingFakeVision(log_path) return OpenAIVisionProvider() +class _RecordingFakeVision: + """Deterministic vision fake for end-to-end tests. Appends every analyzed file + path to its log so a test can prove NSFW assets never reach the provider, and + raises for a path whose stem contains ``boom`` to exercise per-asset error + handling. Constructed only when ``PHOTO_PIPELINE_FAKE_VISION_LOG`` is set.""" + + def __init__(self, log_path: str) -> None: + self._log_path = log_path + + def analyze(self, path: str, *, album_hint: str) -> dict: + with open(self._log_path, "a", encoding="utf-8") as handle: + handle.write(path + "\n") + if "boom" in os.path.splitext(os.path.basename(path))[0]: + raise AnalysisError("fake vision failure") + return { + "description": f"a deterministic scene in {album_hint}", + "tags": ["fixture", "deterministic"], + "people_count": 1, + "setting": "outdoor", + "time_of_day": "day", + "season": "summer", + "mood": "calm", + "location_hint": None, + "approx_year": None, + "_tokens": 7, + } + + class OpenAIVisionProvider: """The real provider: an OpenAI-compatible vision call (Gemini by default). diff --git a/pyproject.toml b/pyproject.toml index 6c63f0b..21eec90 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,3 +27,6 @@ line-length = 100 [tool.pytest.ini_options] testpaths = ["tests"] +markers = [ + "phase_b: Phase B end-to-end acceptance (US02-07) — API, worker-recovery, and browser journeys", +] diff --git a/tests/e2e/_pipeline_harness.py b/tests/e2e/_pipeline_harness.py new file mode 100644 index 0000000..39de202 --- /dev/null +++ b/tests/e2e/_pipeline_harness.py @@ -0,0 +1,161 @@ +"""Reusable Phase B end-to-end harness: seed an isolated library, then launch the +real API server and durable worker as child processes so tests talk to them only +over HTTP/SSE — the same boundaries used in production (concept §18). + +External vision is replaced at its integration edge by the in-repo recording fake +(``PHOTO_PIPELINE_FAKE_VISION_LOG``); it is invoked through the real AnalysisService +and worker, never mocked inside a test. +""" + +from __future__ import annotations + +import os +import socket +import subprocess +import sys +import time +from dataclasses import dataclass +from pathlib import Path + +import httpx +import numpy as np +from PIL import Image + +REPO = Path(__file__).resolve().parents[2] + + +def image(path: Path, seed: int) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + rng = np.random.default_rng(seed) + Image.fromarray(rng.integers(0, 256, (96, 128, 3), dtype=np.uint8)).save(path, quality=90) + + +def free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +@dataclass +class Seeded: + data: Path + lib: Path + asset_ids: dict[str, str] # filename stem -> asset id + + +def seed_library(tmp_path: Path, files: dict[str, int], decisions: dict[str, str]) -> Seeded: + """Scan ``files`` (stem -> seed) into a fresh DB and apply safety ``decisions`` + (stem -> ``sfw``/``nsfw``) in-process before any server starts, so the seed is + deterministic and independent of the API under test.""" + data = tmp_path / "data" + data.mkdir() + lib = tmp_path / "lib" + lib.mkdir() + for stem, seed in files.items(): + image(lib / f"{stem}.jpg", seed) + + from photo_pipeline.config import Config + from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations + from photo_pipeline.models import Asset + from photo_pipeline.services.inventory import InventoryService + from photo_pipeline.services.safety import SafetyService + from sqlalchemy import select + + config = Config.from_env( + {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)} + ) + run_migrations(config.database_url) + engine = create_db_engine(config.database_url) + sf = create_session_factory(engine) + InventoryService(sf).scan(lib) + with sf() as session: + rows = list(session.execute(select(Asset.id, Asset.current_path)).all()) + ids = {Path(path).stem: aid for aid, path in rows} + safety = SafetyService(sf) + for stem, decision in decisions.items(): + safety.decide(ids[stem], decision, write_exif=False) + engine.dispose() + return Seeded(data=data, lib=lib, asset_ids=ids) + + +def _env(seeded: Seeded, port: int, *, extra: dict[str, str] | None = None) -> dict[str, str]: + env = { + "PATH": os.environ.get("PATH", ""), + "PHOTO_PIPELINE_DATA_DIR": str(seeded.data), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(seeded.lib), + "PHOTO_PIPELINE_HOST": "127.0.0.1", + "PHOTO_PIPELINE_PORT": str(port), + } + if extra: + env.update(extra) + return env + + +class Server: + """A real ``photo_pipeline serve`` child process, ready when /health/ready is 200.""" + + def __init__(self, seeded: Seeded, *, extra_env: dict[str, str] | None = None) -> None: + self._seeded = seeded + self._extra = extra_env + self.port = free_port() + self.base = f"http://127.0.0.1:{self.port}" + self.proc: subprocess.Popen | None = None + + def start(self) -> "Server": + self.proc = subprocess.Popen( + [sys.executable, "-m", "photo_pipeline", "serve"], + cwd=str(REPO), + env=_env(self._seeded, self.port, extra=self._extra), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if self.proc.poll() is not None: + _, err = self.proc.communicate() + raise RuntimeError(f"server exited: {err.decode(errors='replace')}") + try: + if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200: + return self + except httpx.HTTPError: + time.sleep(0.2) + self.stop() + raise RuntimeError("server never became ready") + + def stop(self) -> None: + if self.proc is None: + return + self.proc.terminate() + try: + self.proc.wait(timeout=10) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc = None + + +def start_worker(seeded: Seeded, *, fake_vision_log: Path) -> subprocess.Popen: + """Launch a real durable worker wired to the recording vision fake.""" + return subprocess.Popen( + [sys.executable, "-m", "photo_pipeline", "worker", "--id", "e2e-worker"], + cwd=str(REPO), + env=_env( + seeded, + free_port(), # unused by the worker, but keeps the env shape uniform + extra={"PHOTO_PIPELINE_FAKE_VISION_LOG": str(fake_vision_log)}, + ), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + +def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1): + """Poll ``predicate`` until it returns a truthy value or the timeout elapses. + Returns the truthy value; raises on timeout. Synchronizes on observable state + instead of sleeping for a fixed duration (concept §18).""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(interval) + raise AssertionError("condition not met before timeout") diff --git a/tests/e2e/test_analysis_browser.py b/tests/e2e/test_analysis_browser.py new file mode 100644 index 0000000..5efc435 --- /dev/null +++ b/tests/e2e/test_analysis_browser.py @@ -0,0 +1,58 @@ +"""Phase B browser journey (US02-07): start a content-analysis job from the Analyze +view and watch live progress arrive over the real SSE adapter, against a real server +and durable worker. The frontend's SSE→polling event adapter and the worker are +exercised end to end; the vision edge is the recording fake. +""" + +from __future__ import annotations + +import pytest +from playwright.sync_api import expect + +from tests.e2e._pipeline_harness import Server, seed_library, start_worker + +pytestmark = pytest.mark.phase_b + + +@pytest.fixture +def running_stack(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + yield server + finally: + worker.terminate() + server.stop() + + +def test_analyze_start_shows_live_progress_and_completes(page, running_stack): + errors = [] + page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None) + + page.goto(f"{running_stack.base}/app/#/analyze") + run = page.get_by_test_id("run-analysis") + run.wait_for() + assert run.is_enabled() # one confirmed-SFW asset is eligible + + run.click() + + # Live progress streams into the activity log via the browser's SSE adapter. + log = page.get_by_test_id("analyze-log") + log.get_by_text("Started analysis job").wait_for() + + # On completion the view re-renders with the updated counts: the SFW asset is + # now analysed. Waiting on this observable state (not a sleep) proves the whole + # start → worker → SSE → done round trip worked. + analyzed_value = ( + page.get_by_test_id("analyze-counts") + .locator(".stat") + .filter(has_text="Analyzed") + .locator(".stat-value") + ) + # ``expect`` re-queries the locator each poll, so it survives the re-render that + # replaces the counts DOM when the job completes. + expect(analyzed_value).to_have_text("1", timeout=15000) + + assert errors == [], f"console errors: {errors}" diff --git a/tests/e2e/test_phase_b_pipeline.py b/tests/e2e/test_phase_b_pipeline.py new file mode 100644 index 0000000..49f67e3 --- /dev/null +++ b/tests/e2e/test_phase_b_pipeline.py @@ -0,0 +1,254 @@ +"""Phase B black-box end-to-end acceptance (US02-07). + +Launches the real FastAPI server and durable worker as child processes and drives +them only over HTTP/SSE — no service/repository imports for the behaviour under +test. Covers the Phase B journeys the browser suite does not exhaustively assert: +analysis start/progress, resumable SSE reconnect, the polling fallback, +cancellation, per-asset error inspection, one-mutating-job rejection during +read-only browsing, the NSFW→vision privacy gate, and durability across restart. + +External vision is the in-repo recording fake, invoked through the real worker. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest + +from tests.e2e._pipeline_harness import Server, seed_library, start_worker, wait_until + +pytestmark = pytest.mark.phase_b + +TIMEOUT = 10 + + +def _job_state(base: str, job_id: str) -> str: + return httpx.get(f"{base}/api/v1/jobs/{job_id}", timeout=TIMEOUT).json()["state"] + + +def _sse_events( + base: str, + job_id: str, + *, + after: int = 0, + last_event_id: int | None = None, + stop_after: int | None = None, +) -> list[dict]: + """Read the resumable SSE stream, returning ``[{seq, type, message}, ...]``. When + ``stop_after`` is set, close the stream after that many events to model a client + that drops mid-stream. ``last_event_id`` sends the reconnect header.""" + headers = {} + if last_event_id is not None: + headers["Last-Event-ID"] = str(last_event_id) + url = f"{base}/api/v1/jobs/{job_id}/events/stream?after={after}" + events: list[dict] = [] + seq: int | None = None + with httpx.stream("GET", url, headers=headers, timeout=TIMEOUT) as response: + response.raise_for_status() + for line in response.iter_lines(): + if line.startswith("id:"): + seq = int(line[3:].strip()) + elif line.startswith("data:"): + payload = json.loads(line[5:].strip()) + if seq is not None: # skip the terminal `event: done` (empty {} data) + events.append({"seq": seq, **payload}) + seq = None + if stop_after is not None and len(events) >= stop_after: + return events + elif line.startswith("event: done"): + break + return events + + +# ── analysis job: start, progress, and completion ──────────────────────────── + + +def test_analysis_job_runs_to_completion_over_real_http(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json() + wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT) + + counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json() + assert counts["analyzed"] == 1 + # The fake actually ran through the real worker for the one SFW asset. + assert log.read_text().strip().endswith("beach.jpg") + finally: + worker.terminate() + server.stop() + + +# ── resumable SSE reconnect and the polling fallback ───────────────────────── + + +def test_sse_reconnect_resumes_without_gap_or_duplicate(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1, "meadow": 2}, {"beach": "sfw", "meadow": "sfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json() + wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT) + + full = _sse_events(server.base, job["id"]) + assert [e["type"] for e in full][:2] == ["queued", "claimed"] + assert full[-1]["type"] == "state:succeeded" + assert len(full) >= 3 + + # A client that reads only the first event, drops, then reconnects with the + # last id it saw must receive exactly the remaining events — no gap, no dup. + first = _sse_events(server.base, job["id"], stop_after=1) + assert len(first) == 1 + rest = _sse_events(server.base, job["id"], last_event_id=first[0]["seq"]) + assert first + rest == full + seqs = [e["seq"] for e in full] + assert seqs == sorted(set(seqs)) # strictly increasing, unique + finally: + worker.terminate() + server.stop() + + +def test_polling_fallback_returns_the_same_durable_events(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json() + wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT) + + sse = _sse_events(server.base, job["id"]) + polled = httpx.get( + f"{server.base}/api/v1/jobs/{job['id']}/events?after=0", timeout=TIMEOUT + ).json() + assert polled["state"] == "succeeded" + assert [(e["seq"], e["type"]) for e in polled["events"]] == [ + (e["seq"], e["type"]) for e in sse + ] + finally: + worker.terminate() + server.stop() + + +# ── cancellation ───────────────────────────────────────────────────────────── + + +def test_queued_job_cancels_and_runs_no_items(tmp_path): + # No worker: the safety job stays queued so cancellation is deterministic. + seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {}) + server = Server(seeded).start() + try: + job = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT).json() + cancelled = httpx.post( + f"{server.base}/api/v1/jobs/{job['id']}/cancel", timeout=TIMEOUT + ).json() + assert cancelled["state"] == "cancelled" + + snapshot = httpx.get(f"{server.base}/api/v1/jobs/{job['id']}", timeout=TIMEOUT).json() + assert snapshot["progress"]["by_state"].get("succeeded", 0) == 0 + finally: + server.stop() + + +# ── per-asset error inspection without losing completed work ────────────────── + + +def test_one_analysis_error_is_inspectable_and_others_still_complete(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1, "boom": 2}, {"beach": "sfw", "boom": "sfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json() + wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT) + + counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json() + assert counts["analyzed"] == 1 and counts["error"] == 1 # boom failed, beach kept + + boom_id = seeded.asset_ids["boom"] + result = httpx.get( + f"{server.base}/api/v1/analysis/results/{boom_id}", timeout=TIMEOUT + ).json() + assert result["status"] == "error" + finally: + worker.terminate() + server.stop() + + +# ── one mutating job at a time, read-only browsing stays available ─────────── + + +def test_second_mutation_rejected_while_reads_continue(tmp_path): + # No worker: the first safety job holds the library_write lock for the test. + seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {}) + server = Server(seeded).start() + try: + first = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT) + assert first.status_code == 200 + + second = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT) + assert second.status_code == 409 + assert second.json()["error"]["code"] == "lock_held" + + # Read-only browsing is unaffected by the held mutation lock. + for path in ("/api/v1/workflow", "/api/v1/library/assets", "/api/v1/safety/queue"): + assert httpx.get(f"{server.base}{path}", timeout=TIMEOUT).status_code == 200 + finally: + server.stop() + + +# ── privacy gate: NSFW assets never reach the vision provider ──────────────── + + +def test_nsfw_asset_never_reaches_the_vision_provider(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"}) + log = tmp_path / "vision.log" + server = Server(seeded).start() + worker = start_worker(seeded, fake_vision_log=log) + try: + job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json() + wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT) + + logged = log.read_text() + assert "beach.jpg" in logged # the SFW asset was analysed + assert "city.jpg" not in logged # the NSFW asset never reached the provider + + # And the NSFW asset was never even enqueued: no analysis row exists for it. + city_id = seeded.asset_ids["city"] + assert ( + httpx.get( + f"{server.base}/api/v1/analysis/results/{city_id}", timeout=TIMEOUT + ).status_code + == 404 + ) + finally: + worker.terminate() + server.stop() + + +# ── durability: decisions survive a full server restart ────────────────────── + + +def test_safety_decisions_survive_a_restart(tmp_path): + seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"}) + server = Server(seeded).start() + try: + before = httpx.get(f"{server.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT).json() + assert any(row["asset_id"] == seeded.asset_ids["city"] for row in before["items"]) + finally: + server.stop() + + # Fresh process against the same data dir: the durable decision is still there. + restarted = Server(seeded).start() + try: + after = httpx.get( + f"{restarted.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT + ).json() + assert any(row["asset_id"] == seeded.asset_ids["city"] for row in after["items"]) + finally: + restarted.stop() diff --git a/tests/e2e/test_worker_kill.py b/tests/e2e/test_worker_kill.py index 00ff9a7..913ff1d 100644 --- a/tests/e2e/test_worker_kill.py +++ b/tests/e2e/test_worker_kill.py @@ -14,6 +14,10 @@ from photo_pipeline.services.jobs import ItemState, JobService, JobState REPO = Path(__file__).resolve().parents[2] +# Worker-recovery is also part of the Phase B acceptance command (US02-07: the +# "resume" journey); it stays mapped to US02-03 for traceability. +pytestmark = pytest.mark.phase_b + # A worker that claims the job, marks its item running, signals it started, then # blocks — so the test can kill it mid-item. WORKER_SCRIPT = """ diff --git a/tests/story_traceability.json b/tests/story_traceability.json index c0a3bf2..6dfab03 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -64,6 +64,10 @@ "US02-06": [ "tests/integration/test_safety_analysis.py", "tests/e2e/test_workflow_views.py" + ], + "US02-07": [ + "tests/e2e/test_phase_b_pipeline.py", + "tests/e2e/test_analysis_browser.py" ] } } diff --git a/work_item/src/work_item/core.py b/work_item/src/work_item/core.py index 4c755e2..c7a2334 100644 --- a/work_item/src/work_item/core.py +++ b/work_item/src/work_item/core.py @@ -446,19 +446,24 @@ class Gitea: self.tea(*args) def claim(self, story: Story) -> None: - self.tea( - "issues", - "edit", - str(story.number), - "--repo", - self.config.repo_slug, - "--add-assignees", - self.config.assignee, - "--add-labels", - "status/in-progress", - "--remove-labels", - "status/backlog,status/ready,status/blocked,status/review,status/done", + self.edit_labels( + story.number, + add=("status/in-progress",), + remove=( + "status/backlog", + "status/ready", + "status/blocked", + "status/review", + "status/done", + ), ) + self.set_assignee(story.number, self.config.assignee) + + def set_assignee(self, issue: int, assignee: str) -> None: + # Some Gitea/Forgejo deployments 404 on the issues/{n}/assignees + # sub-route that `tea --add-assignees` uses; the issue PATCH endpoint + # accepts the full assignees list and works across those versions. + self.api(f"{self.base}/issues/{issue}", method="PATCH", data={"assignees": [assignee]}) def comment(self, issue: int, body: str) -> None: self.tea("comments", "add", str(issue), body, "--repo", self.config.repo_slug) diff --git a/work_item/tests/test_cli_e2e.py b/work_item/tests/test_cli_e2e.py index 3f4c566..f2892b7 100644 --- a/work_item/tests/test_cli_e2e.py +++ b/work_item/tests/test_cli_e2e.py @@ -50,6 +50,13 @@ if args[0] == "api": out(state["prs"][str(number)]) elif "/commits/" in endpoint and endpoint.endswith("/status"): out({"state": state.get("ci_state", "success")}) + elif "--method" in args and args[args.index("--method") + 1] == "PATCH" and "/issues/" in endpoint: + number = int(endpoint.rsplit("/", 1)[1]) + issue = next(x for x in state["issues"] if x["number"] == number) + payload = json.loads(args[args.index("--data") + 1]) + if "assignees" in payload: + issue["assignees"] = [{"login": login} for login in payload["assignees"]] + save(); out(issue) else: raise SystemExit(f"unsupported api: {endpoint}") elif args[:2] == ["issues", "edit"]: