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/app/routes/dependencies.py b/app/routes/dependencies.py new file mode 100644 index 0000000..3441538 --- /dev/null +++ b/app/routes/dependencies.py @@ -0,0 +1,44 @@ +"""Shared FastAPI dependencies for route modules.""" + +from __future__ import annotations + +from collections.abc import Generator +from typing import cast +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() -> Generator[Session, None, None]: + 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 cast(UUID, 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..c98c701 --- /dev/null +++ b/app/routes/workflow/jobs.py @@ -0,0 +1,258 @@ +"""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_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), + 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_filter 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 87ae44f..f8b2e59 100644 --- a/app/routes/workflows.py +++ b/app/routes/workflows.py @@ -1,158 +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 fastapi import APIRouter -from fastapi import APIRouter, HTTPException, Query, status - -from ..schemas.workflows import ( - CancelWorkflowResponse, - DatasetUploadRequest, - DatasetUploadResponse, - LaunchDetails, - LaunchLogs, - ListRunsResponse, - 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 .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"]) - - -@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), -) -> 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) - - -@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 95f63a9..f29f8b3 100644 --- a/app/schemas/workflows.py +++ b/app/schemas/workflows.py @@ -3,11 +3,46 @@ 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 +179,60 @@ 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") + + +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..bc5891b --- /dev/null +++ b/app/services/job_utils.py @@ -0,0 +1,106 @@ +"""Helpers for job ownership, score handling, and Seqera payload parsing.""" + +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 +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: 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: Mapping[str, Any]) -> str: + workflow = coerce_workflow_payload(payload) + return str(workflow.get("status") or "UNKNOWN") + + +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: + 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 { + str(seqera_run_id): rounded + for seqera_run_id, score in rows + if seqera_run_id and (rounded := _round_score(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 new file mode 100644 index 0000000..ee61888 --- /dev/null +++ b/app/services/seqera.py @@ -0,0 +1,60 @@ +"""Seqera Platform API integration for workflow operations.""" + +from __future__ import annotations + +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 + +__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, + 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.""" + _ = (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, + ) + + +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) + + +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 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..cf823a5 --- /dev/null +++ b/app/services/seqera_client.py @@ -0,0 +1,94 @@ +"""Low-level HTTP calls to Seqera API.""" + +from __future__ import annotations + +import os +from typing import Any, cast + +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[str, Any] | list[Any]: + 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 cast(dict[str, Any] | list[Any], response.json()) + + +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: + 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 cast(dict[str, Any], 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 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 new file mode 100644 index 0000000..86a549d --- /dev/null +++ b/tests/test_routes_jobs.py @@ -0,0 +1,179 @@ +"""Tests for job listing and details endpoints.""" + +from __future__ import annotations + +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.routes.dependencies import get_current_user_id, get_db + + +class _DummyResult: + def all(self): + return [] + + def scalar_one_or_none(self): + return None + + +class _DummyDB: + def execute(self, *_args, **_kwargs): + return _DummyResult() + + def commit(self): + return None + + def rollback(self): + return None + + def delete(self, *_args, **_kwargs): + return None + + +@pytest.fixture +def client(): + app = create_app() + + def override_get_current_user_id() -> UUID: + return UUID("11111111-1111-1111-1111-111111111111") + + def override_get_db(): + yield _DummyDB() + + app.dependency_overrides[get_current_user_id] = override_get_current_user_id + app.dependency_overrides[get_db] = override_get_db + return TestClient(app) + + +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() + 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( + "/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" + + +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): + response = client.get("/api/workflows/jobs?offset=-1") + assert response.status_code == 422 + + +def test_list_jobs_configuration_error(client): + from app.services.seqera import SeqeraConfigurationError + + 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") + + assert response.status_code == 500 + + +def test_list_jobs_api_error(client): + from app.services.seqera import SeqeraAPIError + + with patch( + "app.routes.workflow.jobs.describe_workflow", + new_callable=AsyncMock, + side_effect=SeqeraAPIError("Seqera API is down"), + ), 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..e0132d3 --- /dev/null +++ b/tests/test_routes_workflow_jobs_extra.py @@ -0,0 +1,131 @@ +"""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..2f7e4e0 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,19 @@ 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 +139,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..48de7c5 --- /dev/null +++ b/tests/test_services_job_utils.py @@ -0,0 +1,94 @@ +"""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 new file mode 100644 index 0000000..276e9da --- /dev/null +++ b/tests/test_services_workflow_listing.py @@ -0,0 +1,249 @@ +"""Tests for Seqera workflow listing service.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from app.services.seqera import ( + SeqeraAPIError, + SeqeraConfigurationError, + 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")