From ac6baa9a0b88b62a22bc0301af1fc5e64a463e58 Mon Sep 17 00:00:00 2001 From: laclac102 Date: Tue, 3 Feb 2026 13:33:28 +1100 Subject: [PATCH 1/6] feat: job listing and sample db seed --- app/routes/workflows.py | 157 +++++++++++++- app/schemas/workflows.py | 53 +++++ app/services/seqera.py | 258 +++++++++++++++++++++++ db/sample_seed.sql | 77 +++++++ tests/test_routes_jobs.py | 267 ++++++++++++++++++++++++ tests/test_services_workflow_listing.py | 251 ++++++++++++++++++++++ 6 files changed, 1059 insertions(+), 4 deletions(-) create mode 100644 app/services/seqera.py create mode 100644 db/sample_seed.sql create mode 100644 tests/test_routes_jobs.py create mode 100644 tests/test_services_workflow_listing.py diff --git a/app/routes/workflows.py b/app/routes/workflows.py index 87ae44f..4122659 100644 --- a/app/routes/workflows.py +++ b/app/routes/workflows.py @@ -4,16 +4,25 @@ import asyncio from datetime import datetime, timezone +from uuid import UUID -from fastapi import APIRouter, HTTPException, Query, status +from fastapi import APIRouter, Depends, Header, HTTPException, Query, status +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..db import SessionLocal +from ..db.models.core import AppUser, WorkflowRun from ..schemas.workflows import ( CancelWorkflowResponse, DatasetUploadRequest, DatasetUploadResponse, + JobListItem, + JobListResponse, LaunchDetails, LaunchLogs, ListRunsResponse, + RunInfo, WorkflowLaunchPayload, WorkflowLaunchResponse, ) @@ -27,10 +36,51 @@ create_seqera_dataset, upload_dataset_to_seqera, ) +from ..services.seqera import ( + SeqeraAPIError, + SeqeraConfigurationError, + list_seqera_workflows, +) router = APIRouter(tags=["workflows"]) +def get_db() -> Session: + db = SessionLocal() + try: + yield db + finally: + db.close() + + +def get_current_user_id( + x_auth0_user_id: str | None = Header(None, alias="X-Auth0-User-Id"), + db: Session = Depends(get_db), +) -> UUID: + if not x_auth0_user_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing X-Auth0-User-Id header", + ) + + user = db.execute( + select(AppUser).where(AppUser.auth0_user_id == x_auth0_user_id) + ).scalar_one_or_none() + if not user: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Unknown user", + ) + return user.id + + +def get_owned_run_ids(db: Session, user_id: UUID) -> set[str]: + rows = db.execute( + select(WorkflowRun.seqera_run_id).where(WorkflowRun.owner_user_id == user_id) + ).all() + return {row[0] for row in rows if row[0]} + + @router.post("/launch", response_model=WorkflowLaunchResponse, status_code=status.HTTP_201_CREATED) async def launch_workflow(payload: WorkflowLaunchPayload) -> WorkflowLaunchResponse: """Launch a workflow on the Seqera Platform.""" @@ -70,10 +120,109 @@ async def list_runs( workspace: str | None = Query(None), limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), ) -> ListRunsResponse: - """List workflow runs (placeholder until Seqera list API integration).""" - _ = (status_filter, workspace) # Reserved for future Seqera integration - return ListRunsResponse(runs=[], total=0, limit=limit, offset=offset) + """List workflow runs with a compact response shape for the frontend.""" + parsed_status_filter = None + if status_filter: + parsed_status_filter = [item.strip() for item in status_filter.split(",") if item.strip()] + + try: + workflows, _total = await list_seqera_workflows( + workspace_id=workspace, + status_filter=parsed_status_filter, + limit=limit, + offset=offset, + ) + + owned_run_ids = get_owned_run_ids(db, current_user_id) + owned_workflows = [wf for wf in workflows if wf.workflow_id in owned_run_ids] + + runs = [ + RunInfo( + id=wf.workflow_id, + run=wf.run_name or wf.workflow_id, + workflow=wf.workflow_type or "", + status=wf.ui_status, + date=wf.submitted_at.isoformat() if wf.submitted_at else "", + cancel=f"/api/workflows/{wf.workflow_id}/cancel", + ) + for wf in owned_workflows + ] + return ListRunsResponse(runs=runs, total=len(runs), limit=limit, offset=offset) + except SeqeraConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=str(exc), + ) from exc + except SeqeraAPIError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=str(exc), + ) from exc + + +@router.get("/jobs", response_model=JobListResponse) +async def list_jobs( + search: str | None = Query(None, description="Search by job name or workflow type"), + status: list[str] | None = Query(None, description="Filter by status (Completed, Stopped, Failed)"), + limit: int = Query(50, ge=1, le=200, description="Maximum number of results"), + offset: int = Query(0, ge=0, description="Number of results to skip"), + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> JobListResponse: + """ + Retrieve a paginated list of workflow jobs with search and filtering. + + Query Parameters: + - search: Search by job name or workflow type + - status: Filter by status values (Completed, Stopped, Failed, In progress, In queue) + - limit: Maximum number of results per page + - offset: Number of results to skip for pagination + + Returns: + - Paginated list of jobs with job name, workflow type, status, submitted date, and score + """ + try: + workflows, _total = await list_seqera_workflows( + search_query=search, + status_filter=status, + limit=limit, + offset=offset, + ) + + owned_run_ids = get_owned_run_ids(db, current_user_id) + owned_workflows = [wf for wf in workflows if wf.workflow_id in owned_run_ids] + + jobs = [ + JobListItem( + id=wf.workflow_id, + jobName=wf.run_name or wf.workflow_id, + workflowType=wf.workflow_type, + status=wf.ui_status, + submittedAt=wf.submitted_at or datetime.now(timezone.utc), + score=wf.score, + ) + for wf in owned_workflows + ] + + return JobListResponse( + jobs=jobs, + total=len(jobs), + limit=limit, + offset=offset, + ) + except SeqeraConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=str(exc), + ) from exc + except SeqeraAPIError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=str(exc), + ) from exc @router.get("/{run_id}/logs", response_model=LaunchLogs) diff --git a/app/schemas/workflows.py b/app/schemas/workflows.py index 95f63a9..696462e 100644 --- a/app/schemas/workflows.py +++ b/app/schemas/workflows.py @@ -3,11 +3,44 @@ from __future__ import annotations from datetime import datetime +from enum import Enum from typing import Any from pydantic import BaseModel, ConfigDict, Field, field_validator +class PipelineStatus(str, Enum): + """Pipeline status values from Seqera Platform.""" + SUBMITTED = "SUBMITTED" + RUNNING = "RUNNING" + SUCCEEDED = "SUCCEEDED" + FAILED = "FAILED" + UNKNOWN = "UNKNOWN" + CANCELLED = "CANCELLED" + + +class UIStatus(str, Enum): + """User-facing status values for the frontend.""" + IN_QUEUE = "In queue" + IN_PROGRESS = "In progress" + COMPLETED = "Completed" + FAILED = "Failed" + STOPPED = "Stopped" + + +def map_pipeline_status_to_ui(pipeline_status: str) -> str: + """Map Seqera pipeline status to UI-friendly status.""" + status_mapping = { + PipelineStatus.SUBMITTED.value: UIStatus.IN_QUEUE.value, + PipelineStatus.RUNNING.value: UIStatus.IN_PROGRESS.value, + PipelineStatus.SUCCEEDED.value: UIStatus.COMPLETED.value, + PipelineStatus.FAILED.value: UIStatus.FAILED.value, + PipelineStatus.UNKNOWN.value: UIStatus.FAILED.value, + PipelineStatus.CANCELLED.value: UIStatus.STOPPED.value, + } + return status_mapping.get(pipeline_status, UIStatus.FAILED.value) + + class WorkflowLaunchForm(BaseModel): model_config = ConfigDict(extra="forbid") @@ -144,3 +177,23 @@ class PdbUploadResponse(BaseModel): fileName: str = Field(..., description="Original filename") s3Uri: str = Field(..., description="Full S3 URI (s3://bucket/key) for dataset creation") details: dict[str, Any] | None = Field(default=None, description="Additional upload details") + + +class JobListItem(BaseModel): + """Individual job item in the job listing.""" + + id: str = Field(..., description="Workflow run ID") + jobName: str = Field(..., description="Human-readable job name") + workflowType: str | None = Field(None, description="Workflow type (e.g., BindCraft, De novo design)") + status: str = Field(..., description="UI-friendly status (e.g., Completed, In progress)") + submittedAt: datetime = Field(..., description="Submission date and time") + score: float | None = Field(None, description="Job score/metric") + + +class JobListResponse(BaseModel): + """Paginated response for job listing.""" + + jobs: list[JobListItem] = Field(default_factory=list, description="List of jobs") + total: int = Field(..., description="Total number of jobs matching the criteria") + limit: int = Field(..., description="Maximum number of items per page") + offset: int = Field(..., description="Number of items skipped") diff --git a/app/services/seqera.py b/app/services/seqera.py new file mode 100644 index 0000000..8ffacaf --- /dev/null +++ b/app/services/seqera.py @@ -0,0 +1,258 @@ +"""Seqera Platform API integration for workflow operations.""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from datetime import datetime + +import httpx + +from ..schemas.workflows import map_pipeline_status_to_ui + +logger = logging.getLogger(__name__) + + +class SeqeraConfigurationError(RuntimeError): + """Raised when required Seqera configuration is missing.""" + + +class SeqeraAPIError(RuntimeError): + """Raised when Seqera API calls fail.""" + + +@dataclass +class WorkflowListItem: + """Individual workflow run item from Seqera API.""" + + workflow_id: str + run_name: str | None + workflow_type: str | None + pipeline_status: str + ui_status: str + submitted_at: datetime | None + score: float | None + + +def _masked_headers(headers: dict[str, str]) -> dict[str, str]: + """Mask sensitive headers before logging.""" + masked = dict(headers) + if "Authorization" in masked: + masked["Authorization"] = "Bearer ***" + return masked + + +def _get_required_env(key: str) -> str: + """Get required environment variable or raise error.""" + value = os.getenv(key) + if not value: + raise SeqeraConfigurationError(f"Missing required environment variable: {key}") + return value + + +async def list_seqera_workflows( + workspace_id: str | None = None, + search_query: str | None = None, + status_filter: list[str] | None = None, + limit: int = 50, + offset: int = 0, +) -> tuple[list[WorkflowListItem], int]: + """ + List workflow runs from Seqera Platform. + + Args: + workspace_id: Seqera workspace ID (uses env var if not provided) + search_query: Search term for job name or workflow type + status_filter: List of UI status values to filter by (e.g., ["Completed", "Failed"]) + limit: Maximum number of results to return + offset: Number of results to skip + + Returns: + Tuple of (list of workflow items, total count) + """ + seqera_api_url = _get_required_env("SEQERA_API_URL").rstrip("/") + seqera_token = _get_required_env("SEQERA_ACCESS_TOKEN") + + if not workspace_id: + workspace_id = os.getenv("WORK_SPACE") + + # Build query parameters. Keep workspace optional to match plain GET /workflow usage. + params: dict[str, str | int] = {} + if workspace_id: + params["workspaceId"] = workspace_id + + # Add search query if provided + if search_query: + params["search"] = search_query + + url = f"{seqera_api_url}/workflow" + + headers = { + "Authorization": f"Bearer {seqera_token}", + "Accept": "application/json", + } + + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + response = await client.get(url, headers=headers, params=params) + + + if response.is_error: + body = response.text + logger.error( + "Seqera API error", + extra={ + "status": response.status_code, + "reason": response.reason_phrase, + "body": body, + }, + ) + raise SeqeraAPIError( + f"Failed to list workflows: {response.status_code} {body}" + ) + + data = response.json() + + logger.info( + "Parsed workflow data from Seqera", + extra={ + "total_workflows": len(data.get("workflows", [])), + "total_size": data.get("totalSize", 0), + }, + ) + + # Support both Seqera payload shapes: + # 1) {"workflows": [...], "totalSize": N} + # 2) [{"workflow": {...}}, ...] + if isinstance(data, dict): + workflows_data = data.get("workflows") or data.get("items") or [] + total_count = data.get("totalSize") or data.get("total") or len(workflows_data) + elif isinstance(data, list): + workflows_data = data + total_count = len(workflows_data) + else: + workflows_data = [] + total_count = 0 + + workflow_items: list[WorkflowListItem] = [] + + for item in workflows_data: + # Some Seqera responses wrap run data in {"workflow": {...}} + wf = item.get("workflow", item) if isinstance(item, dict) else {} + + pipeline_status = wf.get("status", "UNKNOWN") + ui_status = map_pipeline_status_to_ui(pipeline_status) + + # Apply status filter if provided + if status_filter and ui_status not in status_filter: + continue + + # Parse submitted date + submitted_at = None + if submit_str := wf.get("submit") or wf.get("dateCreated"): + try: + submitted_at = datetime.fromisoformat(submit_str.replace("Z", "+00:00")) + except (ValueError, AttributeError): + logger.warning(f"Could not parse submit time: {submit_str}") + + workflow_items.append( + WorkflowListItem( + workflow_id=wf.get("id", ""), + run_name=wf.get("runName"), + workflow_type=_extract_workflow_type(wf), + pipeline_status=pipeline_status, + ui_status=ui_status, + submitted_at=submitted_at, + score=None + ) + ) + + logger.info( + f"Returning {len(workflow_items)} workflow items after filtering", + extra={"requested_status_filter": status_filter}, + ) + + return workflow_items, total_count + + +def _extract_workflow_type(workflow_data: dict) -> str | None: + """ + Extract workflow type from workflow data. + + This could be based on: + - Pipeline name/repository + - Project name + - Custom metadata + """ + pipeline = workflow_data.get("projectName") or workflow_data.get("pipeline", "") + + # Map common pipeline names to workflow types + if "bindcraft" in pipeline.lower(): + return "BindCraft" + elif "denovo" in pipeline.lower() or "de-novo" in pipeline.lower(): + return "De novo design" + + return pipeline or None + + +async def describe_workflow(workflow_id: str, workspace_id: str | None = None) -> dict: + """ + Get detailed information about a specific workflow run. + + Args: + workflow_id: Seqera workflow run ID + workspace_id: Seqera workspace ID (uses env var if not provided) + + Returns: + Workflow details dictionary + """ + seqera_api_url = _get_required_env("SEQERA_API_URL").rstrip("/") + seqera_token = _get_required_env("SEQERA_ACCESS_TOKEN") + + if not workspace_id: + workspace_id = os.getenv("WORK_SPACE") + + url = f"{seqera_api_url}/workflow/{workflow_id}" + params: dict[str, str] = {} + if workspace_id: + params["workspaceId"] = workspace_id + + headers = { + "Authorization": f"Bearer {seqera_token}", + "Accept": "application/json", + } + + logger.info( + "Describing workflow from Seqera API: url=%s workflow_id=%s workspace_id=%s params=%s headers=%s", + url, + workflow_id, + workspace_id, + params, + _masked_headers(headers), + ) + + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + response = await client.get(url, headers=headers, params=params) + + logger.info( + "Seqera describe API response: method=GET path=/workflow/%s status=%s raw_body=%s", + workflow_id, + response.status_code, + response.text if response.text else None, + ) + + if response.is_error: + body = response.text + logger.error( + "Seqera API error", + extra={ + "status": response.status_code, + "reason": response.reason_phrase, + "body": body, + }, + ) + raise SeqeraAPIError( + f"Failed to describe workflow: {response.status_code} {body}" + ) + + return response.json() diff --git a/db/sample_seed.sql b/db/sample_seed.sql new file mode 100644 index 0000000..989d0cf --- /dev/null +++ b/db/sample_seed.sql @@ -0,0 +1,77 @@ +-- Sample seed data for local testing of workflow ownership filters. +-- Safe to run multiple times (uses ON CONFLICT upserts where possible). + +BEGIN; + +-- Users used by the X-Auth0-User-Id header in API calls +INSERT INTO app_users (id, auth0_user_id, name, email) +VALUES + ('11111111-1111-1111-1111-111111111111', 'auth0|alice-dev', 'Alice Dev', 'alice@example.com'), + ('22222222-2222-2222-2222-222222222222', 'auth0|bob-dev', 'Bob Dev', 'bob@example.com') +ON CONFLICT (auth0_user_id) DO UPDATE +SET + name = EXCLUDED.name, + email = EXCLUDED.email; + +INSERT INTO workflows (id, name, description, repo_url, default_revision) +VALUES + ( + 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', + 'Bindflow', + 'Sample Bindflow workflow for local ownership tests', + 'https://github.com/Australian-Structural-Biology-Computing/bindflow', + 'dev' + ) +ON CONFLICT (id) DO NOTHING; + +-- Runs owned by Alice +INSERT INTO workflow_runs ( + id, workflow_id, owner_user_id, seqera_dataset_id, seqera_run_id, run_name, work_dir +) +VALUES + ( + 'aaaa1111-aaaa-1111-aaaa-111111111111', + 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', + '11111111-1111-1111-1111-111111111111', + 'dataset-alice-1', + '292kB09WsB4RyC', + 'anne_test_2', + 1001 + ), + ( + 'aaaa2222-aaaa-2222-aaaa-222222222222', + 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', + '11111111-1111-1111-1111-111111111111', + 'dataset-alice-2', + 'B5U7ok7AAyCpV', + 'naughty_goldstine', + 1002 + ) +ON CONFLICT (seqera_run_id) DO NOTHING; + +-- Runs owned by Bob +INSERT INTO workflow_runs ( + id, workflow_id, owner_user_id, seqera_dataset_id, seqera_run_id, run_name, work_dir +) +VALUES + ( + 'bbbb1111-bbbb-1111-bbbb-111111111111', + 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', + '22222222-2222-2222-2222-222222222222', + 'dataset-bob-1', + '30Nta1wQivKcS', + 'marvelous_lamarck', + 2001 + ) +ON CONFLICT (seqera_run_id) DO NOTHING; + +-- Optional metrics for score field +INSERT INTO run_metrics (run_id, max_score) +VALUES + ('aaaa1111-aaaa-1111-aaaa-111111111111', 0.93), + ('aaaa2222-aaaa-2222-aaaa-222222222222', 0.71), + ('bbbb1111-bbbb-1111-bbbb-111111111111', 0.62) +ON CONFLICT (run_id) DO UPDATE +SET max_score = EXCLUDED.max_score; + +COMMIT; diff --git a/tests/test_routes_jobs.py b/tests/test_routes_jobs.py new file mode 100644 index 0000000..f60573a --- /dev/null +++ b/tests/test_routes_jobs.py @@ -0,0 +1,267 @@ +"""Tests for job listing API endpoint.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi.testclient import TestClient + +from app.main import create_app +from app.services.seqera import WorkflowListItem + + +@pytest.fixture +def client(): + """Create test client.""" + app = create_app() + return TestClient(app) + + +def test_list_jobs_endpoint_success(client, monkeypatch): + """Test successful job listing via API endpoint.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_workflows = [ + WorkflowListItem( + workflow_id="wf-123", + run_name="Test Job 1", + workflow_type="BindCraft", + pipeline_status="SUCCEEDED", + ui_status="Completed", + submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), + score=0.95, + ), + WorkflowListItem( + workflow_id="wf-456", + run_name="Test Job 2", + workflow_type="De novo design", + pipeline_status="RUNNING", + ui_status="In progress", + submitted_at=datetime(2026, 2, 2, 11, 0, 0, tzinfo=timezone.utc), + score=None, + ), + ] + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + return_value=(mock_workflows, 2), + ): + response = client.get("/jobs?limit=10&offset=0") + + assert response.status_code == 200 + data = response.json() + + assert data["total"] == 2 + assert data["limit"] == 10 + assert data["offset"] == 0 + assert len(data["jobs"]) == 2 + + job1 = data["jobs"][0] + assert job1["id"] == "wf-123" + assert job1["jobName"] == "Test Job 1" + assert job1["workflowType"] == "BindCraft" + assert job1["status"] == "Completed" + assert job1["score"] == 0.95 + + job2 = data["jobs"][1] + assert job2["id"] == "wf-456" + assert job2["jobName"] == "Test Job 2" + assert job2["workflowType"] == "De novo design" + assert job2["status"] == "In progress" + assert job2["score"] is None + + +def test_list_jobs_with_search_query(client, monkeypatch): + """Test job listing with search query.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_workflows = [ + WorkflowListItem( + workflow_id="wf-789", + run_name="Matching Job", + workflow_type="BindCraft", + pipeline_status="SUCCEEDED", + ui_status="Completed", + submitted_at=datetime(2026, 2, 3, 12, 0, 0, tzinfo=timezone.utc), + score=None, + ) + ] + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + return_value=(mock_workflows, 1), + ) as mock_list: + response = client.get("/jobs?search=Matching&limit=10&offset=0") + + assert response.status_code == 200 + data = response.json() + + # Verify the service was called with search parameter + mock_list.assert_called_once() + call_kwargs = mock_list.call_args.kwargs + assert call_kwargs["search_query"] == "Matching" + + assert data["total"] == 1 + assert len(data["jobs"]) == 1 + assert data["jobs"][0]["jobName"] == "Matching Job" + + +def test_list_jobs_with_status_filter(client, monkeypatch): + """Test job listing with status filter.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_workflows = [ + WorkflowListItem( + workflow_id="wf-100", + run_name="Completed Job", + workflow_type="BindCraft", + pipeline_status="SUCCEEDED", + ui_status="Completed", + submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), + score=None, + ) + ] + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + return_value=(mock_workflows, 1), + ) as mock_list: + response = client.get("/jobs?status=Completed&status=Failed&limit=10&offset=0") + + assert response.status_code == 200 + data = response.json() + + # Verify the service was called with status filter + mock_list.assert_called_once() + call_kwargs = mock_list.call_args.kwargs + assert call_kwargs["status_filter"] == ["Completed", "Failed"] + + assert data["total"] == 1 + assert len(data["jobs"]) == 1 + assert data["jobs"][0]["status"] == "Completed" + + +def test_list_jobs_with_pagination(client, monkeypatch): + """Test job listing with pagination parameters.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_workflows = [ + WorkflowListItem( + workflow_id=f"wf-{i}", + run_name=f"Job {i}", + workflow_type="BindCraft", + pipeline_status="SUCCEEDED", + ui_status="Completed", + submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), + score=None, + ) + for i in range(20, 40) + ] + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + return_value=(mock_workflows, 100), + ) as mock_list: + response = client.get("/jobs?limit=20&offset=20") + + assert response.status_code == 200 + data = response.json() + + # Verify the service was called with correct pagination + mock_list.assert_called_once() + call_kwargs = mock_list.call_args.kwargs + assert call_kwargs["limit"] == 20 + assert call_kwargs["offset"] == 20 + + assert data["total"] == 100 + assert data["limit"] == 20 + assert data["offset"] == 20 + assert len(data["jobs"]) == 20 + + +def test_list_jobs_empty_results(client, monkeypatch): + """Test job listing with no results.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + return_value=([], 0), + ): + response = client.get("/jobs?limit=10&offset=0") + + assert response.status_code == 200 + data = response.json() + + assert data["total"] == 0 + assert data["limit"] == 10 + assert data["offset"] == 0 + assert len(data["jobs"]) == 0 + + +def test_list_jobs_invalid_limit(client, monkeypatch): + """Test job listing with invalid limit parameter.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + response = client.get("/jobs?limit=0") + assert response.status_code == 422 # Validation error + + response = client.get("/jobs?limit=500") + assert response.status_code == 422 # Exceeds max limit + + +def test_list_jobs_invalid_offset(client, monkeypatch): + """Test job listing with invalid offset parameter.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + response = client.get("/jobs?offset=-1") + assert response.status_code == 422 # Validation error + + +def test_list_jobs_configuration_error(client, monkeypatch): + """Test job listing with missing configuration.""" + # Don't set required environment variables + monkeypatch.delenv("SEQERA_API_URL", raising=False) + + response = client.get("/jobs?limit=10&offset=0") + assert response.status_code == 500 + assert "SEQERA_API_URL" in response.json()["detail"] + + +def test_list_jobs_api_error(client, monkeypatch): + """Test job listing when Seqera API fails.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + from app.services.seqera import SeqeraAPIError + + with patch( + "app.routes.workflows.list_seqera_workflows", + new_callable=AsyncMock, + side_effect=SeqeraAPIError("Seqera API is down"), + ): + response = client.get("/jobs?limit=10&offset=0") + + assert response.status_code == 502 + assert "Seqera API is down" in response.json()["detail"] diff --git a/tests/test_services_workflow_listing.py b/tests/test_services_workflow_listing.py new file mode 100644 index 0000000..cc8bbb7 --- /dev/null +++ b/tests/test_services_workflow_listing.py @@ -0,0 +1,251 @@ +"""Tests for Seqera workflow listing service.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from app.services.seqera import ( + SeqeraAPIError, + SeqeraConfigurationError, + WorkflowListItem, + describe_workflow, + list_seqera_workflows, +) + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_success(monkeypatch): + """Test successful workflow listing from Seqera API.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response_data = { + "workflows": [ + { + "id": "wf-123", + "runName": "Test BindCraft Run", + "projectName": "bindcraft-pipeline", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + }, + { + "id": "wf-456", + "runName": "De novo design test", + "projectName": "denovo-pipeline", + "status": "RUNNING", + "submit": "2026-02-02T11:00:00Z", + }, + ], + "totalSize": 2, + } + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = False + mock_response.json.return_value = mock_response_data + + with patch("httpx.AsyncClient.get", return_value=mock_response): + workflows, total = await list_seqera_workflows(limit=10, offset=0) + + assert total == 2 + assert len(workflows) == 2 + assert workflows[0].workflow_id == "wf-123" + assert workflows[0].run_name == "Test BindCraft Run" + assert workflows[0].workflow_type == "BindCraft" + assert workflows[0].ui_status == "Completed" + assert workflows[0].pipeline_status == "SUCCEEDED" + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_with_search(monkeypatch): + """Test workflow listing with search query.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response_data = { + "workflows": [ + { + "id": "wf-789", + "runName": "Matching Job", + "projectName": "test-pipeline", + "status": "FAILED", + "submit": "2026-02-03T12:00:00Z", + } + ], + "totalSize": 1, + } + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = False + mock_response.json.return_value = mock_response_data + + with patch("httpx.AsyncClient.get", return_value=mock_response) as mock_get: + workflows, total = await list_seqera_workflows(search_query="Matching", limit=10, offset=0) + + # Verify search parameter was passed + call_args = mock_get.call_args + assert call_args.kwargs["params"]["search"] == "Matching" + + assert total == 1 + assert len(workflows) == 1 + assert workflows[0].run_name == "Matching Job" + assert workflows[0].ui_status == "Failed" + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_with_status_filter(monkeypatch): + """Test workflow listing with status filter.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response_data = { + "workflows": [ + { + "id": "wf-100", + "runName": "Completed Job", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + }, + { + "id": "wf-101", + "runName": "Failed Job", + "status": "FAILED", + "submit": "2026-02-02T11:00:00Z", + }, + ], + "totalSize": 2, + } + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = False + mock_response.json.return_value = mock_response_data + + with patch("httpx.AsyncClient.get", return_value=mock_response): + workflows, total = await list_seqera_workflows( + status_filter=["Completed"], limit=10, offset=0 + ) + + # Only completed workflows should be returned + assert len(workflows) == 1 + assert workflows[0].ui_status == "Completed" + assert workflows[0].workflow_id == "wf-100" + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_status_mapping(monkeypatch): + """Test that all pipeline statuses are correctly mapped to UI statuses.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response_data = { + "workflows": [ + {"id": "wf-1", "runName": "Job 1", "status": "SUBMITTED"}, + {"id": "wf-2", "runName": "Job 2", "status": "RUNNING"}, + {"id": "wf-3", "runName": "Job 3", "status": "SUCCEEDED"}, + {"id": "wf-4", "runName": "Job 4", "status": "FAILED"}, + {"id": "wf-5", "runName": "Job 5", "status": "CANCELLED"}, + {"id": "wf-6", "runName": "Job 6", "status": "UNKNOWN"}, + ], + "totalSize": 6, + } + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = False + mock_response.json.return_value = mock_response_data + + with patch("httpx.AsyncClient.get", return_value=mock_response): + workflows, total = await list_seqera_workflows(limit=10, offset=0) + + assert len(workflows) == 6 + + status_mapping = { + "SUBMITTED": "In queue", + "RUNNING": "In progress", + "SUCCEEDED": "Completed", + "FAILED": "Failed", + "CANCELLED": "Stopped", + "UNKNOWN": "Failed", + } + + for wf in workflows: + expected_ui_status = status_mapping[wf.pipeline_status] + assert wf.ui_status == expected_ui_status + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_missing_config(monkeypatch): + """Test that missing configuration raises appropriate error.""" + monkeypatch.delenv("SEQERA_API_URL", raising=False) + + with pytest.raises(SeqeraConfigurationError, match="SEQERA_API_URL"): + await list_seqera_workflows() + + +@pytest.mark.asyncio +async def test_list_seqera_workflows_api_error(monkeypatch): + """Test handling of Seqera API errors.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = True + mock_response.status_code = 500 + mock_response.reason_phrase = "Internal Server Error" + mock_response.text = "Server error occurred" + + with patch("httpx.AsyncClient.get", return_value=mock_response): + with pytest.raises(SeqeraAPIError, match="Failed to list workflows"): + await list_seqera_workflows() + + +@pytest.mark.asyncio +async def test_describe_workflow_success(monkeypatch): + """Test describing a specific workflow.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response_data = { + "workflow": { + "id": "wf-123", + "runName": "Test Run", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + "complete": "2026-02-01T11:30:00Z", + } + } + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = False + mock_response.json.return_value = mock_response_data + + with patch("httpx.AsyncClient.get", return_value=mock_response): + result = await describe_workflow("wf-123") + + assert result == mock_response_data + + +@pytest.mark.asyncio +async def test_describe_workflow_not_found(monkeypatch): + """Test describing a non-existent workflow.""" + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") + monkeypatch.setenv("WORK_SPACE", "test-workspace") + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.is_error = True + mock_response.status_code = 404 + mock_response.reason_phrase = "Not Found" + mock_response.text = "Workflow not found" + + with patch("httpx.AsyncClient.get", return_value=mock_response): + with pytest.raises(SeqeraAPIError, match="Failed to describe workflow"): + await describe_workflow("wf-nonexistent") From 43a960ccd4064ae3329cee200ad46e7a7c7990c0 Mon Sep 17 00:00:00 2001 From: laclac102 Date: Tue, 3 Feb 2026 15:11:25 +1100 Subject: [PATCH 2/6] fix: remove sample_seed.sql and add to .gitignore --- .gitignore | 3 ++ db/sample_seed.sql | 77 ---------------------------------------------- 2 files changed, 3 insertions(+), 77 deletions(-) delete mode 100644 db/sample_seed.sql diff --git a/.gitignore b/.gitignore index 9a26805..1a340b6 100644 --- a/.gitignore +++ b/.gitignore @@ -39,3 +39,6 @@ logs/ # UV # Note: uv.lock is tracked for reproducibility # .python-version is tracked if using specific Python version + +# Database +db/sample_seed.sql diff --git a/db/sample_seed.sql b/db/sample_seed.sql deleted file mode 100644 index 989d0cf..0000000 --- a/db/sample_seed.sql +++ /dev/null @@ -1,77 +0,0 @@ --- Sample seed data for local testing of workflow ownership filters. --- Safe to run multiple times (uses ON CONFLICT upserts where possible). - -BEGIN; - --- Users used by the X-Auth0-User-Id header in API calls -INSERT INTO app_users (id, auth0_user_id, name, email) -VALUES - ('11111111-1111-1111-1111-111111111111', 'auth0|alice-dev', 'Alice Dev', 'alice@example.com'), - ('22222222-2222-2222-2222-222222222222', 'auth0|bob-dev', 'Bob Dev', 'bob@example.com') -ON CONFLICT (auth0_user_id) DO UPDATE -SET - name = EXCLUDED.name, - email = EXCLUDED.email; - -INSERT INTO workflows (id, name, description, repo_url, default_revision) -VALUES - ( - 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', - 'Bindflow', - 'Sample Bindflow workflow for local ownership tests', - 'https://github.com/Australian-Structural-Biology-Computing/bindflow', - 'dev' - ) -ON CONFLICT (id) DO NOTHING; - --- Runs owned by Alice -INSERT INTO workflow_runs ( - id, workflow_id, owner_user_id, seqera_dataset_id, seqera_run_id, run_name, work_dir -) -VALUES - ( - 'aaaa1111-aaaa-1111-aaaa-111111111111', - 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', - '11111111-1111-1111-1111-111111111111', - 'dataset-alice-1', - '292kB09WsB4RyC', - 'anne_test_2', - 1001 - ), - ( - 'aaaa2222-aaaa-2222-aaaa-222222222222', - 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', - '11111111-1111-1111-1111-111111111111', - 'dataset-alice-2', - 'B5U7ok7AAyCpV', - 'naughty_goldstine', - 1002 - ) -ON CONFLICT (seqera_run_id) DO NOTHING; - --- Runs owned by Bob -INSERT INTO workflow_runs ( - id, workflow_id, owner_user_id, seqera_dataset_id, seqera_run_id, run_name, work_dir -) -VALUES - ( - 'bbbb1111-bbbb-1111-bbbb-111111111111', - 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', - '22222222-2222-2222-2222-222222222222', - 'dataset-bob-1', - '30Nta1wQivKcS', - 'marvelous_lamarck', - 2001 - ) -ON CONFLICT (seqera_run_id) DO NOTHING; - --- Optional metrics for score field -INSERT INTO run_metrics (run_id, max_score) -VALUES - ('aaaa1111-aaaa-1111-aaaa-111111111111', 0.93), - ('aaaa2222-aaaa-2222-aaaa-222222222222', 0.71), - ('bbbb1111-bbbb-1111-bbbb-111111111111', 0.62) -ON CONFLICT (run_id) DO UPDATE -SET max_score = EXCLUDED.max_score; - -COMMIT; From 0b6dfc401d7b018aeaa6249b72c21327b8af2288 Mon Sep 17 00:00:00 2001 From: laclac102 Date: Tue, 3 Feb 2026 16:21:02 +1100 Subject: [PATCH 3/6] feat: delete, cancel jobs --- app/routes/dependencies.py | 42 ++++ app/routes/workflow/__init__.py | 1 + app/routes/workflow/jobs.py | 245 ++++++++++++++++++++++ app/routes/workflow/launch.py | 80 +++++++ app/routes/workflow/placeholders.py | 57 +++++ app/routes/workflows.py | 309 +--------------------------- app/schemas/workflows.py | 33 +++ app/services/job_utils.py | 99 +++++++++ app/services/seqera.py | 268 +++--------------------- app/services/seqera_client.py | 89 ++++++++ app/services/seqera_errors.py | 9 + app/services/seqera_models.py | 19 ++ app/services/seqera_parsers.py | 74 +++++++ 13 files changed, 786 insertions(+), 539 deletions(-) create mode 100644 app/routes/dependencies.py create mode 100644 app/routes/workflow/__init__.py create mode 100644 app/routes/workflow/jobs.py create mode 100644 app/routes/workflow/launch.py create mode 100644 app/routes/workflow/placeholders.py create mode 100644 app/services/job_utils.py create mode 100644 app/services/seqera_client.py create mode 100644 app/services/seqera_errors.py create mode 100644 app/services/seqera_models.py create mode 100644 app/services/seqera_parsers.py diff --git a/app/routes/dependencies.py b/app/routes/dependencies.py new file mode 100644 index 0000000..5f0374c --- /dev/null +++ b/app/routes/dependencies.py @@ -0,0 +1,42 @@ +"""Shared FastAPI dependencies for route modules.""" + +from __future__ import annotations + +from uuid import UUID + +from fastapi import Depends, Header, HTTPException, status +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..db import SessionLocal +from ..db.models.core import AppUser + + +def get_db() -> Session: + db = SessionLocal() + try: + yield db + finally: + db.close() + + +def get_current_user_id( + x_auth0_user_id: str | None = Header(None, alias="X-Auth0-User-Id"), + db: Session = Depends(get_db), +) -> UUID: + if not x_auth0_user_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing X-Auth0-User-Id header", + ) + + user = db.execute( + select(AppUser).where(AppUser.auth0_user_id == x_auth0_user_id) + ).scalar_one_or_none() + if not user: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Unknown user", + ) + + return user.id diff --git a/app/routes/workflow/__init__.py b/app/routes/workflow/__init__.py new file mode 100644 index 0000000..8c9b802 --- /dev/null +++ b/app/routes/workflow/__init__.py @@ -0,0 +1 @@ +"""Workflow route modules.""" diff --git a/app/routes/workflow/jobs.py b/app/routes/workflow/jobs.py new file mode 100644 index 0000000..0ff2427 --- /dev/null +++ b/app/routes/workflow/jobs.py @@ -0,0 +1,245 @@ +"""Job listing/detail/deletion endpoints.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import delete +from sqlalchemy.orm import Session + +from ...db.models.core import RunInput, RunMetric, RunOutput +from ...schemas.workflows import ( + BulkDeleteJobsRequest, + BulkDeleteJobsResponse, + CancelWorkflowResponse, + DeleteJobResponse, + JobDetailsResponse, + JobListItem, + JobListResponse, + map_pipeline_status_to_ui, +) +from ...services.job_utils import ( + coerce_workflow_payload, + ensure_completed_run_score, + extract_pipeline_status, + get_owned_run, + get_owned_run_ids, + get_score_by_seqera_run_id, + get_workflow_type_by_seqera_run_id, + parse_submit_datetime, +) +from ...services.seqera import ( + SeqeraAPIError, + SeqeraConfigurationError, + cancel_seqera_workflow, + delete_seqera_workflow, + describe_workflow, +) +from ..dependencies import get_current_user_id, get_db + +router = APIRouter() + + +@router.post("/{run_id}/cancel", response_model=CancelWorkflowResponse) +async def cancel_workflow( + run_id: str, + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> CancelWorkflowResponse: + """Cancel a workflow run.""" + owned_run = get_owned_run(db, current_user_id, run_id) + if not owned_run: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + + try: + await cancel_seqera_workflow(run_id) + except SeqeraAPIError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + return CancelWorkflowResponse( + message="Workflow cancelled successfully", + runId=run_id, + status="cancelled", + ) + + +@router.get("/jobs", response_model=JobListResponse) +async def list_jobs( + search: str | None = Query(None, description="Search by job name or workflow type"), + status: list[str] | None = Query(None, description="Filter by status (Completed, Stopped, Failed)"), + limit: int = Query(50, ge=1, le=200, description="Maximum number of results"), + offset: int = Query(0, ge=0, description="Number of results to skip"), + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> JobListResponse: + """Retrieve a paginated list of the current user's jobs with search and filtering.""" + try: + owned_run_ids = get_owned_run_ids(db, current_user_id) + score_by_run_id = get_score_by_seqera_run_id(db, current_user_id) + workflow_type_by_run_id = get_workflow_type_by_seqera_run_id(db, current_user_id) + search_text = (search or "").strip().lower() + allowed_statuses = set(status or []) + jobs: list[JobListItem] = [] + for run_id in owned_run_ids: + payload = await describe_workflow(run_id) + wf = coerce_workflow_payload(payload) + pipeline_status = extract_pipeline_status(payload) + ui_status = map_pipeline_status_to_ui(pipeline_status) + + if allowed_statuses and ui_status not in allowed_statuses: + continue + + workflow_type = workflow_type_by_run_id.get(run_id) + job_name = wf.get("runName") or run_id + if search_text and search_text not in str(job_name).lower() and search_text not in str(workflow_type or "").lower(): + continue + + score = score_by_run_id.get(run_id) + owned_run = get_owned_run(db, current_user_id, run_id) + if score is None and owned_run: + score = await ensure_completed_run_score(db, owned_run, ui_status) + + jobs.append( + JobListItem( + id=run_id, + jobName=job_name, + workflowType=workflow_type, + status=ui_status, + submittedAt=parse_submit_datetime(payload) or datetime.now(timezone.utc), + score=score if ui_status == "Completed" else None, + ) + ) + + jobs.sort(key=lambda item: item.submittedAt, reverse=True) + total = len(jobs) + jobs = jobs[offset : offset + limit] + + return JobListResponse( + jobs=jobs, + total=total, + limit=limit, + offset=offset, + ) + except SeqeraConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=str(exc), + ) from exc + except SeqeraAPIError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=str(exc), + ) from exc + + +@router.get("/jobs/{run_id}", response_model=JobDetailsResponse) +async def get_job_details( + run_id: str, + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> JobDetailsResponse: + """Retrieve a single job with normalized status and score.""" + owned_run = get_owned_run(db, current_user_id, run_id) + if not owned_run: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + + try: + payload = await describe_workflow(run_id) + except SeqeraConfigurationError as exc: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc + except SeqeraAPIError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + wf = coerce_workflow_payload(payload) + pipeline_status = extract_pipeline_status(payload) + ui_status = map_pipeline_status_to_ui(pipeline_status) + submitted_at = parse_submit_datetime(payload) or datetime.now(timezone.utc) + + score = await ensure_completed_run_score(db, owned_run, ui_status) + if ui_status != "Completed": + score = None + + return JobDetailsResponse( + id=run_id, + jobName=wf.get("runName") or run_id, + workflowType=(owned_run.workflow.name if owned_run.workflow else None), + status=ui_status, + submittedAt=submitted_at, + score=score, + ) + + +@router.delete("/jobs/{run_id}", response_model=DeleteJobResponse) +async def delete_job( + run_id: str, + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> DeleteJobResponse: + """Delete a single job. Running jobs are cancelled before deletion.""" + owned_run = get_owned_run(db, current_user_id, run_id) + if not owned_run: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + + cancelled = False + try: + payload = await describe_workflow(run_id) + pipeline_status = extract_pipeline_status(payload) + if pipeline_status in {"SUBMITTED", "RUNNING"}: + await cancel_seqera_workflow(run_id) + cancelled = True + + await delete_seqera_workflow(run_id) + except SeqeraConfigurationError as exc: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc + except SeqeraAPIError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + db.execute(delete(RunMetric).where(RunMetric.run_id == owned_run.id)) + db.execute(delete(RunInput).where(RunInput.run_id == owned_run.id)) + db.execute(delete(RunOutput).where(RunOutput.run_id == owned_run.id)) + db.delete(owned_run) + db.commit() + + return DeleteJobResponse( + runId=run_id, + deleted=True, + cancelledBeforeDelete=cancelled, + message="Job deleted successfully", + ) + + +@router.post("/jobs/bulk-delete", response_model=BulkDeleteJobsResponse) +async def bulk_delete_jobs( + payload: BulkDeleteJobsRequest, + current_user_id: UUID = Depends(get_current_user_id), + db: Session = Depends(get_db), +) -> BulkDeleteJobsResponse: + """Delete multiple jobs. Each running job is cancelled before deletion.""" + deleted: list[str] = [] + failed: dict[str, str] = {} + + for run_id in payload.runIds: + owned_run = get_owned_run(db, current_user_id, run_id) + if not owned_run: + failed[run_id] = "Job not found" + continue + + try: + details = await describe_workflow(run_id) + if extract_pipeline_status(details) in {"SUBMITTED", "RUNNING"}: + await cancel_seqera_workflow(run_id) + await delete_seqera_workflow(run_id) + + db.execute(delete(RunMetric).where(RunMetric.run_id == owned_run.id)) + db.execute(delete(RunInput).where(RunInput.run_id == owned_run.id)) + db.execute(delete(RunOutput).where(RunOutput.run_id == owned_run.id)) + db.delete(owned_run) + db.commit() + deleted.append(run_id) + except (SeqeraConfigurationError, SeqeraAPIError) as exc: + db.rollback() + failed[run_id] = str(exc) + + return BulkDeleteJobsResponse(deleted=deleted, failed=failed) diff --git a/app/routes/workflow/launch.py b/app/routes/workflow/launch.py new file mode 100644 index 0000000..c1677fb --- /dev/null +++ b/app/routes/workflow/launch.py @@ -0,0 +1,80 @@ +"""Launch and dataset workflow endpoints.""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +from fastapi import APIRouter, HTTPException, status + +from ...schemas.workflows import ( + DatasetUploadRequest, + DatasetUploadResponse, + WorkflowLaunchPayload, + WorkflowLaunchResponse, +) +from ...services.bindflow_executor import ( + BindflowConfigurationError, + BindflowExecutorError, + BindflowLaunchResult, + launch_bindflow_workflow, +) +from ...services.datasets import create_seqera_dataset, upload_dataset_to_seqera + +router = APIRouter() + + +@router.post("/launch", response_model=WorkflowLaunchResponse, status_code=status.HTTP_201_CREATED) +async def launch_workflow(payload: WorkflowLaunchPayload) -> WorkflowLaunchResponse: + """Launch a workflow on the Seqera Platform.""" + try: + dataset_id = payload.datasetId + result: BindflowLaunchResult = await launch_bindflow_workflow(payload.launch, dataset_id) + except BindflowConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + except BindflowExecutorError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + return WorkflowLaunchResponse( + message="Workflow launched successfully", + runId=result.workflow_id, + status=result.status, + submitTime=datetime.now(timezone.utc), + ) + + +@router.post("/datasets/upload", response_model=DatasetUploadResponse) +async def upload_dataset(payload: DatasetUploadRequest) -> DatasetUploadResponse: + """Create a Seqera dataset and upload form data as CSV content.""" + try: + dataset = await create_seqera_dataset( + name=payload.datasetName, description=payload.datasetDescription + ) + except BindflowConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + except BindflowExecutorError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + await asyncio.sleep(2) + + try: + upload_result = await upload_dataset_to_seqera(dataset.dataset_id, payload.formData) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + except BindflowConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + except BindflowExecutorError as exc: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc + + return DatasetUploadResponse( + message="Dataset created and uploaded successfully", + datasetId=upload_result.dataset_id, + success=upload_result.success, + details=upload_result.raw_response, + ) diff --git a/app/routes/workflow/placeholders.py b/app/routes/workflow/placeholders.py new file mode 100644 index 0000000..a19827d --- /dev/null +++ b/app/routes/workflow/placeholders.py @@ -0,0 +1,57 @@ +"""Placeholder endpoints kept for compatibility.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from fastapi import APIRouter + +from ...schemas.workflows import LaunchDetails, LaunchLogs + +router = APIRouter() + + +@router.get("/{run_id}/logs", response_model=LaunchLogs) +async def get_logs() -> LaunchLogs: + """Retrieve workflow logs (placeholder).""" + return LaunchLogs( + truncated=False, + entries=[], + rewindToken="", + forwardToken="", + pending=False, + message="Logs endpoint - implementation pending", + downloads=[], + ) + + +@router.get("/{run_id}/details", response_model=LaunchDetails) +async def get_details(run_id: str) -> LaunchDetails: + """Return workflow details (placeholder).""" + iso_now = datetime.now(timezone.utc).isoformat() + return LaunchDetails( + requiresAttention=False, + status="UNKNOWN", + ownerId=0, + repository="", + id=run_id, + submit="", + start="", + complete="", + dateCreated=iso_now, + lastUpdated=iso_now, + runName="", + sessionId="", + profile="", + workDir="", + commitId="", + userName="", + scriptId="", + revision="", + commandLine="", + projectName="", + scriptName="", + launchId="", + configFiles=[], + params={}, + ) diff --git a/app/routes/workflows.py b/app/routes/workflows.py index 4122659..f8b2e59 100644 --- a/app/routes/workflows.py +++ b/app/routes/workflows.py @@ -1,307 +1,14 @@ -"""Workflow-related HTTP routes.""" +"""Workflow router composed from smaller route modules.""" from __future__ import annotations -import asyncio -from datetime import datetime, timezone -from uuid import UUID +from fastapi import APIRouter -from fastapi import APIRouter, Depends, Header, HTTPException, Query, status -from sqlalchemy import select -from sqlalchemy.orm import Session - -from ..db import SessionLocal -from ..db.models.core import AppUser, WorkflowRun - -from ..schemas.workflows import ( - CancelWorkflowResponse, - DatasetUploadRequest, - DatasetUploadResponse, - JobListItem, - JobListResponse, - LaunchDetails, - LaunchLogs, - ListRunsResponse, - RunInfo, - WorkflowLaunchPayload, - WorkflowLaunchResponse, -) -from ..services.bindflow_executor import ( - BindflowConfigurationError, - BindflowExecutorError, - BindflowLaunchResult, - launch_bindflow_workflow, -) -from ..services.datasets import ( - create_seqera_dataset, - upload_dataset_to_seqera, -) -from ..services.seqera import ( - SeqeraAPIError, - SeqeraConfigurationError, - list_seqera_workflows, -) +from .workflow.jobs import router as jobs_router +from .workflow.launch import router as launch_router +from .workflow.placeholders import router as placeholders_router router = APIRouter(tags=["workflows"]) - - -def get_db() -> Session: - db = SessionLocal() - try: - yield db - finally: - db.close() - - -def get_current_user_id( - x_auth0_user_id: str | None = Header(None, alias="X-Auth0-User-Id"), - db: Session = Depends(get_db), -) -> UUID: - if not x_auth0_user_id: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Missing X-Auth0-User-Id header", - ) - - user = db.execute( - select(AppUser).where(AppUser.auth0_user_id == x_auth0_user_id) - ).scalar_one_or_none() - if not user: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Unknown user", - ) - return user.id - - -def get_owned_run_ids(db: Session, user_id: UUID) -> set[str]: - rows = db.execute( - select(WorkflowRun.seqera_run_id).where(WorkflowRun.owner_user_id == user_id) - ).all() - return {row[0] for row in rows if row[0]} - - -@router.post("/launch", response_model=WorkflowLaunchResponse, status_code=status.HTTP_201_CREATED) -async def launch_workflow(payload: WorkflowLaunchPayload) -> WorkflowLaunchResponse: - """Launch a workflow on the Seqera Platform.""" - try: - dataset_id = payload.datasetId - - # Use the dataset created from /datasets/upload endpoint - result: BindflowLaunchResult = await launch_bindflow_workflow(payload.launch, dataset_id) - except BindflowConfigurationError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) - ) from exc - except BindflowExecutorError as exc: - raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc - - return WorkflowLaunchResponse( - message="Workflow launched successfully", - runId=result.workflow_id, - status=result.status, - submitTime=datetime.now(timezone.utc), - ) - - -@router.post("/{run_id}/cancel", response_model=CancelWorkflowResponse) -async def cancel_workflow(run_id: str) -> CancelWorkflowResponse: - """Cancel a workflow run (placeholder implementation).""" - return CancelWorkflowResponse( - message="Workflow cancelled successfully", - runId=run_id, - status="cancelled", - ) - - -@router.get("/runs", response_model=ListRunsResponse) -async def list_runs( - status_filter: str | None = Query(None, alias="status"), - workspace: str | None = Query(None), - limit: int = Query(50, ge=1, le=200), - offset: int = Query(0, ge=0), - current_user_id: UUID = Depends(get_current_user_id), - db: Session = Depends(get_db), -) -> ListRunsResponse: - """List workflow runs with a compact response shape for the frontend.""" - parsed_status_filter = None - if status_filter: - parsed_status_filter = [item.strip() for item in status_filter.split(",") if item.strip()] - - try: - workflows, _total = await list_seqera_workflows( - workspace_id=workspace, - status_filter=parsed_status_filter, - limit=limit, - offset=offset, - ) - - owned_run_ids = get_owned_run_ids(db, current_user_id) - owned_workflows = [wf for wf in workflows if wf.workflow_id in owned_run_ids] - - runs = [ - RunInfo( - id=wf.workflow_id, - run=wf.run_name or wf.workflow_id, - workflow=wf.workflow_type or "", - status=wf.ui_status, - date=wf.submitted_at.isoformat() if wf.submitted_at else "", - cancel=f"/api/workflows/{wf.workflow_id}/cancel", - ) - for wf in owned_workflows - ] - return ListRunsResponse(runs=runs, total=len(runs), limit=limit, offset=offset) - except SeqeraConfigurationError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=str(exc), - ) from exc - except SeqeraAPIError as exc: - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=str(exc), - ) from exc - - -@router.get("/jobs", response_model=JobListResponse) -async def list_jobs( - search: str | None = Query(None, description="Search by job name or workflow type"), - status: list[str] | None = Query(None, description="Filter by status (Completed, Stopped, Failed)"), - limit: int = Query(50, ge=1, le=200, description="Maximum number of results"), - offset: int = Query(0, ge=0, description="Number of results to skip"), - current_user_id: UUID = Depends(get_current_user_id), - db: Session = Depends(get_db), -) -> JobListResponse: - """ - Retrieve a paginated list of workflow jobs with search and filtering. - - Query Parameters: - - search: Search by job name or workflow type - - status: Filter by status values (Completed, Stopped, Failed, In progress, In queue) - - limit: Maximum number of results per page - - offset: Number of results to skip for pagination - - Returns: - - Paginated list of jobs with job name, workflow type, status, submitted date, and score - """ - try: - workflows, _total = await list_seqera_workflows( - search_query=search, - status_filter=status, - limit=limit, - offset=offset, - ) - - owned_run_ids = get_owned_run_ids(db, current_user_id) - owned_workflows = [wf for wf in workflows if wf.workflow_id in owned_run_ids] - - jobs = [ - JobListItem( - id=wf.workflow_id, - jobName=wf.run_name or wf.workflow_id, - workflowType=wf.workflow_type, - status=wf.ui_status, - submittedAt=wf.submitted_at or datetime.now(timezone.utc), - score=wf.score, - ) - for wf in owned_workflows - ] - - return JobListResponse( - jobs=jobs, - total=len(jobs), - limit=limit, - offset=offset, - ) - except SeqeraConfigurationError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=str(exc), - ) from exc - except SeqeraAPIError as exc: - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=str(exc), - ) from exc - - -@router.get("/{run_id}/logs", response_model=LaunchLogs) -async def get_logs() -> LaunchLogs: - """Retrieve workflow logs (placeholder).""" - return LaunchLogs( - truncated=False, - entries=[], - rewindToken="", - forwardToken="", - pending=False, - message="Logs endpoint - implementation pending", - downloads=[], - ) - - -@router.get("/{run_id}/details", response_model=LaunchDetails) -async def get_details(run_id: str) -> LaunchDetails: - """Return workflow details (placeholder).""" - iso_now = datetime.now(timezone.utc).isoformat() - return LaunchDetails( - requiresAttention=False, - status="UNKNOWN", - ownerId=0, - repository="", - id=run_id, - submit="", - start="", - complete="", - dateCreated=iso_now, - lastUpdated=iso_now, - runName="", - sessionId="", - profile="", - workDir="", - commitId="", - userName="", - scriptId="", - revision="", - commandLine="", - projectName="", - scriptName="", - launchId="", - configFiles=[], - params={}, - ) - - -@router.post("/datasets/upload", response_model=DatasetUploadResponse) -async def upload_dataset(payload: DatasetUploadRequest) -> DatasetUploadResponse: - """Create a Seqera dataset and upload form data as CSV content.""" - try: - dataset = await create_seqera_dataset( - name=payload.datasetName, description=payload.datasetDescription - ) - except BindflowConfigurationError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) - ) from exc - except BindflowExecutorError as exc: - raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc - - # Allow Seqera time to finish dataset initialization before uploading - await asyncio.sleep(2) - - try: - upload_result = await upload_dataset_to_seqera(dataset.dataset_id, payload.formData) - except ValueError as exc: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - except BindflowConfigurationError as exc: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) - ) from exc - except BindflowExecutorError as exc: - raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc - - return DatasetUploadResponse( - message="Dataset created and uploaded successfully", - datasetId=upload_result.dataset_id, - success=upload_result.success, - details=upload_result.raw_response, - ) +router.include_router(launch_router) +router.include_router(jobs_router) +router.include_router(placeholders_router) diff --git a/app/schemas/workflows.py b/app/schemas/workflows.py index 696462e..94d0ed5 100644 --- a/app/schemas/workflows.py +++ b/app/schemas/workflows.py @@ -197,3 +197,36 @@ class JobListResponse(BaseModel): total: int = Field(..., description="Total number of jobs matching the criteria") limit: int = Field(..., description="Maximum number of items per page") offset: int = Field(..., description="Number of items skipped") + + +class JobDetailsResponse(BaseModel): + """Detailed response for a single job.""" + + id: str = Field(..., description="Workflow run ID") + jobName: str = Field(..., description="Human-readable job name") + workflowType: str | None = Field(None, description="Workflow type (e.g., BindCraft, De novo design)") + status: str = Field(..., description="UI-friendly status") + submittedAt: datetime = Field(..., description="Submission date and time") + score: float | None = Field(None, description="Job max score rounded to 3 decimals") + + +class DeleteJobResponse(BaseModel): + """Response for single job deletion.""" + + runId: str + deleted: bool + cancelledBeforeDelete: bool = False + message: str + + +class BulkDeleteJobsRequest(BaseModel): + """Request payload for bulk job deletion.""" + + runIds: list[str] = Field(..., min_length=1) + + +class BulkDeleteJobsResponse(BaseModel): + """Response for bulk job deletion.""" + + deleted: list[str] = Field(default_factory=list) + failed: dict[str, str] = Field(default_factory=dict) diff --git a/app/services/job_utils.py b/app/services/job_utils.py new file mode 100644 index 0000000..8d0a540 --- /dev/null +++ b/app/services/job_utils.py @@ -0,0 +1,99 @@ +"""Helpers for job ownership, score handling, and Seqera payload parsing.""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..db.models.core import RunMetric, Workflow, WorkflowRun +from .s3 import S3ConfigurationError, S3ServiceError, calculate_csv_column_max + + +def coerce_workflow_payload(payload: dict) -> dict: + if isinstance(payload, dict) and isinstance(payload.get("workflow"), dict): + return payload["workflow"] + return payload + + +def extract_pipeline_status(payload: dict) -> str: + workflow = coerce_workflow_payload(payload) + return str(workflow.get("status") or "UNKNOWN") + + +def parse_submit_datetime(payload: dict) -> datetime | None: + workflow = coerce_workflow_payload(payload) + submit_str = workflow.get("submit") or workflow.get("dateCreated") + if not submit_str: + return None + try: + return datetime.fromisoformat(str(submit_str).replace("Z", "+00:00")) + except ValueError: + return None + + +def get_owned_run_ids(db: Session, user_id: UUID) -> set[str]: + rows = db.execute( + select(WorkflowRun.seqera_run_id).where(WorkflowRun.owner_user_id == user_id) + ).all() + return {row[0] for row in rows if row[0]} + + +def get_owned_run(db: Session, user_id: UUID, run_id: str) -> WorkflowRun | None: + return db.execute( + select(WorkflowRun).where( + WorkflowRun.owner_user_id == user_id, + WorkflowRun.seqera_run_id == run_id, + ) + ).scalar_one_or_none() + + +def _round_score(value: float | Decimal | None) -> float | None: + if value is None: + return None + return round(float(value), 3) + + +def get_score_by_seqera_run_id(db: Session, user_id: UUID) -> dict[str, float]: + rows = db.execute( + select(WorkflowRun.seqera_run_id, RunMetric.max_score) + .outerjoin(RunMetric, RunMetric.run_id == WorkflowRun.id) + .where(WorkflowRun.owner_user_id == user_id) + ).all() + return {seqera_run_id: _round_score(score) for seqera_run_id, score in rows if score is not None} + + +def get_workflow_type_by_seqera_run_id(db: Session, user_id: UUID) -> dict[str, str]: + """Return workflow type labels from the local DB workflows table.""" + rows = db.execute( + select(WorkflowRun.seqera_run_id, Workflow.name) + .outerjoin(Workflow, Workflow.id == WorkflowRun.workflow_id) + .where(WorkflowRun.owner_user_id == user_id) + ).all() + return {seqera_run_id: workflow_name for seqera_run_id, workflow_name in rows if workflow_name} + + +async def ensure_completed_run_score(db: Session, run: WorkflowRun, ui_status: str) -> float | None: + if ui_status != "Completed": + return None + + existing = db.execute(select(RunMetric).where(RunMetric.run_id == run.id)).scalar_one_or_none() + if existing and existing.max_score is not None: + return _round_score(existing.max_score) + + file_key = f"results/{run.seqera_run_id}/ranker/s1_final_design_stats.csv" + try: + max_score = await calculate_csv_column_max(file_key=file_key, column_name="Average_i_pTM") + except (S3ConfigurationError, S3ServiceError, ValueError): + return None + + bounded_score = max(0.0, min(1.0, float(max_score))) + if existing: + existing.max_score = bounded_score + else: + db.add(RunMetric(run_id=run.id, max_score=bounded_score)) + db.commit() + return _round_score(bounded_score) diff --git a/app/services/seqera.py b/app/services/seqera.py index 8ffacaf..9c05269 100644 --- a/app/services/seqera.py +++ b/app/services/seqera.py @@ -2,53 +2,15 @@ from __future__ import annotations -import logging -import os -from dataclasses import dataclass -from datetime import datetime - -import httpx - -from ..schemas.workflows import map_pipeline_status_to_ui - -logger = logging.getLogger(__name__) - - -class SeqeraConfigurationError(RuntimeError): - """Raised when required Seqera configuration is missing.""" - - -class SeqeraAPIError(RuntimeError): - """Raised when Seqera API calls fail.""" - - -@dataclass -class WorkflowListItem: - """Individual workflow run item from Seqera API.""" - - workflow_id: str - run_name: str | None - workflow_type: str | None - pipeline_status: str - ui_status: str - submitted_at: datetime | None - score: float | None - - -def _masked_headers(headers: dict[str, str]) -> dict[str, str]: - """Mask sensitive headers before logging.""" - masked = dict(headers) - if "Authorization" in masked: - masked["Authorization"] = "Bearer ***" - return masked - - -def _get_required_env(key: str) -> str: - """Get required environment variable or raise error.""" - value = os.getenv(key) - if not value: - raise SeqeraConfigurationError(f"Missing required environment variable: {key}") - return value +from .seqera_client import ( + cancel_workflow_raw, + delete_workflow_raw, + describe_workflow_raw, + list_workflows_raw, +) +from .seqera_errors import SeqeraAPIError, SeqeraConfigurationError +from .seqera_models import WorkflowListItem +from .seqera_parsers import extract_workflow_type, parse_workflow_list_payload async def list_seqera_workflows( @@ -58,201 +20,31 @@ async def list_seqera_workflows( limit: int = 50, offset: int = 0, ) -> tuple[list[WorkflowListItem], int]: - """ - List workflow runs from Seqera Platform. - - Args: - workspace_id: Seqera workspace ID (uses env var if not provided) - search_query: Search term for job name or workflow type - status_filter: List of UI status values to filter by (e.g., ["Completed", "Failed"]) - limit: Maximum number of results to return - offset: Number of results to skip - - Returns: - Tuple of (list of workflow items, total count) - """ - seqera_api_url = _get_required_env("SEQERA_API_URL").rstrip("/") - seqera_token = _get_required_env("SEQERA_ACCESS_TOKEN") - - if not workspace_id: - workspace_id = os.getenv("WORK_SPACE") - - # Build query parameters. Keep workspace optional to match plain GET /workflow usage. - params: dict[str, str | int] = {} - if workspace_id: - params["workspaceId"] = workspace_id - - # Add search query if provided - if search_query: - params["search"] = search_query - - url = f"{seqera_api_url}/workflow" - - headers = { - "Authorization": f"Bearer {seqera_token}", - "Accept": "application/json", - } - - async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: - response = await client.get(url, headers=headers, params=params) - - - if response.is_error: - body = response.text - logger.error( - "Seqera API error", - extra={ - "status": response.status_code, - "reason": response.reason_phrase, - "body": body, - }, - ) - raise SeqeraAPIError( - f"Failed to list workflows: {response.status_code} {body}" - ) - - data = response.json() - - logger.info( - "Parsed workflow data from Seqera", - extra={ - "total_workflows": len(data.get("workflows", [])), - "total_size": data.get("totalSize", 0), - }, + """List workflow runs from Seqera Platform.""" + _ = (limit, offset) # API keeps these for backward compatibility. + data = await list_workflows_raw(workspace_id=workspace_id, search_query=search_query) + return parse_workflow_list_payload( + data, + status_filter=status_filter, + search_query=search_query, ) - - # Support both Seqera payload shapes: - # 1) {"workflows": [...], "totalSize": N} - # 2) [{"workflow": {...}}, ...] - if isinstance(data, dict): - workflows_data = data.get("workflows") or data.get("items") or [] - total_count = data.get("totalSize") or data.get("total") or len(workflows_data) - elif isinstance(data, list): - workflows_data = data - total_count = len(workflows_data) - else: - workflows_data = [] - total_count = 0 - - workflow_items: list[WorkflowListItem] = [] - - for item in workflows_data: - # Some Seqera responses wrap run data in {"workflow": {...}} - wf = item.get("workflow", item) if isinstance(item, dict) else {} - pipeline_status = wf.get("status", "UNKNOWN") - ui_status = map_pipeline_status_to_ui(pipeline_status) - - # Apply status filter if provided - if status_filter and ui_status not in status_filter: - continue - - # Parse submitted date - submitted_at = None - if submit_str := wf.get("submit") or wf.get("dateCreated"): - try: - submitted_at = datetime.fromisoformat(submit_str.replace("Z", "+00:00")) - except (ValueError, AttributeError): - logger.warning(f"Could not parse submit time: {submit_str}") - - workflow_items.append( - WorkflowListItem( - workflow_id=wf.get("id", ""), - run_name=wf.get("runName"), - workflow_type=_extract_workflow_type(wf), - pipeline_status=pipeline_status, - ui_status=ui_status, - submitted_at=submitted_at, - score=None - ) - ) - - logger.info( - f"Returning {len(workflow_items)} workflow items after filtering", - extra={"requested_status_filter": status_filter}, - ) - - return workflow_items, total_count +async def describe_workflow(workflow_id: str, workspace_id: str | None = None) -> dict: + """Get detailed information about a specific workflow run.""" + return await describe_workflow_raw(workflow_id=workflow_id, workspace_id=workspace_id) -def _extract_workflow_type(workflow_data: dict) -> str | None: - """ - Extract workflow type from workflow data. - - This could be based on: - - Pipeline name/repository - - Project name - - Custom metadata - """ - pipeline = workflow_data.get("projectName") or workflow_data.get("pipeline", "") - - # Map common pipeline names to workflow types - if "bindcraft" in pipeline.lower(): - return "BindCraft" - elif "denovo" in pipeline.lower() or "de-novo" in pipeline.lower(): - return "De novo design" - - return pipeline or None +async def cancel_seqera_workflow(workflow_id: str, workspace_id: str | None = None) -> None: + """Cancel a Seqera workflow run.""" + await cancel_workflow_raw(workflow_id=workflow_id, workspace_id=workspace_id) -async def describe_workflow(workflow_id: str, workspace_id: str | None = None) -> dict: - """ - Get detailed information about a specific workflow run. - - Args: - workflow_id: Seqera workflow run ID - workspace_id: Seqera workspace ID (uses env var if not provided) - - Returns: - Workflow details dictionary - """ - seqera_api_url = _get_required_env("SEQERA_API_URL").rstrip("/") - seqera_token = _get_required_env("SEQERA_ACCESS_TOKEN") - - if not workspace_id: - workspace_id = os.getenv("WORK_SPACE") - - url = f"{seqera_api_url}/workflow/{workflow_id}" - params: dict[str, str] = {} - if workspace_id: - params["workspaceId"] = workspace_id - - headers = { - "Authorization": f"Bearer {seqera_token}", - "Accept": "application/json", - } - - logger.info( - "Describing workflow from Seqera API: url=%s workflow_id=%s workspace_id=%s params=%s headers=%s", - url, - workflow_id, - workspace_id, - params, - _masked_headers(headers), - ) - - async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: - response = await client.get(url, headers=headers, params=params) - logger.info( - "Seqera describe API response: method=GET path=/workflow/%s status=%s raw_body=%s", - workflow_id, - response.status_code, - response.text if response.text else None, - ) - - if response.is_error: - body = response.text - logger.error( - "Seqera API error", - extra={ - "status": response.status_code, - "reason": response.reason_phrase, - "body": body, - }, - ) - raise SeqeraAPIError( - f"Failed to describe workflow: {response.status_code} {body}" - ) - - return response.json() +async def delete_seqera_workflow(workflow_id: str, workspace_id: str | None = None) -> None: + """Delete a Seqera workflow run.""" + await delete_workflow_raw(workflow_id=workflow_id, workspace_id=workspace_id) + + +def _extract_workflow_type(workflow_data: dict) -> str | None: + """Backward-compatible alias used by tests and callers.""" + return extract_workflow_type(workflow_data) diff --git a/app/services/seqera_client.py b/app/services/seqera_client.py new file mode 100644 index 0000000..aa15e46 --- /dev/null +++ b/app/services/seqera_client.py @@ -0,0 +1,89 @@ +"""Low-level HTTP calls to Seqera API.""" + +from __future__ import annotations + +import os + +import httpx + +from .seqera_errors import SeqeraAPIError, SeqeraConfigurationError + + +def _get_required_env(key: str) -> str: + value = os.getenv(key) + if not value: + raise SeqeraConfigurationError(f"Missing required environment variable: {key}") + return value + + +def _get_api_context(workspace_id: str | None = None) -> tuple[str, str, dict[str, str]]: + api_url = _get_required_env("SEQERA_API_URL").rstrip("/") + token = _get_required_env("SEQERA_ACCESS_TOKEN") + resolved_workspace = workspace_id or os.getenv("WORK_SPACE") + params: dict[str, str] = {} + if resolved_workspace: + params["workspaceId"] = resolved_workspace + return api_url, token, params + + +def _headers(token: str) -> dict[str, str]: + return { + "Authorization": f"Bearer {token}", + "Accept": "application/json", + } + + +async def list_workflows_raw( + workspace_id: str | None = None, + search_query: str | None = None, +) -> dict | list: + api_url, token, params = _get_api_context(workspace_id) + if search_query: + params["search"] = search_query + + url = f"{api_url}/workflow" + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + response = await client.get(url, headers=_headers(token), params=params) + + if response.is_error: + raise SeqeraAPIError(f"Failed to list workflows: {response.status_code} {response.text}") + return response.json() + + +async def describe_workflow_raw(workflow_id: str, workspace_id: str | None = None) -> dict: + api_url, token, params = _get_api_context(workspace_id) + url = f"{api_url}/workflow/{workflow_id}" + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + response = await client.get(url, headers=_headers(token), params=params) + + if response.is_error: + raise SeqeraAPIError(f"Failed to describe workflow: {response.status_code} {response.text}") + return response.json() + + +async def cancel_workflow_raw(workflow_id: str, workspace_id: str | None = None) -> None: + api_url, token, params = _get_api_context(workspace_id) + candidate_paths = [ + f"{api_url}/workflow/{workflow_id}/cancel", + f"{api_url}/workflow/{workflow_id}/kill", + ] + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + last_error = None + for url in candidate_paths: + response = await client.post(url, headers=_headers(token), params=params) + if not response.is_error: + return + last_error = f"{response.status_code} {response.text}" + raise SeqeraAPIError(f"Failed to cancel workflow {workflow_id}: {last_error}") + + +async def delete_workflow_raw(workflow_id: str, workspace_id: str | None = None) -> None: + api_url, token, params = _get_api_context(workspace_id) + url = f"{api_url}/workflow/{workflow_id}" + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + response = await client.delete(url, headers=_headers(token), params=params) + + if response.status_code == 404: + return + if response.is_error: + raise SeqeraAPIError(f"Failed to delete workflow {workflow_id}: {response.status_code} {response.text}") diff --git a/app/services/seqera_errors.py b/app/services/seqera_errors.py new file mode 100644 index 0000000..7db5b28 --- /dev/null +++ b/app/services/seqera_errors.py @@ -0,0 +1,9 @@ +"""Error types for Seqera service operations.""" + + +class SeqeraConfigurationError(RuntimeError): + """Raised when required Seqera configuration is missing.""" + + +class SeqeraAPIError(RuntimeError): + """Raised when Seqera API calls fail.""" diff --git a/app/services/seqera_models.py b/app/services/seqera_models.py new file mode 100644 index 0000000..68d563d --- /dev/null +++ b/app/services/seqera_models.py @@ -0,0 +1,19 @@ +"""Shared Seqera service models.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + + +@dataclass +class WorkflowListItem: + """Individual workflow run item from Seqera API.""" + + workflow_id: str + run_name: str | None + workflow_type: str | None + pipeline_status: str + ui_status: str + submitted_at: datetime | None + score: float | None diff --git a/app/services/seqera_parsers.py b/app/services/seqera_parsers.py new file mode 100644 index 0000000..354a4c2 --- /dev/null +++ b/app/services/seqera_parsers.py @@ -0,0 +1,74 @@ +"""Parsing helpers for Seqera workflow payloads.""" + +from __future__ import annotations + +from datetime import datetime + +from ..schemas.workflows import map_pipeline_status_to_ui +from .seqera_models import WorkflowListItem + + +def extract_workflow_type(workflow_data: dict) -> str | None: + """Extract workflow type from workflow data.""" + pipeline = workflow_data.get("projectName") or workflow_data.get("pipeline", "") + + if "bindcraft" in pipeline.lower(): + return "BindCraft" + if "denovo" in pipeline.lower() or "de-novo" in pipeline.lower(): + return "De novo design" + return pipeline or None + + +def parse_workflow_list_payload( + data: dict | list, + status_filter: list[str] | None = None, + search_query: str | None = None, +) -> tuple[list[WorkflowListItem], int]: + """Normalize Seqera list responses into workflow items.""" + if isinstance(data, dict): + workflows_data = data.get("workflows") or data.get("items") or [] + total_count = data.get("totalSize") or data.get("total") or len(workflows_data) + elif isinstance(data, list): + workflows_data = data + total_count = len(workflows_data) + else: + workflows_data = [] + total_count = 0 + + items: list[WorkflowListItem] = [] + for item in workflows_data: + wf = item.get("workflow", item) if isinstance(item, dict) else {} + pipeline_status = wf.get("status", "UNKNOWN") + ui_status = map_pipeline_status_to_ui(pipeline_status) + + if status_filter and ui_status not in status_filter: + continue + + submitted_at = None + if submit_str := wf.get("submit") or wf.get("dateCreated"): + try: + submitted_at = datetime.fromisoformat(submit_str.replace("Z", "+00:00")) + except (ValueError, AttributeError): + submitted_at = None + + items.append( + WorkflowListItem( + workflow_id=wf.get("id", ""), + run_name=wf.get("runName"), + workflow_type=extract_workflow_type(wf), + pipeline_status=pipeline_status, + ui_status=ui_status, + submitted_at=submitted_at, + score=None, + ) + ) + + if search_query: + needle = search_query.lower() + items = [ + wf + for wf in items + if needle in (wf.run_name or "").lower() or needle in (wf.workflow_type or "").lower() + ] + + return items, total_count From 6cb289a1566d23972a848d7d251d0fe60c44a977 Mon Sep 17 00:00:00 2001 From: laclac102 Date: Tue, 3 Feb 2026 17:02:59 +1100 Subject: [PATCH 4/6] fix: improve tests --- app/routes/workflow/jobs.py | 8 +- app/schemas/workflows.py | 4 +- app/services/seqera.py | 10 + tests/test_additional_coverage.py | 27 +- tests/test_main.py | 2 +- tests/test_routes_dependencies.py | 43 ++++ tests/test_routes_jobs.py | 300 +++++++---------------- tests/test_routes_workflow_jobs_extra.py | 93 +++++++ tests/test_routes_workflows.py | 67 ++--- tests/test_services_job_utils.py | 88 +++++++ tests/test_services_seqera_client.py | 81 ++++++ tests/test_services_workflow_listing.py | 2 - 12 files changed, 435 insertions(+), 290 deletions(-) create mode 100644 tests/test_routes_dependencies.py create mode 100644 tests/test_routes_workflow_jobs_extra.py create mode 100644 tests/test_services_job_utils.py create mode 100644 tests/test_services_seqera_client.py diff --git a/app/routes/workflow/jobs.py b/app/routes/workflow/jobs.py index 0ff2427..2a2a25d 100644 --- a/app/routes/workflow/jobs.py +++ b/app/routes/workflow/jobs.py @@ -68,7 +68,11 @@ async def cancel_workflow( @router.get("/jobs", response_model=JobListResponse) async def list_jobs( search: str | None = Query(None, description="Search by job name or workflow type"), - status: list[str] | None = Query(None, description="Filter by status (Completed, Stopped, Failed)"), + status_filter: list[str] | None = Query( + None, + alias="status", + description="Filter by status (Completed, Stopped, Failed)", + ), limit: int = Query(50, ge=1, le=200, description="Maximum number of results"), offset: int = Query(0, ge=0, description="Number of results to skip"), current_user_id: UUID = Depends(get_current_user_id), @@ -80,7 +84,7 @@ async def list_jobs( score_by_run_id = get_score_by_seqera_run_id(db, current_user_id) workflow_type_by_run_id = get_workflow_type_by_seqera_run_id(db, current_user_id) search_text = (search or "").strip().lower() - allowed_statuses = set(status or []) + allowed_statuses = set(status_filter or []) jobs: list[JobListItem] = [] for run_id in owned_run_ids: payload = await describe_workflow(run_id) diff --git a/app/schemas/workflows.py b/app/schemas/workflows.py index 94d0ed5..0681e4f 100644 --- a/app/schemas/workflows.py +++ b/app/schemas/workflows.py @@ -181,7 +181,7 @@ class PdbUploadResponse(BaseModel): class JobListItem(BaseModel): """Individual job item in the job listing.""" - + id: str = Field(..., description="Workflow run ID") jobName: str = Field(..., description="Human-readable job name") workflowType: str | None = Field(None, description="Workflow type (e.g., BindCraft, De novo design)") @@ -192,7 +192,7 @@ class JobListItem(BaseModel): class JobListResponse(BaseModel): """Paginated response for job listing.""" - + jobs: list[JobListItem] = Field(default_factory=list, description="List of jobs") total: int = Field(..., description="Total number of jobs matching the criteria") limit: int = Field(..., description="Maximum number of items per page") diff --git a/app/services/seqera.py b/app/services/seqera.py index 9c05269..ee61888 100644 --- a/app/services/seqera.py +++ b/app/services/seqera.py @@ -12,6 +12,16 @@ from .seqera_models import WorkflowListItem from .seqera_parsers import extract_workflow_type, parse_workflow_list_payload +__all__ = [ + "SeqeraAPIError", + "SeqeraConfigurationError", + "WorkflowListItem", + "list_seqera_workflows", + "describe_workflow", + "cancel_seqera_workflow", + "delete_seqera_workflow", +] + async def list_seqera_workflows( workspace_id: str | None = None, diff --git a/tests/test_additional_coverage.py b/tests/test_additional_coverage.py index 913f158..8a70918 100644 --- a/tests/test_additional_coverage.py +++ b/tests/test_additional_coverage.py @@ -7,14 +7,15 @@ import pytest from fastapi import HTTPException, status -from app.routes.workflows import get_details, upload_dataset +from app.routes.workflow.launch import upload_dataset +from app.routes.workflow.placeholders import get_details from app.schemas.workflows import DatasetUploadRequest from app.services.bindflow_executor import BindflowConfigurationError, BindflowExecutorError from app.services.datasets import DatasetUploadResult -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_success(mock_create, mock_upload): """Test successful dataset upload.""" # Mock dataset creation @@ -58,8 +59,8 @@ async def test_get_details_returns_placeholder(): assert isinstance(result.params, dict) -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_create_config_error(mock_create, mock_upload): """Test dataset upload handles BindflowConfigurationError during creation.""" # Mock dataset creation to raise error @@ -77,8 +78,8 @@ async def test_upload_dataset_create_config_error(mock_create, mock_upload): assert "Config error" in str(exc_info.value.detail) -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_create_service_error(mock_create, mock_upload): """Test dataset upload handles BindflowExecutorError during creation.""" mock_create.side_effect = BindflowExecutorError("Service error") @@ -95,8 +96,8 @@ async def test_upload_dataset_create_service_error(mock_create, mock_upload): assert "Service error" in str(exc_info.value.detail) -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_upload_value_error(mock_create, mock_upload): """Test dataset upload handles ValueError during upload.""" # Mock successful creation @@ -119,8 +120,8 @@ async def test_upload_dataset_upload_value_error(mock_create, mock_upload): assert "Invalid data" in str(exc_info.value.detail) -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_upload_config_error(mock_create, mock_upload): """Test dataset upload handles BindflowConfigurationError during upload.""" # Mock successful creation @@ -143,8 +144,8 @@ async def test_upload_dataset_upload_config_error(mock_create, mock_upload): assert "Upload config error" in str(exc_info.value.detail) -@patch("app.routes.workflows.upload_dataset_to_seqera") -@patch("app.routes.workflows.create_seqera_dataset") +@patch("app.routes.workflow.launch.upload_dataset_to_seqera") +@patch("app.routes.workflow.launch.create_seqera_dataset") async def test_upload_dataset_upload_service_error(mock_create, mock_upload): """Test dataset upload handles BindflowExecutorError during upload.""" # Mock successful creation diff --git a/tests/test_main.py b/tests/test_main.py index 0d6ae56..b761a2e 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -57,7 +57,7 @@ def test_workflow_router_included(app: FastAPI): route_paths = [route.path for route in app.routes] assert "/api/workflows/launch" in route_paths - assert "/api/workflows/runs" in route_paths + assert "/api/workflows/jobs" in route_paths def test_exception_handler(client: TestClient): diff --git a/tests/test_routes_dependencies.py b/tests/test_routes_dependencies.py new file mode 100644 index 0000000..473358f --- /dev/null +++ b/tests/test_routes_dependencies.py @@ -0,0 +1,43 @@ +"""Coverage tests for route dependencies.""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.routes.dependencies import get_current_user_id + + +class _Result: + def __init__(self, value): + self._value = value + + def scalar_one_or_none(self): + return self._value + + +class _DB: + def __init__(self, value): + self.value = value + + def execute(self, *_args, **_kwargs): + return _Result(self.value) + + +def test_get_current_user_id_missing_header(): + with pytest.raises(HTTPException) as exc: + get_current_user_id(None, _DB(None)) + assert exc.value.status_code == 401 + + +def test_get_current_user_id_unknown_user(): + with pytest.raises(HTTPException) as exc: + get_current_user_id("auth0|x", _DB(None)) + assert exc.value.status_code == 401 + + +def test_get_current_user_id_success(): + user = SimpleNamespace(id="u-1") + assert get_current_user_id("auth0|x", _DB(user)) == "u-1" diff --git a/tests/test_routes_jobs.py b/tests/test_routes_jobs.py index f60573a..38ad0f3 100644 --- a/tests/test_routes_jobs.py +++ b/tests/test_routes_jobs.py @@ -1,267 +1,133 @@ -"""Tests for job listing API endpoint.""" +"""Tests for job listing and details endpoints.""" from __future__ import annotations -from datetime import datetime, timezone from unittest.mock import AsyncMock, patch +from uuid import UUID import pytest from fastapi.testclient import TestClient from app.main import create_app -from app.services.seqera import WorkflowListItem +from app.routes.dependencies import get_current_user_id, get_db -@pytest.fixture -def client(): - """Create test client.""" - app = create_app() - return TestClient(app) - - -def test_list_jobs_endpoint_success(client, monkeypatch): - """Test successful job listing via API endpoint.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - mock_workflows = [ - WorkflowListItem( - workflow_id="wf-123", - run_name="Test Job 1", - workflow_type="BindCraft", - pipeline_status="SUCCEEDED", - ui_status="Completed", - submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), - score=0.95, - ), - WorkflowListItem( - workflow_id="wf-456", - run_name="Test Job 2", - workflow_type="De novo design", - pipeline_status="RUNNING", - ui_status="In progress", - submitted_at=datetime(2026, 2, 2, 11, 0, 0, tzinfo=timezone.utc), - score=None, - ), - ] +class _DummyResult: + def all(self): + return [] - with patch( - "app.routes.workflows.list_seqera_workflows", - new_callable=AsyncMock, - return_value=(mock_workflows, 2), - ): - response = client.get("/jobs?limit=10&offset=0") - - assert response.status_code == 200 - data = response.json() + def scalar_one_or_none(self): + return None - assert data["total"] == 2 - assert data["limit"] == 10 - assert data["offset"] == 0 - assert len(data["jobs"]) == 2 - job1 = data["jobs"][0] - assert job1["id"] == "wf-123" - assert job1["jobName"] == "Test Job 1" - assert job1["workflowType"] == "BindCraft" - assert job1["status"] == "Completed" - assert job1["score"] == 0.95 - - job2 = data["jobs"][1] - assert job2["id"] == "wf-456" - assert job2["jobName"] == "Test Job 2" - assert job2["workflowType"] == "De novo design" - assert job2["status"] == "In progress" - assert job2["score"] is None - - -def test_list_jobs_with_search_query(client, monkeypatch): - """Test job listing with search query.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - mock_workflows = [ - WorkflowListItem( - workflow_id="wf-789", - run_name="Matching Job", - workflow_type="BindCraft", - pipeline_status="SUCCEEDED", - ui_status="Completed", - submitted_at=datetime(2026, 2, 3, 12, 0, 0, tzinfo=timezone.utc), - score=None, - ) - ] +class _DummyDB: + def execute(self, *_args, **_kwargs): + return _DummyResult() - with patch( - "app.routes.workflows.list_seqera_workflows", - new_callable=AsyncMock, - return_value=(mock_workflows, 1), - ) as mock_list: - response = client.get("/jobs?search=Matching&limit=10&offset=0") + def commit(self): + return None - assert response.status_code == 200 - data = response.json() + def rollback(self): + return None - # Verify the service was called with search parameter - mock_list.assert_called_once() - call_kwargs = mock_list.call_args.kwargs - assert call_kwargs["search_query"] == "Matching" + def delete(self, *_args, **_kwargs): + return None - assert data["total"] == 1 - assert len(data["jobs"]) == 1 - assert data["jobs"][0]["jobName"] == "Matching Job" +@pytest.fixture +def client(): + app = create_app() -def test_list_jobs_with_status_filter(client, monkeypatch): - """Test job listing with status filter.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - mock_workflows = [ - WorkflowListItem( - workflow_id="wf-100", - run_name="Completed Job", - workflow_type="BindCraft", - pipeline_status="SUCCEEDED", - ui_status="Completed", - submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), - score=None, - ) - ] + def override_get_current_user_id() -> UUID: + return UUID("11111111-1111-1111-1111-111111111111") - with patch( - "app.routes.workflows.list_seqera_workflows", - new_callable=AsyncMock, - return_value=(mock_workflows, 1), - ) as mock_list: - response = client.get("/jobs?status=Completed&status=Failed&limit=10&offset=0") + def override_get_db(): + yield _DummyDB() - assert response.status_code == 200 - data = response.json() - - # Verify the service was called with status filter - mock_list.assert_called_once() - call_kwargs = mock_list.call_args.kwargs - assert call_kwargs["status_filter"] == ["Completed", "Failed"] + app.dependency_overrides[get_current_user_id] = override_get_current_user_id + app.dependency_overrides[get_db] = override_get_db + return TestClient(app) - assert data["total"] == 1 - assert len(data["jobs"]) == 1 - assert data["jobs"][0]["status"] == "Completed" - - -def test_list_jobs_with_pagination(client, monkeypatch): - """Test job listing with pagination parameters.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - mock_workflows = [ - WorkflowListItem( - workflow_id=f"wf-{i}", - run_name=f"Job {i}", - workflow_type="BindCraft", - pipeline_status="SUCCEEDED", - ui_status="Completed", - submitted_at=datetime(2026, 2, 1, 10, 0, 0, tzinfo=timezone.utc), - score=None, - ) - for i in range(20, 40) - ] - with patch( - "app.routes.workflows.list_seqera_workflows", - new_callable=AsyncMock, - return_value=(mock_workflows, 100), - ) as mock_list: - response = client.get("/jobs?limit=20&offset=20") +def test_list_jobs_endpoint_success(client): + with ( + patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-123", "wf-456"}), + patch("app.routes.workflow.jobs.get_score_by_seqera_run_id", return_value={"wf-123": 0.953}), + patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), + patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=0.953), + patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + side_effect=[ + {"workflow": {"id": "wf-123", "runName": "Job A", "projectName": "BindCraft", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}, + {"workflow": {"id": "wf-456", "runName": "Job B", "projectName": "De novo design", "status": "RUNNING", "submit": "2026-02-02T10:00:00Z"}}, + ], + ), + ): + response = client.get("/api/workflows/jobs?limit=10&offset=0") assert response.status_code == 200 data = response.json() - - # Verify the service was called with correct pagination - mock_list.assert_called_once() - call_kwargs = mock_list.call_args.kwargs - assert call_kwargs["limit"] == 20 - assert call_kwargs["offset"] == 20 - - assert data["total"] == 100 - assert data["limit"] == 20 - assert data["offset"] == 20 - assert len(data["jobs"]) == 20 - - -def test_list_jobs_empty_results(client, monkeypatch): - """Test job listing with no results.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - with patch( - "app.routes.workflows.list_seqera_workflows", - new_callable=AsyncMock, - return_value=([], 0), + assert data["total"] == 2 + assert len(data["jobs"]) == 2 + assert data["jobs"][1]["score"] == 0.953 + + +def test_list_jobs_with_search_and_status_filter(client): + with ( + patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-1", "wf-2"}), + patch("app.routes.workflow.jobs.get_score_by_seqera_run_id", return_value={}), + patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), + patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=None), + patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + side_effect=[ + {"workflow": {"id": "wf-1", "runName": "Matching Job", "projectName": "BindCraft", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}, + {"workflow": {"id": "wf-2", "runName": "Other Job", "projectName": "BindCraft", "status": "FAILED", "submit": "2026-02-01T11:00:00Z"}}, + ], + ), ): - response = client.get("/jobs?limit=10&offset=0") + response = client.get("/api/workflows/jobs?search=Matching&status=Completed&limit=10&offset=0") assert response.status_code == 200 data = response.json() + assert data["total"] == 1 + assert data["jobs"][0]["jobName"] == "Matching Job" - assert data["total"] == 0 - assert data["limit"] == 10 - assert data["offset"] == 0 - assert len(data["jobs"]) == 0 - - -def test_list_jobs_invalid_limit(client, monkeypatch): - """Test job listing with invalid limit parameter.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") - - response = client.get("/jobs?limit=0") - assert response.status_code == 422 # Validation error - response = client.get("/jobs?limit=500") - assert response.status_code == 422 # Exceeds max limit +def test_list_jobs_invalid_limit(client): + response = client.get("/api/workflows/jobs?limit=0") + assert response.status_code == 422 -def test_list_jobs_invalid_offset(client, monkeypatch): - """Test job listing with invalid offset parameter.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") +def test_list_jobs_invalid_offset(client): + response = client.get("/api/workflows/jobs?offset=-1") + assert response.status_code == 422 - response = client.get("/jobs?offset=-1") - assert response.status_code == 422 # Validation error +def test_list_jobs_configuration_error(client): + from app.services.seqera import SeqeraConfigurationError -def test_list_jobs_configuration_error(client, monkeypatch): - """Test job listing with missing configuration.""" - # Don't set required environment variables - monkeypatch.delenv("SEQERA_API_URL", raising=False) + with patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + side_effect=SeqeraConfigurationError("Missing required environment variable: SEQERA_API_URL"), + ), patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-123"}): + response = client.get("/api/workflows/jobs?limit=10&offset=0") - response = client.get("/jobs?limit=10&offset=0") assert response.status_code == 500 - assert "SEQERA_API_URL" in response.json()["detail"] - -def test_list_jobs_api_error(client, monkeypatch): - """Test job listing when Seqera API fails.""" - monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") - monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "test-token") - monkeypatch.setenv("WORK_SPACE", "test-workspace") +def test_list_jobs_api_error(client): from app.services.seqera import SeqeraAPIError with patch( - "app.routes.workflows.list_seqera_workflows", + "app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, side_effect=SeqeraAPIError("Seqera API is down"), - ): - response = client.get("/jobs?limit=10&offset=0") + ), patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-123"}): + response = client.get("/api/workflows/jobs?limit=10&offset=0") assert response.status_code == 502 assert "Seqera API is down" in response.json()["detail"] diff --git a/tests/test_routes_workflow_jobs_extra.py b/tests/test_routes_workflow_jobs_extra.py new file mode 100644 index 0000000..e7d82ee --- /dev/null +++ b/tests/test_routes_workflow_jobs_extra.py @@ -0,0 +1,93 @@ +"""Extra coverage for workflow jobs route handlers.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch +from uuid import UUID + +import pytest +from fastapi import HTTPException + +from app.routes.workflow.jobs import ( + bulk_delete_jobs, + cancel_workflow, + delete_job, + get_job_details, +) +from app.schemas.workflows import BulkDeleteJobsRequest +from app.services.seqera import SeqeraAPIError + + +@pytest.mark.asyncio +async def test_cancel_workflow_not_owned_raises_404(): + with patch("app.routes.workflow.jobs.get_owned_run", return_value=None): + with pytest.raises(HTTPException) as exc: + await cancel_workflow("wf-1", UUID("11111111-1111-1111-1111-111111111111"), Mock()) + assert exc.value.status_code == 404 + + +@pytest.mark.asyncio +async def test_cancel_workflow_api_error_maps_502(): + with ( + patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), + patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, side_effect=SeqeraAPIError("down")), + ): + with pytest.raises(HTTPException) as exc: + await cancel_workflow("wf-1", UUID("11111111-1111-1111-1111-111111111111"), Mock()) + assert exc.value.status_code == 502 + + +@pytest.mark.asyncio +async def test_get_job_details_success(): + owned_run = SimpleNamespace(workflow=SimpleNamespace(name="BindCraft"), id="rid") + with ( + patch("app.routes.workflow.jobs.get_owned_run", return_value=owned_run), + patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"runName": "job-x", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}), + patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=0.912), + ): + result = await get_job_details("wf-1", UUID("11111111-1111-1111-1111-111111111111"), Mock()) + + assert result.id == "wf-1" + assert result.workflowType == "BindCraft" + assert result.score == 0.912 + + +@pytest.mark.asyncio +async def test_delete_job_success_cancels_running_and_deletes_local_rows(): + db = Mock() + owned_run = SimpleNamespace(id="rid", workflow=None) + with ( + patch("app.routes.workflow.jobs.get_owned_run", return_value=owned_run), + patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"status": "RUNNING"}}), + patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, return_value=None), + patch("app.routes.workflow.jobs.delete_seqera_workflow", new_callable=AsyncMock, return_value=None), + ): + resp = await delete_job("wf-1", UUID("11111111-1111-1111-1111-111111111111"), db) + + assert resp.deleted is True + assert resp.cancelledBeforeDelete is True + assert db.execute.call_count == 3 + db.commit.assert_called_once() + + +@pytest.mark.asyncio +async def test_bulk_delete_jobs_mixed_results(): + db = Mock() + + def _owned(_db, _uid, run_id): + return None if run_id == "missing" else SimpleNamespace(id=f"id-{run_id}") + + with ( + patch("app.routes.workflow.jobs.get_owned_run", side_effect=_owned), + patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"status": "FAILED"}}), + patch("app.routes.workflow.jobs.delete_seqera_workflow", new_callable=AsyncMock, return_value=None), + ): + out = await bulk_delete_jobs( + BulkDeleteJobsRequest(runIds=["ok", "missing"]), + UUID("11111111-1111-1111-1111-111111111111"), + db, + ) + + assert out.deleted == ["ok"] + assert out.failed["missing"] == "Job not found" diff --git a/tests/test_routes_workflows.py b/tests/test_routes_workflows.py index 615470a..a4b7871 100644 --- a/tests/test_routes_workflows.py +++ b/tests/test_routes_workflows.py @@ -2,10 +2,12 @@ from __future__ import annotations -from unittest.mock import patch +from unittest.mock import AsyncMock, patch +from uuid import UUID from fastapi.testclient import TestClient +from app.routes.dependencies import get_current_user_id, get_db from app.services.bindflow_executor import ( BindflowConfigurationError, BindflowExecutorError, @@ -13,7 +15,7 @@ ) -@patch("app.routes.workflows.launch_bindflow_workflow") +@patch("app.routes.workflow.launch.launch_bindflow_workflow") async def test_launch_success_without_dataset(mock_launch, client: TestClient): """Test successful workflow launch without dataset.""" mock_launch.return_value = BindflowLaunchResult( @@ -38,7 +40,7 @@ async def test_launch_success_without_dataset(mock_launch, client: TestClient): assert "submitTime" in data -@patch("app.routes.workflows.launch_bindflow_workflow") +@patch("app.routes.workflow.launch.launch_bindflow_workflow") async def test_launch_success_with_dataset_id(mock_launch, client: TestClient): """Test successful workflow launch with pre-created dataset ID.""" # Mock workflow launch @@ -67,7 +69,7 @@ async def test_launch_success_with_dataset_id(mock_launch, client: TestClient): assert call_args[0][1] == "dataset_456" # Second argument is dataset_id -@patch("app.routes.workflows.launch_bindflow_workflow") +@patch("app.routes.workflow.launch.launch_bindflow_workflow") async def test_launch_configuration_error(mock_launch, client: TestClient): """Test launch with configuration error.""" mock_launch.side_effect = BindflowConfigurationError("Missing API token") @@ -84,7 +86,7 @@ async def test_launch_configuration_error(mock_launch, client: TestClient): assert "Missing API token" in response.json()["detail"] -@patch("app.routes.workflows.launch_bindflow_workflow") +@patch("app.routes.workflow.launch.launch_bindflow_workflow") async def test_launch_service_error(mock_launch, client: TestClient): """Test launch with Seqera service error.""" mock_launch.side_effect = BindflowExecutorError("API returned 502") @@ -116,7 +118,13 @@ def test_launch_invalid_payload(client: TestClient): def test_cancel_workflow_success(client: TestClient): """Test successful workflow cancellation.""" - response = client.post("/api/workflows/run_123/cancel") + client.app.dependency_overrides[get_current_user_id] = lambda: UUID("11111111-1111-1111-1111-111111111111") + client.app.dependency_overrides[get_db] = lambda: iter([None]) + with ( + patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), + patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, return_value=None), + ): + response = client.post("/api/workflows/run_123/cancel") assert response.status_code == 200 data = response.json() @@ -125,53 +133,6 @@ def test_cancel_workflow_success(client: TestClient): assert "message" in data -def test_list_runs_default_params(client: TestClient): - """Test listing runs with default parameters.""" - response = client.get("/api/workflows/runs") - - assert response.status_code == 200 - data = response.json() - assert "runs" in data - assert data["limit"] == 50 - assert data["offset"] == 0 - assert data["total"] == 0 - - -def test_list_runs_with_filters(client: TestClient): - """Test listing runs with filter parameters.""" - response = client.get( - "/api/workflows/runs", - params={ - "status": "running", - "workspace": "test_ws", - "limit": 10, - "offset": 5, - }, - ) - - assert response.status_code == 200 - data = response.json() - assert data["limit"] == 10 - assert data["offset"] == 5 - - -def test_list_runs_limit_validation(client: TestClient): - """Test that limit must be between 1 and 200.""" - # Test limit too high - response = client.get("/api/workflows/runs", params={"limit": 300}) - assert response.status_code == 422 - - # Test limit too low - response = client.get("/api/workflows/runs", params={"limit": 0}) - assert response.status_code == 422 - - -def test_list_runs_offset_validation(client: TestClient): - """Test that offset must be non-negative.""" - response = client.get("/api/workflows/runs", params={"offset": -1}) - assert response.status_code == 422 - - def test_get_logs_success(client: TestClient): """Test successful log retrieval.""" response = client.get("/api/workflows/run_123/logs") diff --git a/tests/test_services_job_utils.py b/tests/test_services_job_utils.py new file mode 100644 index 0000000..68abfdc --- /dev/null +++ b/tests/test_services_job_utils.py @@ -0,0 +1,88 @@ +"""Coverage tests for job utility helpers.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch +from uuid import UUID + +import pytest + +from app.services import job_utils + + +class _Result: + def __init__(self, all_value=None, scalar_value=None): + self._all = all_value or [] + self._scalar = scalar_value + + def all(self): + return self._all + + def scalar_one_or_none(self): + return self._scalar + + +class _DB: + def __init__(self, all_rows=None, scalar=None): + self._all_rows = all_rows or [] + self._scalar = scalar + self.added = None + self.committed = False + + def execute(self, *_args, **_kwargs): + return _Result(all_value=self._all_rows, scalar_value=self._scalar) + + def add(self, obj): + self.added = obj + + def commit(self): + self.committed = True + + +def test_coerce_and_extract_helpers(): + payload = {"workflow": {"status": "RUNNING"}} + assert job_utils.coerce_workflow_payload(payload) == payload["workflow"] + assert job_utils.extract_pipeline_status(payload) == "RUNNING" + + +def test_parse_submit_datetime_invalid_returns_none(): + assert job_utils.parse_submit_datetime({"workflow": {"submit": "bad"}}) is None + + +def test_get_owned_run_ids_and_scores_and_workflow_type(): + uid = UUID("11111111-1111-1111-1111-111111111111") + + db_ids = _DB(all_rows=[("wf-1",), ("wf-2",)]) + assert job_utils.get_owned_run_ids(db_ids, uid) == {"wf-1", "wf-2"} + + db_scores = _DB(all_rows=[("wf-1", 0.9123), ("wf-2", None)]) + assert job_utils.get_score_by_seqera_run_id(db_scores, uid) == {"wf-1": 0.912} + + db_types = _DB(all_rows=[("wf-1", "BindCraft")]) + assert job_utils.get_workflow_type_by_seqera_run_id(db_types, uid) == {"wf-1": "BindCraft"} + + +@pytest.mark.asyncio +async def test_ensure_completed_run_score_branches(): + run = SimpleNamespace(id="rid", seqera_run_id="wf-1") + + # non-completed status + assert await job_utils.ensure_completed_run_score(_DB(), run, "Failed") is None + + # existing score path + db_existing = _DB(scalar=SimpleNamespace(max_score=0.9)) + assert await job_utils.ensure_completed_run_score(db_existing, run, "Completed") == 0.9 + + # calculate + add path + db_new = _DB(scalar=None) + with patch("app.services.job_utils.calculate_csv_column_max", new_callable=AsyncMock, return_value=1.23): + score = await job_utils.ensure_completed_run_score(db_new, run, "Completed") + assert score == 1.0 + assert db_new.added is not None + assert db_new.committed is True + + # calculate failure path + db_fail = _DB(scalar=None) + with patch("app.services.job_utils.calculate_csv_column_max", new_callable=AsyncMock, side_effect=ValueError("bad")): + assert await job_utils.ensure_completed_run_score(db_fail, run, "Completed") is None diff --git a/tests/test_services_seqera_client.py b/tests/test_services_seqera_client.py new file mode 100644 index 0000000..d53a4a3 --- /dev/null +++ b/tests/test_services_seqera_client.py @@ -0,0 +1,81 @@ +"""Coverage tests for low-level Seqera client helpers.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from app.services.seqera_client import ( + cancel_workflow_raw, + delete_workflow_raw, + describe_workflow_raw, + list_workflows_raw, +) +from app.services.seqera_errors import SeqeraAPIError, SeqeraConfigurationError + + +@pytest.mark.asyncio +async def test_list_workflows_raw_missing_config(monkeypatch): + monkeypatch.delenv("SEQERA_API_URL", raising=False) + with pytest.raises(SeqeraConfigurationError): + await list_workflows_raw() + + +@pytest.mark.asyncio +async def test_describe_and_list_success(monkeypatch): + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "token") + + ok = AsyncMock(spec=httpx.Response) + ok.is_error = False + ok.json.return_value = {"ok": True} + + with patch("httpx.AsyncClient.get", return_value=ok): + assert await list_workflows_raw() == {"ok": True} + assert await describe_workflow_raw("wf-1") == {"ok": True} + + +@pytest.mark.asyncio +async def test_cancel_and_delete_paths(monkeypatch): + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "token") + + err = AsyncMock(spec=httpx.Response) + err.is_error = True + err.status_code = 500 + err.text = "no" + + ok = AsyncMock(spec=httpx.Response) + ok.is_error = False + + with patch("httpx.AsyncClient.post", side_effect=[err, ok]): + await cancel_workflow_raw("wf-1") + + not_found = AsyncMock(spec=httpx.Response) + not_found.status_code = 404 + not_found.is_error = True + not_found.text = "missing" + + with patch("httpx.AsyncClient.delete", return_value=not_found): + await delete_workflow_raw("wf-1") + + +@pytest.mark.asyncio +async def test_cancel_and_delete_errors(monkeypatch): + monkeypatch.setenv("SEQERA_API_URL", "https://api.seqera.test") + monkeypatch.setenv("SEQERA_ACCESS_TOKEN", "token") + + err = AsyncMock(spec=httpx.Response) + err.is_error = True + err.status_code = 500 + err.text = "err" + + with patch("httpx.AsyncClient.post", return_value=err): + with pytest.raises(SeqeraAPIError): + await cancel_workflow_raw("wf-1") + + with patch("httpx.AsyncClient.delete", return_value=err): + with pytest.raises(SeqeraAPIError): + await delete_workflow_raw("wf-1") diff --git a/tests/test_services_workflow_listing.py b/tests/test_services_workflow_listing.py index cc8bbb7..276e9da 100644 --- a/tests/test_services_workflow_listing.py +++ b/tests/test_services_workflow_listing.py @@ -2,7 +2,6 @@ from __future__ import annotations -from datetime import datetime, timezone from unittest.mock import AsyncMock, patch import httpx @@ -11,7 +10,6 @@ from app.services.seqera import ( SeqeraAPIError, SeqeraConfigurationError, - WorkflowListItem, describe_workflow, list_seqera_workflows, ) From 8baca748560e10a00abb3df26cb71a5f55126994 Mon Sep 17 00:00:00 2001 From: laclac102 Date: Wed, 4 Feb 2026 09:56:21 +1100 Subject: [PATCH 5/6] fix: black formatting --- app/routes/workflow/jobs.py | 17 +++++-- app/schemas/workflows.py | 10 +++- app/services/job_utils.py | 4 +- app/services/seqera_client.py | 4 +- tests/test_routes_jobs.py | 64 ++++++++++++++++++++---- tests/test_routes_workflow_jobs_extra.py | 54 +++++++++++++++++--- tests/test_routes_workflows.py | 10 +++- tests/test_services_job_utils.py | 10 +++- 8 files changed, 144 insertions(+), 29 deletions(-) diff --git a/app/routes/workflow/jobs.py b/app/routes/workflow/jobs.py index 2a2a25d..c98c701 100644 --- a/app/routes/workflow/jobs.py +++ b/app/routes/workflow/jobs.py @@ -68,7 +68,8 @@ async def cancel_workflow( @router.get("/jobs", response_model=JobListResponse) async def list_jobs( search: str | None = Query(None, description="Search by job name or workflow type"), - status_filter: list[str] | None = Query( + status_filter: list[str] + | None = Query( None, alias="status", description="Filter by status (Completed, Stopped, Failed)", @@ -97,7 +98,11 @@ async def list_jobs( workflow_type = workflow_type_by_run_id.get(run_id) job_name = wf.get("runName") or run_id - if search_text and search_text not in str(job_name).lower() and search_text not in str(workflow_type or "").lower(): + if ( + search_text + and search_text not in str(job_name).lower() + and search_text not in str(workflow_type or "").lower() + ): continue score = score_by_run_id.get(run_id) @@ -152,7 +157,9 @@ async def get_job_details( try: payload = await describe_workflow(run_id) except SeqeraConfigurationError as exc: - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc except SeqeraAPIError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @@ -196,7 +203,9 @@ async def delete_job( await delete_seqera_workflow(run_id) except SeqeraConfigurationError as exc: - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc except SeqeraAPIError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc diff --git a/app/schemas/workflows.py b/app/schemas/workflows.py index 0681e4f..f29f8b3 100644 --- a/app/schemas/workflows.py +++ b/app/schemas/workflows.py @@ -11,6 +11,7 @@ class PipelineStatus(str, Enum): """Pipeline status values from Seqera Platform.""" + SUBMITTED = "SUBMITTED" RUNNING = "RUNNING" SUCCEEDED = "SUCCEEDED" @@ -21,6 +22,7 @@ class PipelineStatus(str, Enum): class UIStatus(str, Enum): """User-facing status values for the frontend.""" + IN_QUEUE = "In queue" IN_PROGRESS = "In progress" COMPLETED = "Completed" @@ -184,7 +186,9 @@ class JobListItem(BaseModel): id: str = Field(..., description="Workflow run ID") jobName: str = Field(..., description="Human-readable job name") - workflowType: str | None = Field(None, description="Workflow type (e.g., BindCraft, De novo design)") + workflowType: str | None = Field( + None, description="Workflow type (e.g., BindCraft, De novo design)" + ) status: str = Field(..., description="UI-friendly status (e.g., Completed, In progress)") submittedAt: datetime = Field(..., description="Submission date and time") score: float | None = Field(None, description="Job score/metric") @@ -204,7 +208,9 @@ class JobDetailsResponse(BaseModel): id: str = Field(..., description="Workflow run ID") jobName: str = Field(..., description="Human-readable job name") - workflowType: str | None = Field(None, description="Workflow type (e.g., BindCraft, De novo design)") + workflowType: str | None = Field( + None, description="Workflow type (e.g., BindCraft, De novo design)" + ) status: str = Field(..., description="UI-friendly status") submittedAt: datetime = Field(..., description="Submission date and time") score: float | None = Field(None, description="Job max score rounded to 3 decimals") diff --git a/app/services/job_utils.py b/app/services/job_utils.py index 8d0a540..0b37e7a 100644 --- a/app/services/job_utils.py +++ b/app/services/job_utils.py @@ -63,7 +63,9 @@ def get_score_by_seqera_run_id(db: Session, user_id: UUID) -> dict[str, float]: .outerjoin(RunMetric, RunMetric.run_id == WorkflowRun.id) .where(WorkflowRun.owner_user_id == user_id) ).all() - return {seqera_run_id: _round_score(score) for seqera_run_id, score in rows if score is not None} + return { + seqera_run_id: _round_score(score) for seqera_run_id, score in rows if score is not None + } def get_workflow_type_by_seqera_run_id(db: Session, user_id: UUID) -> dict[str, str]: diff --git a/app/services/seqera_client.py b/app/services/seqera_client.py index aa15e46..afa7344 100644 --- a/app/services/seqera_client.py +++ b/app/services/seqera_client.py @@ -86,4 +86,6 @@ async def delete_workflow_raw(workflow_id: str, workspace_id: str | None = None) if response.status_code == 404: return if response.is_error: - raise SeqeraAPIError(f"Failed to delete workflow {workflow_id}: {response.status_code} {response.text}") + raise SeqeraAPIError( + f"Failed to delete workflow {workflow_id}: {response.status_code} {response.text}" + ) diff --git a/tests/test_routes_jobs.py b/tests/test_routes_jobs.py index 38ad0f3..86a549d 100644 --- a/tests/test_routes_jobs.py +++ b/tests/test_routes_jobs.py @@ -52,15 +52,37 @@ def override_get_db(): def test_list_jobs_endpoint_success(client): with ( patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-123", "wf-456"}), - patch("app.routes.workflow.jobs.get_score_by_seqera_run_id", return_value={"wf-123": 0.953}), + patch( + "app.routes.workflow.jobs.get_score_by_seqera_run_id", return_value={"wf-123": 0.953} + ), patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), - patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=0.953), + patch( + "app.routes.workflow.jobs.ensure_completed_run_score", + new_callable=AsyncMock, + return_value=0.953, + ), patch( "app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, side_effect=[ - {"workflow": {"id": "wf-123", "runName": "Job A", "projectName": "BindCraft", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}, - {"workflow": {"id": "wf-456", "runName": "Job B", "projectName": "De novo design", "status": "RUNNING", "submit": "2026-02-02T10:00:00Z"}}, + { + "workflow": { + "id": "wf-123", + "runName": "Job A", + "projectName": "BindCraft", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + } + }, + { + "workflow": { + "id": "wf-456", + "runName": "Job B", + "projectName": "De novo design", + "status": "RUNNING", + "submit": "2026-02-02T10:00:00Z", + } + }, ], ), ): @@ -78,17 +100,39 @@ def test_list_jobs_with_search_and_status_filter(client): patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-1", "wf-2"}), patch("app.routes.workflow.jobs.get_score_by_seqera_run_id", return_value={}), patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), - patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=None), + patch( + "app.routes.workflow.jobs.ensure_completed_run_score", + new_callable=AsyncMock, + return_value=None, + ), patch( "app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, side_effect=[ - {"workflow": {"id": "wf-1", "runName": "Matching Job", "projectName": "BindCraft", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}, - {"workflow": {"id": "wf-2", "runName": "Other Job", "projectName": "BindCraft", "status": "FAILED", "submit": "2026-02-01T11:00:00Z"}}, + { + "workflow": { + "id": "wf-1", + "runName": "Matching Job", + "projectName": "BindCraft", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + } + }, + { + "workflow": { + "id": "wf-2", + "runName": "Other Job", + "projectName": "BindCraft", + "status": "FAILED", + "submit": "2026-02-01T11:00:00Z", + } + }, ], ), ): - response = client.get("/api/workflows/jobs?search=Matching&status=Completed&limit=10&offset=0") + response = client.get( + "/api/workflows/jobs?search=Matching&status=Completed&limit=10&offset=0" + ) assert response.status_code == 200 data = response.json() @@ -112,7 +156,9 @@ def test_list_jobs_configuration_error(client): with patch( "app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, - side_effect=SeqeraConfigurationError("Missing required environment variable: SEQERA_API_URL"), + side_effect=SeqeraConfigurationError( + "Missing required environment variable: SEQERA_API_URL" + ), ), patch("app.routes.workflow.jobs.get_owned_run_ids", return_value={"wf-123"}): response = client.get("/api/workflows/jobs?limit=10&offset=0") diff --git a/tests/test_routes_workflow_jobs_extra.py b/tests/test_routes_workflow_jobs_extra.py index e7d82ee..e0132d3 100644 --- a/tests/test_routes_workflow_jobs_extra.py +++ b/tests/test_routes_workflow_jobs_extra.py @@ -31,7 +31,11 @@ async def test_cancel_workflow_not_owned_raises_404(): async def test_cancel_workflow_api_error_maps_502(): with ( patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), - patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, side_effect=SeqeraAPIError("down")), + patch( + "app.routes.workflow.jobs.cancel_seqera_workflow", + new_callable=AsyncMock, + side_effect=SeqeraAPIError("down"), + ), ): with pytest.raises(HTTPException) as exc: await cancel_workflow("wf-1", UUID("11111111-1111-1111-1111-111111111111"), Mock()) @@ -43,8 +47,22 @@ async def test_get_job_details_success(): owned_run = SimpleNamespace(workflow=SimpleNamespace(name="BindCraft"), id="rid") with ( patch("app.routes.workflow.jobs.get_owned_run", return_value=owned_run), - patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"runName": "job-x", "status": "SUCCEEDED", "submit": "2026-02-01T10:00:00Z"}}), - patch("app.routes.workflow.jobs.ensure_completed_run_score", new_callable=AsyncMock, return_value=0.912), + patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + return_value={ + "workflow": { + "runName": "job-x", + "status": "SUCCEEDED", + "submit": "2026-02-01T10:00:00Z", + } + }, + ), + patch( + "app.routes.workflow.jobs.ensure_completed_run_score", + new_callable=AsyncMock, + return_value=0.912, + ), ): result = await get_job_details("wf-1", UUID("11111111-1111-1111-1111-111111111111"), Mock()) @@ -59,9 +77,21 @@ async def test_delete_job_success_cancels_running_and_deletes_local_rows(): owned_run = SimpleNamespace(id="rid", workflow=None) with ( patch("app.routes.workflow.jobs.get_owned_run", return_value=owned_run), - patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"status": "RUNNING"}}), - patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, return_value=None), - patch("app.routes.workflow.jobs.delete_seqera_workflow", new_callable=AsyncMock, return_value=None), + patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + return_value={"workflow": {"status": "RUNNING"}}, + ), + patch( + "app.routes.workflow.jobs.cancel_seqera_workflow", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "app.routes.workflow.jobs.delete_seqera_workflow", + new_callable=AsyncMock, + return_value=None, + ), ): resp = await delete_job("wf-1", UUID("11111111-1111-1111-1111-111111111111"), db) @@ -80,8 +110,16 @@ def _owned(_db, _uid, run_id): with ( patch("app.routes.workflow.jobs.get_owned_run", side_effect=_owned), - patch("app.routes.workflow.jobs.describe_workflow", new_callable=AsyncMock, return_value={"workflow": {"status": "FAILED"}}), - patch("app.routes.workflow.jobs.delete_seqera_workflow", new_callable=AsyncMock, return_value=None), + patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + return_value={"workflow": {"status": "FAILED"}}, + ), + patch( + "app.routes.workflow.jobs.delete_seqera_workflow", + new_callable=AsyncMock, + return_value=None, + ), ): out = await bulk_delete_jobs( BulkDeleteJobsRequest(runIds=["ok", "missing"]), diff --git a/tests/test_routes_workflows.py b/tests/test_routes_workflows.py index a4b7871..2f7e4e0 100644 --- a/tests/test_routes_workflows.py +++ b/tests/test_routes_workflows.py @@ -118,11 +118,17 @@ def test_launch_invalid_payload(client: TestClient): def test_cancel_workflow_success(client: TestClient): """Test successful workflow cancellation.""" - client.app.dependency_overrides[get_current_user_id] = lambda: UUID("11111111-1111-1111-1111-111111111111") + client.app.dependency_overrides[get_current_user_id] = lambda: UUID( + "11111111-1111-1111-1111-111111111111" + ) client.app.dependency_overrides[get_db] = lambda: iter([None]) with ( patch("app.routes.workflow.jobs.get_owned_run", return_value=object()), - patch("app.routes.workflow.jobs.cancel_seqera_workflow", new_callable=AsyncMock, return_value=None), + patch( + "app.routes.workflow.jobs.cancel_seqera_workflow", + new_callable=AsyncMock, + return_value=None, + ), ): response = client.post("/api/workflows/run_123/cancel") diff --git a/tests/test_services_job_utils.py b/tests/test_services_job_utils.py index 68abfdc..48de7c5 100644 --- a/tests/test_services_job_utils.py +++ b/tests/test_services_job_utils.py @@ -76,7 +76,9 @@ async def test_ensure_completed_run_score_branches(): # calculate + add path db_new = _DB(scalar=None) - with patch("app.services.job_utils.calculate_csv_column_max", new_callable=AsyncMock, return_value=1.23): + with patch( + "app.services.job_utils.calculate_csv_column_max", new_callable=AsyncMock, return_value=1.23 + ): score = await job_utils.ensure_completed_run_score(db_new, run, "Completed") assert score == 1.0 assert db_new.added is not None @@ -84,5 +86,9 @@ async def test_ensure_completed_run_score_branches(): # calculate failure path db_fail = _DB(scalar=None) - with patch("app.services.job_utils.calculate_csv_column_max", new_callable=AsyncMock, side_effect=ValueError("bad")): + with patch( + "app.services.job_utils.calculate_csv_column_max", + new_callable=AsyncMock, + side_effect=ValueError("bad"), + ): assert await job_utils.ensure_completed_run_score(db_fail, run, "Completed") is None From 8635d6b16e561769e81724163cfaeeac9b7da95c Mon Sep 17 00:00:00 2001 From: laclac102 Date: Wed, 4 Feb 2026 09:59:11 +1100 Subject: [PATCH 6/6] fix: improve lint --- app/routes/dependencies.py | 6 ++++-- app/services/job_utils.py | 19 ++++++++++++------- app/services/seqera_client.py | 11 +++++++---- 3 files changed, 23 insertions(+), 13 deletions(-) diff --git a/app/routes/dependencies.py b/app/routes/dependencies.py index 5f0374c..3441538 100644 --- a/app/routes/dependencies.py +++ b/app/routes/dependencies.py @@ -2,6 +2,8 @@ from __future__ import annotations +from collections.abc import Generator +from typing import cast from uuid import UUID from fastapi import Depends, Header, HTTPException, status @@ -12,7 +14,7 @@ from ..db.models.core import AppUser -def get_db() -> Session: +def get_db() -> Generator[Session, None, None]: db = SessionLocal() try: yield db @@ -39,4 +41,4 @@ def get_current_user_id( detail="Unknown user", ) - return user.id + return cast(UUID, user.id) diff --git a/app/services/job_utils.py b/app/services/job_utils.py index 0b37e7a..bc5891b 100644 --- a/app/services/job_utils.py +++ b/app/services/job_utils.py @@ -2,8 +2,10 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime from decimal import Decimal +from typing import Any from uuid import UUID from sqlalchemy import select @@ -13,18 +15,19 @@ from .s3 import S3ConfigurationError, S3ServiceError, calculate_csv_column_max -def coerce_workflow_payload(payload: dict) -> dict: - if isinstance(payload, dict) and isinstance(payload.get("workflow"), dict): - return payload["workflow"] - return payload +def coerce_workflow_payload(payload: Mapping[str, Any]) -> dict[str, Any]: + workflow = payload.get("workflow") + if isinstance(workflow, Mapping): + return dict(workflow) + return dict(payload) -def extract_pipeline_status(payload: dict) -> str: +def extract_pipeline_status(payload: Mapping[str, Any]) -> str: workflow = coerce_workflow_payload(payload) return str(workflow.get("status") or "UNKNOWN") -def parse_submit_datetime(payload: dict) -> datetime | None: +def parse_submit_datetime(payload: Mapping[str, Any]) -> datetime | None: workflow = coerce_workflow_payload(payload) submit_str = workflow.get("submit") or workflow.get("dateCreated") if not submit_str: @@ -64,7 +67,9 @@ def get_score_by_seqera_run_id(db: Session, user_id: UUID) -> dict[str, float]: .where(WorkflowRun.owner_user_id == user_id) ).all() return { - seqera_run_id: _round_score(score) for seqera_run_id, score in rows if score is not None + str(seqera_run_id): rounded + for seqera_run_id, score in rows + if seqera_run_id and (rounded := _round_score(score)) is not None } diff --git a/app/services/seqera_client.py b/app/services/seqera_client.py index afa7344..cf823a5 100644 --- a/app/services/seqera_client.py +++ b/app/services/seqera_client.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from typing import Any, cast import httpx @@ -36,7 +37,7 @@ def _headers(token: str) -> dict[str, str]: async def list_workflows_raw( workspace_id: str | None = None, search_query: str | None = None, -) -> dict | list: +) -> dict[str, Any] | list[Any]: api_url, token, params = _get_api_context(workspace_id) if search_query: params["search"] = search_query @@ -47,10 +48,12 @@ async def list_workflows_raw( if response.is_error: raise SeqeraAPIError(f"Failed to list workflows: {response.status_code} {response.text}") - return response.json() + return cast(dict[str, Any] | list[Any], response.json()) -async def describe_workflow_raw(workflow_id: str, workspace_id: str | None = None) -> dict: +async def describe_workflow_raw( + workflow_id: str, workspace_id: str | None = None +) -> dict[str, Any]: api_url, token, params = _get_api_context(workspace_id) url = f"{api_url}/workflow/{workflow_id}" async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: @@ -58,7 +61,7 @@ async def describe_workflow_raw(workflow_id: str, workspace_id: str | None = Non if response.is_error: raise SeqeraAPIError(f"Failed to describe workflow: {response.status_code} {response.text}") - return response.json() + return cast(dict[str, Any], response.json()) async def cancel_workflow_raw(workflow_id: str, workspace_id: str | None = None) -> None: