diff --git a/.env.example b/.env.example index 821e99a6..10621969 100644 --- a/.env.example +++ b/.env.example @@ -63,3 +63,10 @@ GLOBUS_S3_COLLECTION_ID=dev-globus-s3-collection-id GLOBUS_GADI_COLLECTION_ROOT=on-gadi-collection-root GLOBUS_INPUT_DIR=on-gadi-input-dir GLOBUS_OUTPUT_DIR=on-gadi-output-dir + +# GitHub App used to look up workflow repo commits (see workflow_repo_staging.py). +# Preferred over a personal access token since this is an org-owned service. +GITHUB_WORKFLOW_STAGING_AUTOMATION_APP_ID=github-app-id +GITHUB_WORKFLOW_STAGING_AUTOMATION_APP_PRIVATE_KEY= +# Local-dev-only fallback if you don't want to set up a GitHub App - never use in production. +GITHUB_WORKFLOW_STAGING_AUTOMATION_TOKEN= diff --git a/Dockerfile b/Dockerfile index ece27f57..8135c4b8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,6 +6,14 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ WORKDIR /app +# git is required by app/services/workflow_repo_staging.py, which clones +# workflow pipeline repos (with real .git metadata, not just extracted source) +# to stage them onto Gadi via S3 + Globus - Gadi compute nodes have no network +# access, so Nextflow can't fetch/clone them itself at run time. +RUN apt-get update \ + && apt-get install -y --no-install-recommends git \ + && rm -rf /var/lib/apt/lists/* + # Install UV COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv diff --git a/alembic/versions/20260818_033139_add_workflow_repo_staging_cache_b1f4c9a7e021.py b/alembic/versions/20260818_033139_add_workflow_repo_staging_cache_b1f4c9a7e021.py new file mode 100644 index 00000000..f348bc27 --- /dev/null +++ b/alembic/versions/20260818_033139_add_workflow_repo_staging_cache_b1f4c9a7e021.py @@ -0,0 +1,29 @@ +"""add workflow repo staging cache""" + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'b1f4c9a7e021' +down_revision = '70ac6b86efb4' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column('workflows', sa.Column('repo_staged_commit_sha', sa.Text(), nullable=True)) + op.add_column('workflows', sa.Column('repo_staging_status', sa.String(length=20), nullable=True)) + op.add_column('workflows', sa.Column('repo_gadi_path', sa.Text(), nullable=True)) + op.add_column('workflows', sa.Column('repo_staging_transfer_id', sa.Text(), nullable=True)) + op.add_column('workflows', sa.Column('repo_staging_error_message', sa.Text(), nullable=True)) + op.add_column('workflows', sa.Column('repo_staging_updated_at', sa.DateTime(timezone=True), nullable=True)) + + +def downgrade() -> None: + op.drop_column('workflows', 'repo_staging_updated_at') + op.drop_column('workflows', 'repo_staging_error_message') + op.drop_column('workflows', 'repo_staging_transfer_id') + op.drop_column('workflows', 'repo_gadi_path') + op.drop_column('workflows', 'repo_staging_status') + op.drop_column('workflows', 'repo_staged_commit_sha') diff --git a/app/config.py b/app/config.py index f65a5849..0658d6e3 100644 --- a/app/config.py +++ b/app/config.py @@ -75,6 +75,7 @@ class AdminSettings(NestedSettings): cookie_secure: bool session_secret: str roles_claim: str + title: str = "SBP Backend Admin" model_config = SettingsConfigDict(env_prefix="DB_ADMIN_") @@ -125,6 +126,30 @@ def reject_empty_values(cls, value: str) -> str: model_config = SettingsConfigDict(env_prefix="GLOBUS_") +class GithubSettings(NestedSettings): + """Credentials for GitHub API calls in workflow_repo_staging.py. + + Prefer the workflow-staging-automation GitHub App + (workflow_staging_automation_app_id/_private_key) - org-owned, not tied + to a person. workflow_staging_automation_token (a PAT) is a local-dev-only + fallback. + """ + + workflow_staging_automation_app_id: str | None = None + workflow_staging_automation_app_private_key: str | None = None + workflow_staging_automation_token: str | None = None + + @field_validator("workflow_staging_automation_app_private_key") + @classmethod + def _normalize_private_key_newlines(cls, value: str | None) -> str | None: + # Secrets managers often flatten PEM newlines to literal "\n" - restore them. + if value is None: + return None + return value.replace("\\n", "\n") + + model_config = SettingsConfigDict(env_prefix="GITHUB_") + + class Settings(BaseSettings): """ Core settings for the app. @@ -150,6 +175,7 @@ class Settings(BaseSettings): admin: AdminSettings = Field(default_factory=AdminSettings) auth: AuthSettings = Field(default_factory=AuthSettings) globus: GlobusSettings = Field(default_factory=GlobusSettings) + github: GithubSettings = Field(default_factory=GithubSettings) model_config = SettingsConfigDict(env_file=".env", dotenv_filtering="only_existing") diff --git a/app/db/admin.py b/app/db/admin.py index 16e38aa1..a06c92d3 100644 --- a/app/db/admin.py +++ b/app/db/admin.py @@ -141,12 +141,27 @@ class WorkflowAdmin(ModelView): "default_revision", "config_path", "prerun_script_path", + # Cache of the repo checkout currently staged on Gadi via Globus for + # this workflow (see app/services/workflow_repo_staging.py) - a single + # slot shared by every run, not per-run history. + "repo_staged_commit_sha", + "repo_staging_status", + "repo_gadi_path", + "repo_staging_transfer_id", + "repo_staging_updated_at", + "repo_staging_error_message", ] + exclude_fields_from_list = ["repo_staging_transfer_id", "repo_staging_error_message"] _NULLABLE_FIELDS = ( "description", "tool", "prerun_script_path", + "repo_staged_commit_sha", + "repo_staging_status", + "repo_gadi_path", + "repo_staging_transfer_id", + "repo_staging_error_message", ) def _nullify_empty_fields(self, obj: Any) -> None: @@ -681,7 +696,7 @@ def mount_db_admin(app: FastAPI, settings: Settings) -> None: # before this mount) so it stays available independently of the dashboard. _mount_db_debug_api(app) _mount_admin_ui_assets(app) - _mount_starlette_admin(app) + _mount_starlette_admin(app, settings) def _mount_admin_ui_assets(app: FastAPI) -> None: @@ -700,7 +715,7 @@ def workflow_run_export_js() -> Response: app.include_router(router) -def _mount_starlette_admin(app: FastAPI) -> None: +def _mount_starlette_admin(app: FastAPI, settings: Settings) -> None: session_cookie_name = _get_admin_session_cookie_name() oauth_state_cookie_name = "sbp_admin_oauth_state" oauth_verifier_cookie_name = "sbp_admin_oauth_verifier" @@ -937,7 +952,7 @@ async def serialize_value( admin = Admin( engine=engine, - title=os.getenv("DB_ADMIN_TITLE", "SBP Backend Admin"), + title=settings.admin.title, templates_dir=_ADMIN_TEMPLATES_DIR, auth_provider=Auth0AdminAuthProvider(), # Timestamps are stored as UTC; always display them in Sydney/Melbourne diff --git a/app/db/models/core.py b/app/db/models/core.py index 0286af9a..9d52bef6 100644 --- a/app/db/models/core.py +++ b/app/db/models/core.py @@ -53,6 +53,11 @@ class AppUser(Base): workflow_runs: Mapped[list[WorkflowRun]] = relationship(back_populates="owner") +# Mirrors DataTransferStatus (defined further below) - kept as its own alias +# since a workflow's repo staging is a distinct cache concept, not a DataTransfer. +RepoStagingStatus = Literal["pending", "in_progress", "completed", "failed"] + + class Workflow(Base): __tablename__ = "workflows" __table_args__ = ( @@ -81,6 +86,20 @@ class Workflow(Base): config_path: Mapped[str] = mapped_column(Text, nullable=False) prerun_script_path: Mapped[str | None] = mapped_column(Text, nullable=True) tool: Mapped[str | None] = mapped_column(Text, nullable=True) + # Cache of the most recently staged (repo_url, default_revision) commit on + # Gadi, shared across every run of this workflow + repo_staged_commit_sha: Mapped[str | None] = mapped_column(Text, nullable=True) + repo_staging_status: Mapped[RepoStagingStatus | None] = mapped_column( + String(length=20), nullable=True + ) + repo_gadi_path: Mapped[str | None] = mapped_column(Text, nullable=True) + # Holds Globus's submission id until submission succeeds, then the real + # Globus task id thereafter - same reuse trick as DataTransfer.transfer_id. + repo_staging_transfer_id: Mapped[str | None] = mapped_column(Text, nullable=True) + repo_staging_error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + repo_staging_updated_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) runs: Mapped[list[WorkflowRun]] = relationship(back_populates="workflow") diff --git a/app/routes/workflows.py b/app/routes/workflows.py index 41f44eb8..4fe13e8f 100644 --- a/app/routes/workflows.py +++ b/app/routes/workflows.py @@ -44,7 +44,7 @@ SinglePredictionEntity, validate_single_prediction_entities, ) -from ..services.bindflow_executor import prepare_bindflow_workflow +from ..services.bindflow_executor import prepare_bindflow_workflow, resolve_bindflow_asset_path from ..services.credits import ( WorkflowCreditsResponse, is_credits_enabled, @@ -69,6 +69,7 @@ ) from ..services.seqera_errors import WorkflowLaunchError from ..services.wisps_executor import prepare_wisps_workflow +from ..services.workflow_repo_staging import RepoStagingError, ensure_repo_staging_requested from .dependencies import ( get_client_ip, get_current_user_id, @@ -281,6 +282,64 @@ async def _stage_referenced_samplesheet_file( return csv_upload.file_key +async def _rewrite_bindflow_settings_asset_columns( + *, s3_input_key: str, repo_assets_path: str +) -> str: + """Fill in settings_filters/settings_advanced samplesheet columns with the + local Gadi path to bindflow's bundled default JSON files - the frontend + leaves these fields unset (see sbp-portal's de-novo-design.ts), so + resolve_bindflow_asset_path fills in the known default; any other, + genuinely custom value is left untouched. + + Unlike starting_pdb (_stage_referenced_samplesheet_file, above) these + columns don't need their own Globus transfer or RunInput/DataTransfer + bookkeeping - they reference files that are already part of the workflow + repo, staged as a whole. This is a plain string rewrite, re-uploaded only + if something actually changed. + """ + try: + samplesheet_rows = await read_csv_from_s3(s3_input_key) + except S3ConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"S3 configuration error: {exc}", + ) from exc + except S3ServiceError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"Failed to read samplesheet at s3InputKey: {exc}", + ) from exc + if not samplesheet_rows: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Samplesheet at s3InputKey is empty.", + ) + samplesheet_row = samplesheet_rows[0] + changed = False + for field_name in ("settings_filters", "settings_advanced"): + resolved = resolve_bindflow_asset_path( + field_name, samplesheet_row.get(field_name), repo_assets_path=repo_assets_path + ) + if resolved is not None and resolved != samplesheet_row.get(field_name): + samplesheet_row[field_name] = resolved + changed = True + if not changed: + return s3_input_key + try: + csv_upload = await upload_csv_to_s3(samplesheet_row) + except S3ConfigurationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"S3 configuration error: {exc}", + ) from exc + except S3ServiceError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"Failed to re-upload corrected samplesheet: {exc}", + ) from exc + return csv_upload.file_key + + def _stage_wisps_fasta( *, db_session: Session, @@ -417,6 +476,28 @@ async def launch_workflow( detail=f"Workflow '{workflow.name}' is missing default_revision in workflows table.", ) + # Gadi compute nodes have no network access, so Nextflow can't fetch the + # pipeline from GitHub itself - it must already be staged there + try: + repo_staging_locations = ensure_repo_staging_requested( + db_session, workflow, settings=settings + ) + except RepoStagingError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"Failed to resolve workflow repo: {exc}", + ) from exc + # Seqera's launch API validates `pipeline` as a URL, Nextflow's local-path scheme is "file:" with a + # single slash before the (already-absolute) path + repo_gadi_path = repo_staging_locations.gadi_path + pipeline_url = f"file:{repo_gadi_path}" + # bindcraft's settings_filters/settings_advanced reference files bundled + # inside the workflow repo itself - repo_gadi_path is a bare git repo + # (see build_repo_gadi_path) with no working-tree files on disk, so asset + # resolution uses the separate plain checkout staged alongside it instead + # (see build_repo_assets_gadi_path, _rewrite_bindflow_settings_asset_columns). + repo_assets_path = repo_staging_locations.assets_gadi_path + user = db_session.execute( select(AppUser.email).where(AppUser.id == current_user_id) ).one_or_none() @@ -518,6 +599,9 @@ async def launch_workflow( workflow_name=workflow_name, globus_settings=settings.globus, ) + s3_input_key = await _rewrite_bindflow_settings_asset_columns( + s3_input_key=s3_input_key, repo_assets_path=repo_assets_path + ) elif is_proteinfold_launch: s3_input_key = await _stage_referenced_samplesheet_file( db_session=db_session, @@ -583,7 +667,7 @@ async def launch_workflow( settings=settings, db_session=db_session, workflow_run=workflow_run, - pipeline=workflow.repo_url, + pipeline=pipeline_url, config_path=workflow.config_path, revision=workflow.default_revision, output_id=str(run_id), @@ -603,7 +687,7 @@ async def launch_workflow( settings=settings, db_session=db_session, workflow_run=workflow_run, - pipeline=workflow.repo_url, + pipeline=pipeline_url, config_path=workflow.config_path, revision=workflow.default_revision, output_id=str(run_id), @@ -617,12 +701,14 @@ async def launch_workflow( settings=settings, db_session=db_session, workflow_run=workflow_run, - pipeline=workflow.repo_url, + pipeline=pipeline_url, config_path=workflow.config_path, revision=workflow.default_revision, output_id=str(run_id), + form_data=payload.formData, user_details=user_details, staged_input_location=staged_input_location, + repo_assets_path=repo_assets_path, ) elif workflow_name in ("interaction-screening", "bulk-prediction"): assert wisps_form_data is not None @@ -633,7 +719,7 @@ async def launch_workflow( settings=settings, db_session=db_session, workflow_run=workflow_run, - pipeline=workflow.repo_url, + pipeline=pipeline_url, revision=workflow.default_revision, config_path=workflow.config_path, form_data=wisps_form_data, diff --git a/app/run_scheduler.py b/app/run_scheduler.py index 58c4f4b2..c47f8019 100644 --- a/app/run_scheduler.py +++ b/app/run_scheduler.py @@ -18,11 +18,13 @@ submit_pending_jobs, sync_completed_workflow_runs, sync_data_transfers, + sync_workflow_repo_staging, ) SUBMIT_INTERVAL = IntervalTrigger(minutes=5) SYNC_INTERVAL = IntervalTrigger(minutes=10) DATA_TRANSFER_SYNC_INTERVAL = IntervalTrigger(minutes=2) +REPO_STAGING_SYNC_INTERVAL = IntervalTrigger(minutes=2) # Fixed AEST (UTC+10), no DST. Not Australia/Sydney: APScheduler 3.11.3's CronTrigger # miscalculates day=1 across Sydney's October DST switch and skips November entirely. MONTHLY_TRIGGER = CronTrigger(day=1, hour=0, minute=0, timezone="Australia/Brisbane") @@ -68,6 +70,20 @@ def main(dry_run: bool = False): max_instances=1, replace_existing=True, ) + logger.info( + f"Adding sync_workflow_repo_staging to scheduler: trigger = {REPO_STAGING_SYNC_INTERVAL}" + ) + SCHEDULER.add_job( + sync_workflow_repo_staging, + kwargs={"dry_run": dry_run}, + jobstore="memory", + trigger=REPO_STAGING_SYNC_INTERVAL, + next_run_time=datetime.now(tz=UTC) + timedelta(minutes=1), + id="sync_workflow_repo_staging", + misfire_grace_time=60, + max_instances=1, + replace_existing=True, + ) logger.info( f"Adding monthly refresh_user_credits to scheduler: trigger = {MONTHLY_TRIGGER}" ) diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 566a4d2d..ec6b6469 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -11,11 +11,11 @@ from sqlalchemy.orm import Session from ..config import Settings, get_settings -from ..db.models.core import AppUser, DataTransfer +from ..db.models.core import AppUser, DataTransfer, Workflow from ..db.models.job_queue import QueuedJob from ..routes.dependencies import get_db from ..schemas.workflows.shared import WorkflowName -from ..services import globus_transfer, health, seqera +from ..services import globus_transfer, health, seqera, workflow_repo_staging from ..services.bindflow_executor import launch_bindflow_workflow from ..services.credits import MONTHLY_CREDIT_REFRESH_ACTOR, SBP_USER_CREDIT_ALLOWANCE from ..services.job_sync import get_runs_requiring_sync, sync_workflow_runs @@ -328,3 +328,26 @@ def sync_data_transfers(dry_run: bool = False, *, db_session: Session | None = N f"checked={result.checked}, submitted={result.submitted}, " f"completed={result.completed}, failed={result.failed}, errored={result.errored}." ) + + +def sync_workflow_repo_staging(dry_run: bool = False): + """Submit pending and poll in-progress workflow repo stagings (GitHub repo + checkouts cached on Gadi via S3 + Globus, see workflow_repo_staging.py), + promoting any run still "staging" on a workflow whose repo just finished.""" + logger.info("Checking for workflow repo stagings to sync...") + db_session = next(get_db()) + if dry_run: + pending_count = db_session.scalar( + select(func.count()) + .select_from(Workflow) + .where(Workflow.repo_staging_status.in_(["pending", "in_progress"])) + ) + logger.info(f"Dry run - found {pending_count} workflow repo staging(s) requiring sync.") + return + + result = workflow_repo_staging.sync_workflow_repo_staging(db_session) + logger.info( + "Finished syncing workflow repo stagings: " + f"checked={result.checked}, submitted={result.submitted}, " + f"completed={result.completed}, failed={result.failed}, errored={result.errored}." + ) diff --git a/app/services/bindflow_executor.py b/app/services/bindflow_executor.py index cf3805a3..b291a1c6 100644 --- a/app/services/bindflow_executor.py +++ b/app/services/bindflow_executor.py @@ -10,7 +10,7 @@ from ..config import Settings, get_settings from ..db.models import QueuedJob, WorkflowRun -from ..schemas.workflows.shared import WorkflowLaunchForm, WorkflowUserDetails +from ..schemas.workflows.shared import WorkflowFormData, WorkflowLaunchForm, WorkflowUserDetails from .bindflow_config import ( get_bindflow_config_profiles, get_bindflow_config_text, @@ -31,6 +31,29 @@ logger = logging.getLogger(__name__) +# settings_filters/settings_advanced reference default JSON files bundled in +# the bindflow repo itself. The frontend leaves these fields unset (see +# sbp-portal's de-novo-design.ts) and relies on the backend to fill in the +# local Gadi path: resolve_bindflow_asset_path below fills in the known +# default for an empty value; anything else (a genuinely custom value) is +# passed through unchanged rather than guessed at, since staging arbitrary +# user-supplied settings files isn't supported yet. +_BINDFLOW_DEFAULT_ASSET_RELATIVE_PATHS = { + "settings_filters": "assets/bindcraft/default_filters.json", + "settings_advanced": "assets/bindcraft/default_4stage_multimer.json", +} + + +def resolve_bindflow_asset_path( + field_name: str, value: object, *, repo_assets_path: str +) -> str | None: + if isinstance(value, str) and value.strip(): + return value + default_relative_path = _BINDFLOW_DEFAULT_ASSET_RELATIVE_PATHS.get(field_name) + if default_relative_path is None: + return None + return f"{repo_assets_path}/{default_relative_path}" + async def prepare_bindflow_workflow( # pylint: disable=too-many-locals form: WorkflowLaunchForm, @@ -42,8 +65,10 @@ async def prepare_bindflow_workflow( # pylint: disable=too-many-locals config_path: str, revision: str | None = None, output_id: str | None = None, + form_data: WorkflowFormData, user_details: WorkflowUserDetails, staged_input_location: str, + repo_assets_path: str, commit: bool = False, ) -> QueuedJob: """Build and queue a bindflow launch payload.""" @@ -62,6 +87,20 @@ async def prepare_bindflow_workflow( # pylint: disable=too-many-locals out_dir = f"s3://{s3_bucket}/{output_key}" default_params = get_bindflow_default_params(out_dir, staged_input_location) + settings_filters = resolve_bindflow_asset_path( + "settings_filters", + form_data.extra_fields.get("settings_filters"), + repo_assets_path=repo_assets_path, + ) + settings_advanced = resolve_bindflow_asset_path( + "settings_advanced", + form_data.extra_fields.get("settings_advanced"), + repo_assets_path=repo_assets_path, + ) + if settings_filters: + default_params["settings_filters"] = settings_filters + if settings_advanced: + default_params["settings_advanced"] = settings_advanced # Serialize to YAML params_text = params_to_yaml_text(default_params) diff --git a/app/services/globus_transfer.py b/app/services/globus_transfer.py index 4fbcd6d3..950a3700 100644 --- a/app/services/globus_transfer.py +++ b/app/services/globus_transfer.py @@ -13,6 +13,7 @@ from ..config import GlobusSettings, get_settings from ..db.models.core import DataTransfer, DataTransferStatus +from ..db.models.job_queue import QueuedJob from .globus_client import get_transfer_client from .globus_errors import GlobusConfigurationError, GlobusTransferError @@ -205,7 +206,8 @@ def poll_transfer( def _notify_launcher(db: Session, data_transfer: DataTransfer) -> None: - """After an input transfer settles, move the run's QueuedJob out of "staging". + """After an input transfer settles, try to move the run's QueuedJob out of + "staging". Observed once in production: all input transfers for a run showed ``completed`` minutes before Nextflow started, yet the first process still @@ -222,23 +224,48 @@ def _notify_launcher(db: Session, data_transfer: DataTransfer) -> None: return queued_job = data_transfer.workflow_run.get_queued_job(db) - if queued_job is None or queued_job.status != "staging": + if queued_job is None: return + _try_promote_staging_job(db, queued_job) - if data_transfer.status == "failed": - queued_job.status = "failed" - queued_job.error = f"Input staging failed: {data_transfer.error_message}" - db.add(queued_job) - db.commit() + +def _try_promote_staging_job(db: Session, queued_job: QueuedJob) -> None: + """Flip a "staging" QueuedJob to "pending" once every gate clears: all of + its input DataTransfers completed, and its workflow's repo staging (if + applicable - see workflow_repo_staging.py) completed. Fails the job + immediately if any gate reports failure, rather than leaving it stuck. + + Called both after an input transfer settles (_notify_launcher, above) and + after a workflow's repo staging settles (workflow_repo_staging.py) - either + event can be the last one a given run was waiting on. + """ + if queued_job.status != "staging": return - other_input_transfers = db.scalars( + input_transfers = db.scalars( select(DataTransfer).where( - DataTransfer.workflow_run_id == data_transfer.workflow_run_id, + DataTransfer.workflow_run_id == queued_job.workflow_run_id, DataTransfer.direction == "input", ) ).all() - if not all(transfer.status == "completed" for transfer in other_input_transfers): + failed_transfer = next((t for t in input_transfers if t.status == "failed"), None) + if failed_transfer is not None: + queued_job.status = "failed" + queued_job.error = f"Input staging failed: {failed_transfer.error_message}" + db.add(queued_job) + db.commit() + return + if not all(transfer.status == "completed" for transfer in input_transfers): + return + + workflow = queued_job.workflow + if workflow.repo_staging_status == "failed": + queued_job.status = "failed" + queued_job.error = f"Workflow repo staging failed: {workflow.repo_staging_error_message}" + db.add(queued_job) + db.commit() + return + if workflow.repo_staging_status is not None and workflow.repo_staging_status != "completed": return queued_job.status = "pending" diff --git a/app/services/workflow_repo_staging.py b/app/services/workflow_repo_staging.py new file mode 100644 index 00000000..b32c390a --- /dev/null +++ b/app/services/workflow_repo_staging.py @@ -0,0 +1,480 @@ +"""Stage a workflow's GitHub repo onto Gadi via S3 + Globus, cached by commit. + +Gadi compute nodes have no network access, so Nextflow can't fetch pipeline +code from GitHub itself - the repo must already be on Gadi's filesystem, the +same way input files are staged there via Globus. A repo checkout is shared +across every run of a workflow, so it's cached on the Workflow row (keyed by +commit sha) instead of per run: if default_revision still resolves to the +commit already staged/staging, launches reuse it for free. +""" + +from __future__ import annotations + +import io +import logging +import subprocess +import tarfile +import tempfile +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import cast +from urllib.parse import urlparse + +import globus_sdk +from github import Auth, Github, GithubException, GithubIntegration +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..config import GlobusSettings, Settings, get_settings +from ..db.models.core import Workflow +from .globus_client import get_transfer_client +from .globus_errors import GlobusTransferError +from .globus_transfer import _gadi_relative_path +from .s3 import get_s3_client + +logger = logging.getLogger(__name__) + +_DOWNLOAD_TIMEOUT = 120 + + +class RepoStagingError(RuntimeError): + """Raised when resolving or staging a workflow's GitHub repo fails.""" + + +def parse_github_repo(repo_url: str) -> tuple[str, str]: + """Extract (owner, repo) from a ``https://github.com//`` URL.""" + parsed = urlparse(repo_url) + if parsed.netloc != "github.com": + raise RepoStagingError(f"Only github.com repo URLs are supported, got: {repo_url}") + parts = [part for part in parsed.path.split("/") if part] + if len(parts) < 2: + raise RepoStagingError(f"Could not parse owner/repo from repo_url: {repo_url}") + owner, repo = parts[0], parts[1] + return owner, repo.removesuffix(".git") + + +def _get_github_client(owner: str, repo: str, *, settings: Settings) -> Github: + """Build a PyGithub client authorized to read owner/repo. + + Prefers GitHub App installation auth. + get_repo_installation looks up which installation of the App covers this + repo, and get_github_for_installation returns a client whose short-lived + token PyGithub mints and refreshes automatically. Falls back to a plain + token, then to unauthenticated, for local development only. + """ + github_settings = settings.github + if ( + github_settings.workflow_staging_automation_app_id + and github_settings.workflow_staging_automation_app_private_key + ): + integration = GithubIntegration( + auth=Auth.AppAuth( + github_settings.workflow_staging_automation_app_id, + github_settings.workflow_staging_automation_app_private_key, + ) + ) + installation = integration.get_repo_installation(owner, repo) + return integration.get_github_for_installation(installation.id) + if github_settings.workflow_staging_automation_token: + return Github(auth=Auth.Token(github_settings.workflow_staging_automation_token)) + return Github() + + +def resolve_latest_commit_sha( + repo_url: str, revision: str, *, settings: Settings | None = None +) -> str: + """Resolve a branch/tag/ref to its current commit sha via the GitHub API. + + One lightweight network call - no local clone needed - cheap enough to + run on every launch to check whether the staged copy is still current. + """ + settings = settings or get_settings() + owner, repo = parse_github_repo(repo_url) + client = _get_github_client(owner, repo, settings=settings) + try: + gh_repo = client.get_repo(f"{owner}/{repo}") + commit = gh_repo.get_commit(revision) + except GithubException as exc: + raise RepoStagingError( + f"Failed to resolve commit for {owner}/{repo}@{revision}: {exc}" + ) from exc + if not commit.sha: + raise RepoStagingError(f"GitHub API returned no commit sha for {owner}/{repo}@{revision}") + return cast(str, commit.sha) + + +def build_repo_s3_prefix(owner: str, repo: str, commit_sha: str) -> str: + return f"workflow-repos/{owner}-{repo}/{commit_sha}" + + +def build_repo_assets_s3_prefix(owner: str, repo: str, commit_sha: str) -> str: + return f"workflow-repos-assets/{owner}-{repo}/{commit_sha}" + + +def build_repo_gadi_path( + owner: str, repo: str, commit_sha: str, *, globus_settings: GlobusSettings +) -> str: + # ".git" suffix required: Seqera's launch API rejects a `pipeline` + # "file:" URL without it, and Nextflow then resolves that path as a + # git-dir directly (`--git-dir=`) - so this must be a real bare + # repo, not a checkout with a nested .git/ one level down. + return f"{globus_settings.gadi_collection_root}/workflow_repos/{owner}-{repo}/{commit_sha}.git" + + +def build_repo_assets_gadi_path( + owner: str, repo: str, commit_sha: str, *, globus_settings: GlobusSettings +) -> str: + """Path to a plain (non-bare) checkout of the same commit, staged next to + the bare repo build_repo_gadi_path points at - the bare repo has no + working-tree files, so pipeline-bundled assets (e.g. bindcraft's default + settings JSON) need a real checkout to be read as plain files.""" + return f"{globus_settings.gadi_collection_root}/workflow_repos/{owner}-{repo}/{commit_sha}" + + +@dataclass(frozen=True) +class RepoStagingLocations: + """Where a workflow's staged repo lives on Gadi: the bare repo for + Seqera's `pipeline` field, and the plain checkout alongside it for + pipeline-bundled assets (see build_repo_assets_gadi_path).""" + + gadi_path: str + assets_gadi_path: str + + +def ensure_repo_staging_requested( + db: Session, workflow: Workflow, *, settings: Settings | None = None +) -> RepoStagingLocations: + """Resolve the workflow's current commit, returning the Gadi paths it + will live at, and kick off staging if that commit isn't already + staged/staging. + + Called synchronously at launch time - only the cheap GitHub API call + happens here. The actual clone/upload/Globus transfer happens later via + sync_workflow_repo_staging, so launch requests stay fast. + """ + settings = settings or get_settings() + owner, repo = parse_github_repo(workflow.repo_url) + commit_sha = resolve_latest_commit_sha( + workflow.repo_url, workflow.default_revision, settings=settings + ) + gadi_path = build_repo_gadi_path(owner, repo, commit_sha, globus_settings=settings.globus) + assets_gadi_path = build_repo_assets_gadi_path( + owner, repo, commit_sha, globus_settings=settings.globus + ) + + up_to_date = workflow.repo_staged_commit_sha == commit_sha and workflow.repo_staging_status in ( + "pending", + "in_progress", + "completed", + ) + if not up_to_date: + workflow.repo_staged_commit_sha = commit_sha + workflow.repo_staging_status = "pending" + workflow.repo_gadi_path = gadi_path + workflow.repo_staging_transfer_id = None + workflow.repo_staging_error_message = None + workflow.repo_staging_updated_at = datetime.now(UTC) + db.add(workflow) + db.commit() + + return RepoStagingLocations(gadi_path=gadi_path, assets_gadi_path=assets_gadi_path) + + +def _run_git(*args: str, cwd: str) -> None: + """Run a git command, raising RepoStagingError with its stderr on failure.""" + result = subprocess.run( # noqa: S603 - fixed "git" executable, args are list-form (no shell) + ["git", *args], + cwd=cwd, + capture_output=True, + text=True, + timeout=_DOWNLOAD_TIMEOUT, + ) + if result.returncode != 0: + raise RepoStagingError( + f"git {' '.join(args)} failed (exit {result.returncode}): {result.stderr.strip()}" + ) + + +def _extract_plain_checkout(bare_dir: str, revision: str, dest_dir: str) -> None: + """Materialize a plain checkout of revision from an already-fetched bare + repo via `git archive`, piped into Python's tarfile - avoids a second + network fetch just to get working-tree files.""" + result = subprocess.run( # noqa: S603 - fixed "git" executable, args are list-form (no shell) + ["git", f"--git-dir={bare_dir}", "archive", revision], + capture_output=True, + timeout=_DOWNLOAD_TIMEOUT, + ) + if result.returncode != 0: + raise RepoStagingError( + f"git archive {revision} failed (exit {result.returncode}): " + f"{result.stderr.decode(errors='replace').strip()}" + ) + with tarfile.open(fileobj=io.BytesIO(result.stdout), mode="r|") as tar: + tar.extractall(dest_dir, filter="data") + + +def _clone_and_upload_repo( + owner: str, + repo: str, + commit_sha: str, + repo_url: str, + s3_prefix: str, + *, + revision: str, + settings: Settings, +) -> None: + """Fetch commit_sha into a bare repo and upload every file to S3 under + s3_prefix (bare repo contents land directly at s3_prefix's root, not + nested under .git/), plus a plain checkout of the same commit under + build_repo_assets_s3_prefix for pipeline-bundled asset files. + + Fetches into a branch named after `revision` rather than leaving the + commit as bare FETCH_HEAD, since a shallow-fetched bare repo has no other + branches and the `revision` sent to Seqera must reference a real ref. + """ + s3_client = get_s3_client(settings) + bucket = settings.aws.s3_bucket + refspec = f"{commit_sha}:refs/heads/{revision}" + + with tempfile.TemporaryDirectory() as tmp_dir: + logger.info("Cloning %s/%s@%s (bare) into %s", owner, repo, commit_sha, tmp_dir) + try: + _run_git("init", "--bare", cwd=tmp_dir) + _run_git("remote", "add", "origin", repo_url, cwd=tmp_dir) + _run_git("fetch", "--depth", "1", "origin", refspec, cwd=tmp_dir) + _run_git("symbolic-ref", "HEAD", f"refs/heads/{revision}", cwd=tmp_dir) + # A shallow fetch stores objects loose (one file per blob/tree) - + # `gc` packs them into one file, since a real pipeline's tree can + # be hundreds of small files otherwise. gc.packRefs=false keeps + # the branch ref itself as a loose file: we only upload actual + # files (no empty-directory markers), so a ref packed away would + # leave refs/heads with nothing to upload and never reappear on + # Gadi. + _run_git("-c", "gc.packRefs=false", "gc", cwd=tmp_dir) + except RepoStagingError as exc: + raise RepoStagingError(f"Failed to clone {owner}/{repo}@{commit_sha}: {exc}") from exc + + tmp_path = Path(tmp_dir) + uploaded = 0 + for file_path in tmp_path.rglob("*"): + if not file_path.is_file(): + continue + relative_path = file_path.relative_to(tmp_path).as_posix() + s3_client.upload_file(str(file_path), bucket, f"{s3_prefix}/{relative_path}") + uploaded += 1 + logger.info( + "Uploaded %d file(s) for bare repo %s/%s@%s to s3://%s/%s", + uploaded, + owner, + repo, + commit_sha, + bucket, + s3_prefix, + ) + + assets_s3_prefix = build_repo_assets_s3_prefix(owner, repo, commit_sha) + with tempfile.TemporaryDirectory() as assets_tmp_dir: + try: + _extract_plain_checkout(tmp_dir, revision, assets_tmp_dir) + except RepoStagingError as exc: + raise RepoStagingError( + f"Failed to extract plain checkout for {owner}/{repo}@{commit_sha}: {exc}" + ) from exc + + assets_path = Path(assets_tmp_dir) + assets_uploaded = 0 + for file_path in assets_path.rglob("*"): + if not file_path.is_file(): + continue + relative_path = file_path.relative_to(assets_path).as_posix() + s3_client.upload_file(str(file_path), bucket, f"{assets_s3_prefix}/{relative_path}") + assets_uploaded += 1 + logger.info( + "Uploaded %d plain-checkout file(s) for %s/%s@%s to s3://%s/%s", + assets_uploaded, + owner, + repo, + commit_sha, + bucket, + assets_s3_prefix, + ) + + +def stage_pending_repo( + db: Session, workflow: Workflow, *, settings: Settings | None = None +) -> None: + """Submit the S3 upload + Globus transfer for a "pending" workflow repo.""" + settings = settings or get_settings() + owner, repo = parse_github_repo(workflow.repo_url) + commit_sha = workflow.repo_staged_commit_sha + if not commit_sha: + raise RepoStagingError(f"Workflow {workflow.id} has no commit sha to stage") + + s3_prefix = build_repo_s3_prefix(owner, repo, commit_sha) + gadi_path = workflow.repo_gadi_path or build_repo_gadi_path( + owner, repo, commit_sha, globus_settings=settings.globus + ) + assets_s3_prefix = build_repo_assets_s3_prefix(owner, repo, commit_sha) + assets_gadi_path = build_repo_assets_gadi_path( + owner, repo, commit_sha, globus_settings=settings.globus + ) + + try: + _clone_and_upload_repo( + owner, + repo, + commit_sha, + workflow.repo_url, + s3_prefix, + revision=workflow.default_revision, + settings=settings, + ) + + transfer_client = get_transfer_client(settings.globus) + # Persist the submission id before submit_transfer, same as + # DataTransfer.transfer_id, so a crash-and-retry doesn't double-submit. + submission_id = workflow.repo_staging_transfer_id + if not submission_id: + submission_id = cast(str, transfer_client.get_submission_id()["value"]) + workflow.repo_staging_transfer_id = submission_id + db.add(workflow) + db.commit() + + transfer_data = globus_sdk.TransferData( + settings.globus.s3_collection_id, + settings.globus.gadi_collection_id, + submission_id=submission_id, + label=f"sbp-repo-{owner}-{repo}-{commit_sha[:12]}", + ) + # gadi_path is absolute, but the Gadi collection's root maps to + # GLOBUS_GADI_COLLECTION_ROOT, not "/" - convert before add_item. + destination_path = _gadi_relative_path(gadi_path, globus_settings=settings.globus) + transfer_data.add_item(f"/{s3_prefix}", destination_path, recursive=True) + # Second item: the plain checkout, landing as a sibling directory. + assets_destination_path = _gadi_relative_path( + assets_gadi_path, globus_settings=settings.globus + ) + transfer_data.add_item(f"/{assets_s3_prefix}", assets_destination_path, recursive=True) + result = transfer_client.submit_transfer(transfer_data) + except (RepoStagingError, globus_sdk.GlobusAPIError) as exc: + logger.warning("Workflow repo staging failed for %s: %s", workflow.id, exc) + workflow.repo_staging_status = "failed" + workflow.repo_staging_transfer_id = None + workflow.repo_staging_error_message = str(exc) + workflow.repo_staging_updated_at = datetime.now(UTC) + db.add(workflow) + db.commit() + return + + workflow.repo_gadi_path = gadi_path + workflow.repo_staging_transfer_id = result["task_id"] + workflow.repo_staging_status = "in_progress" + workflow.repo_staging_updated_at = datetime.now(UTC) + db.add(workflow) + db.commit() + + +def poll_repo_staging(db: Session, workflow: Workflow, *, settings: Settings | None = None) -> None: + """Poll Globus for an "in_progress" workflow repo transfer's task status.""" + if not workflow.repo_staging_transfer_id: + raise GlobusTransferError(f"Workflow {workflow.id} has no transfer_id to poll") + + settings = settings or get_settings() + transfer_client = get_transfer_client(settings.globus) + try: + task = transfer_client.get_task(workflow.repo_staging_transfer_id) + except globus_sdk.GlobusAPIError as exc: + # A poll failure doesn't mean the transfer failed - record it for + # visibility without touching status, same as poll_transfer. + workflow.repo_staging_error_message = f"Poll failed: {exc}" + workflow.repo_staging_updated_at = datetime.now(UTC) + db.add(workflow) + db.commit() + raise GlobusTransferError(f"Failed to poll workflow repo transfer: {exc}") from exc + + globus_status = task["status"] + if globus_status == "SUCCEEDED": + workflow.repo_staging_status = "completed" + elif globus_status == "FAILED": + workflow.repo_staging_status = "failed" + fatal_error = task.get("fatal_error") or {} + workflow.repo_staging_error_message = ( + fatal_error.get("description") or "Globus transfer failed" + ) + else: + return + + workflow.repo_staging_updated_at = datetime.now(UTC) + db.add(workflow) + db.commit() + _promote_queued_jobs_waiting_on_repo(db, workflow) + + +def _promote_queued_jobs_waiting_on_repo(db: Session, workflow: Workflow) -> None: + """Once a workflow's repo staging settles, re-check every run still + waiting on it - a repo staging event can unblock several runs at once.""" + from ..db.models.job_queue import QueuedJob + from .globus_transfer import _try_promote_staging_job # local import: avoid import cycle + + queued_jobs = db.scalars( + select(QueuedJob).where(QueuedJob.workflow_id == workflow.id, QueuedJob.status == "staging") + ).all() + for queued_job in queued_jobs: + _try_promote_staging_job(db, queued_job) + + +@dataclass(frozen=True) +class RepoStagingSyncResult: + """Outcome for one batch of workflow repo staging sync work.""" + + checked: int + submitted: int = 0 + completed: int = 0 + failed: int = 0 + errored: int = 0 + + +def sync_workflow_repo_staging( + db: Session, *, settings: Settings | None = None +) -> RepoStagingSyncResult: + """Submit pending and poll in-progress workflow repo stagings for one batch.""" + settings = settings or get_settings() + workflows = list( + db.scalars( + select(Workflow).where(Workflow.repo_staging_status.in_(["pending", "in_progress"])) + ) + ) + + submitted = completed = failed = errored = 0 + for workflow in workflows: + try: + if workflow.repo_staging_status == "pending": + stage_pending_repo(db, workflow, settings=settings) + if workflow.repo_staging_status == "in_progress": + submitted += 1 + else: + poll_repo_staging(db, workflow, settings=settings) + except (RepoStagingError, GlobusTransferError) as exc: + db.rollback() + logger.warning("Failed to sync workflow repo staging for %s: %s", workflow.id, exc) + errored += 1 + continue + except Exception: + db.rollback() + logger.exception("Unexpected error syncing workflow repo staging for %s", workflow.id) + errored += 1 + continue + + if workflow.repo_staging_status == "failed": + failed += 1 + elif workflow.repo_staging_status == "completed": + completed += 1 + + return RepoStagingSyncResult( + checked=len(workflows), + submitted=submitted, + completed=completed, + failed=failed, + errored=errored, + ) diff --git a/docs/schema_diagram b/docs/schema_diagram index ca8a2693..eef080fb 100644 --- a/docs/schema_diagram +++ b/docs/schema_diagram @@ -27,6 +27,9 @@ digraph { credit_updated_by TEXT () + + sbp_bundle_credit_granted_at + DATETIME () > URL="http://AppUser_details.html"] AppUser -> WorkflowRun [label=workflow_runs color="#1E88E5" style=dashed tooltip="Relation between AppUser and WorkflowRun"] Workflow [label=< @@ -56,6 +59,24 @@ digraph { tool TEXT () + + repo_staged_commit_sha + TEXT () + + repo_staging_status + VARCHAR(20) () + + repo_gadi_path + TEXT () + + repo_staging_transfer_id + TEXT () + + repo_staging_error_message + TEXT () + + repo_staging_updated_at + DATETIME () > URL="http://Workflow_details.html"] Workflow -> WorkflowRun [label=runs color="#1E88E5" style=dashed tooltip="Relation between Workflow and WorkflowRun"] WorkflowRun [label=< @@ -100,6 +121,12 @@ digraph { service_usage FLOAT () + + seqera_final_status + TEXT () + + sync_completed_at + DATETIME () > URL="http://WorkflowRun_details.html"] WorkflowRun -> AppUser [label=owner color="#1E88E5" style=dashed tooltip="Relation between WorkflowRun and AppUser"] WorkflowRun -> Workflow [label=workflow color="#1E88E5" style=dashed tooltip="Relation between WorkflowRun and Workflow"] @@ -199,9 +226,6 @@ digraph { status VARCHAR(20) () - provider_metadata - JSON () - created_at DATETIME () diff --git a/docs/schema_diagram.svg b/docs/schema_diagram.svg index cbe6bf65..73fa5276 100644 --- a/docs/schema_diagram.svg +++ b/docs/schema_diagram.svg @@ -4,59 +4,65 @@ - - - + + + AppUser - - -AppUser - - -id - - -UUID (PK) - - -auth0_user_id - - -TEXT (Unique) - - -name - - -TEXT () - - -email - - -TEXT (Unique) - - -credit - - -BIGINT () - - -credit_updated_at - - -DATETIME () - - -credit_updated_by - - -TEXT () + + +AppUser + + +id + + +UUID (PK) + + +auth0_user_id + + +TEXT (Unique) + + +name + + +TEXT () + + +email + + +TEXT (Unique) + + +credit + + +BIGINT () + + +credit_updated_at + + +DATETIME () + + +credit_updated_by + + +TEXT () + + +sbp_bundle_credit_granted_at + + +DATETIME () @@ -64,87 +70,99 @@ WorkflowRun - - -WorkflowRun - - -id - - -UUID (PK) - - -workflow_id - - -UUID () - - -owner_user_id - - -UUID () - - -seqera_run_id - - -TEXT () - - -binder_name - - -TEXT () - - -sample_id - - -TEXT () - - -run_name - - -TEXT () - - -submitted_form_data - - -JSON () - - -work_dir - - -TEXT () - - -launch_ip - - -TEXT () - - -submission_timestamp - - -DATETIME () - - -tool - - -TEXT () - - -service_usage - - -FLOAT () + + +WorkflowRun + + +id + + +UUID (PK) + + +workflow_id + + +UUID () + + +owner_user_id + + +UUID () + + +seqera_run_id + + +TEXT () + + +binder_name + + +TEXT () + + +sample_id + + +TEXT () + + +run_name + + +TEXT () + + +submitted_form_data + + +JSON () + + +work_dir + + +TEXT () + + +launch_ip + + +TEXT () + + +submission_timestamp + + +DATETIME () + + +tool + + +TEXT () + + +service_usage + + +FLOAT () + + +seqera_final_status + + +TEXT () + + +sync_completed_at + + +DATETIME () @@ -152,77 +170,113 @@ AppUser->WorkflowRun - - + + -workflow_runs +workflow_runs WorkflowRun->AppUser - - + + -owner +owner Workflow - - -Workflow - - -id - - -UUID (PK) - - -name - - -TEXT () - - -description - - -TEXT () - - -repo_url - - -TEXT () - - -default_revision - - -TEXT () - - -config_path - - -TEXT () - - -prerun_script_path - - -TEXT () - - -tool - - -TEXT () + + +Workflow + + +id + + +UUID (PK) + + +name + + +TEXT () + + +description + + +TEXT () + + +repo_url + + +TEXT () + + +default_revision + + +TEXT () + + +config_path + + +TEXT () + + +prerun_script_path + + +TEXT () + + +tool + + +TEXT () + + +repo_staged_commit_sha + + +TEXT () + + +repo_staging_status + + +VARCHAR(20) () + + +repo_gadi_path + + +TEXT () + + +repo_staging_transfer_id + + +TEXT () + + +repo_staging_error_message + + +TEXT () + + +repo_staging_updated_at + + +DATETIME () @@ -230,37 +284,37 @@ WorkflowRun->Workflow - - + + -workflow +workflow RunMetric - - -RunMetric - - -run_id - - -UUID (PK) - - -max_score - - -NUMERIC(8, 2) () - - -final_design_count - - -BIGINT () + + +RunMetric + + +run_id + + +UUID (PK) + + +max_score + + +NUMERIC(8, 2) () + + +final_design_count + + +BIGINT () @@ -268,37 +322,37 @@ WorkflowRun->RunMetric - - + + -metrics +metrics RunInput - - -RunInput - - -run_id - - -UUID (PK) - - -s3_object_id - - -TEXT (PK) - - -data_transfer_id - - -UUID () + + +RunInput + + +run_id + + +UUID (PK) + + +s3_object_id + + +TEXT (PK) + + +data_transfer_id + + +UUID () @@ -306,37 +360,37 @@ WorkflowRun->RunInput - - + + -inputs +inputs RunOutput - - -RunOutput - - -run_id - - -UUID (PK) - - -s3_object_id - - -TEXT (PK) - - -data_transfer_id - - -UUID () + + +RunOutput + + +run_id + + +UUID (PK) + + +s3_object_id + + +TEXT (PK) + + +data_transfer_id + + +UUID () @@ -344,91 +398,85 @@ WorkflowRun->RunOutput - - + + -outputs +outputs DataTransfer - - -DataTransfer - - -id - - -UUID (PK) - - -workflow_run_id - - -UUID () - - -direction - - -VARCHAR(10) () - - -provider - - -TEXT () - - -source_location - - -TEXT () - - -destination_location - - -TEXT () - - -transfer_id - - -TEXT () - - -status - - -VARCHAR(20) () - - -provider_metadata - - -JSON () - - -created_at - - -DATETIME () - - -updated_at - - -DATETIME () - - -error_message - - -TEXT () + + +DataTransfer + + +id + + +UUID (PK) + + +workflow_run_id + + +UUID () + + +direction + + +VARCHAR(10) () + + +provider + + +TEXT () + + +source_location + + +TEXT () + + +destination_location + + +TEXT () + + +transfer_id + + +TEXT () + + +status + + +VARCHAR(20) () + + +created_at + + +DATETIME () + + +updated_at + + +DATETIME () + + +error_message + + +TEXT () @@ -436,83 +484,83 @@ WorkflowRun->DataTransfer - - + + -data_transfers +data_transfers Workflow->WorkflowRun - - + + -runs +runs RunMetric->WorkflowRun - - + + -run +run RunInput->WorkflowRun - - + + -run +run RunInput->DataTransfer - - + + -data_transfer +data_transfer S3Object - - -S3Object - - -object_key - - -TEXT (PK) - - -URI - - -TEXT () - - -version_id - - -TEXT () - - -size_bytes - - -BIGINT () + + +S3Object + + +object_key + + +TEXT (PK) + + +URI + + +TEXT () + + +version_id + + +TEXT () + + +size_bytes + + +BIGINT () @@ -520,165 +568,165 @@ RunInput->S3Object - - + + -s3_object +s3_object RunOutput->WorkflowRun - - + + -run +run RunOutput->DataTransfer - - + + -data_transfer +data_transfer RunOutput->S3Object - - + + -s3_object +s3_object DataTransfer->WorkflowRun - - + + -workflow_run +workflow_run DataTransfer->RunInput - - + + -run_input +run_input DataTransfer->RunOutput - - + + -run_output +run_output S3Object->RunInput - - + + -run_inputs +run_inputs S3Object->RunOutput - - + + -run_outputs +run_outputs QueuedJob - - -QueuedJob - - -id - - -UUID (PK) - - -workflow_run_id - - -UUID () - - -workflow_id - - -UUID () - - -launch_payload - - -JSON () - - -status - - -VARCHAR(20) () - - -attempts - - -INTEGER () - - -queued_at - - -DATETIME () - - -last_attempt_at - - -DATETIME () - - -next_attempt_at - - -DATETIME () - - -submitted_at - - -DATETIME () - - -error - - -TEXT () + + +QueuedJob + + +id + + +UUID (PK) + + +workflow_run_id + + +UUID () + + +workflow_id + + +UUID () + + +launch_payload + + +JSON () + + +status + + +VARCHAR(20) () + + +attempts + + +INTEGER () + + +queued_at + + +DATETIME () + + +last_attempt_at + + +DATETIME () + + +next_attempt_at + + +DATETIME () + + +submitted_at + + +DATETIME () + + +error + + +TEXT () @@ -686,59 +734,59 @@ QueuedJob->WorkflowRun - - + + -workflow_run +workflow_run QueuedJob->Workflow - - + + -workflow +workflow SystemStatusCache - - -SystemStatusCache - - -key - - -TEXT (PK) - - -payload - - -JSON () - - -checked_at - - -DATETIME () - - -expires_at - - -DATETIME () - - -updated_at - - -DATETIME () + + +SystemStatusCache + + +key + + +TEXT (PK) + + +payload + + +JSON () + + +checked_at + + +DATETIME () + + +expires_at + + +DATETIME () + + +updated_at + + +DATETIME () @@ -746,45 +794,45 @@ SystemStatusIncident - - -SystemStatusIncident - - -id - - -INTEGER (PK) - - -component - - -TEXT () - - -status - - -TEXT () - - -started_at - - -DATETIME () - - -ended_at - - -DATETIME () - - -message - - -TEXT () + + +SystemStatusIncident + + +id + + +INTEGER (PK) + + +component + + +TEXT () + + +status + + +TEXT () + + +started_at + + +DATETIME () + + +ended_at + + +DATETIME () + + +message + + +TEXT () diff --git a/pyproject.toml b/pyproject.toml index e81cb096..69609870 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ dependencies = [ "rdkit~=2026.3", "pydantic-settings~=2.15", "globus-sdk>=4.9.0", + "pygithub>=2.10.0", ] [tool.setuptools] diff --git a/tests/conftest.py b/tests/conftest.py index ec234dcd..9f3545c1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,6 +15,7 @@ AdminSettings, AuthSettings, AwsSettings, + GithubSettings, GlobusSettings, SeqeraSettings, Settings, @@ -321,6 +322,10 @@ class GlobusSettingsNoEnv(GlobusSettings): model_config = {**GlobusSettings.model_config, "env_file": None} +class GithubSettingsNoEnv(GithubSettings): + model_config = {**GithubSettings.model_config, "env_file": None} + + class SettingsNoEnv(Settings): """ Settings class that ignores any .env files for testing @@ -332,6 +337,7 @@ class SettingsNoEnv(Settings): admin: AdminSettings = Field(default_factory=AdminSettingsNoEnv) auth: AuthSettings = Field(default_factory=AuthSettingsNoEnv) globus: GlobusSettings = Field(default_factory=GlobusSettingsNoEnv) + github: GithubSettings = Field(default_factory=GithubSettingsNoEnv) @pytest.fixture @@ -347,6 +353,24 @@ def override_settings(mock_settings): fastapi_app.dependency_overrides.clear() +@pytest.fixture(autouse=True) +def mock_repo_staging(mocker): + """launch_workflow resolves+stages the workflow's GitHub repo on every call + (workflow_repo_staging.ensure_repo_staging_requested), which makes a real + GitHub API request - stub it out by default so tests don't hit the network. + Tests that specifically exercise repo staging behavior can override this + with their own patch of the same target.""" + from app.services.workflow_repo_staging import RepoStagingLocations + + return mocker.patch( + "app.routes.workflows.ensure_repo_staging_requested", + return_value=RepoStagingLocations( + gadi_path="/staged/workflow-repo/path", + assets_gadi_path="/staged/workflow-repo/assets", + ), + ) + + @pytest.fixture def test_get_settings(): """ diff --git a/tests/db/test_db_admin.py b/tests/db/test_db_admin.py index bc7830f2..31a06558 100644 --- a/tests/db/test_db_admin.py +++ b/tests/db/test_db_admin.py @@ -78,7 +78,7 @@ def test_mount_db_admin_mounts_both_when_enabled(mocker, mock_settings): mock_settings.enable_db_admin = True mount_db_admin(app, mock_settings) - mount_admin.assert_called_once_with(app) + mount_admin.assert_called_once_with(app, mock_settings) mount_debug.assert_called_once_with(app) diff --git a/tests/scheduler/test_run_scheduler.py b/tests/scheduler/test_run_scheduler.py index d05b37f7..655d8e02 100644 --- a/tests/scheduler/test_run_scheduler.py +++ b/tests/scheduler/test_run_scheduler.py @@ -46,10 +46,17 @@ def test_main_adds_expected_jobs_and_starts_scheduler(monkeypatch): "submit_pending_jobs", "sync_completed_workflow_runs", "sync_data_transfers", + "sync_workflow_repo_staging", "refresh_user_credits", ] - submit_job, sync_job, data_transfer_job, refresh_job = scheduler.added_jobs + ( + submit_job, + sync_job, + data_transfer_job, + repo_staging_job, + refresh_job, + ) = scheduler.added_jobs submit_func, submit_config = submit_job assert submit_func is run_scheduler.submit_pending_jobs assert submit_config["kwargs"] == {"dry_run": True} @@ -68,6 +75,12 @@ def test_main_adds_expected_jobs_and_starts_scheduler(monkeypatch): assert data_transfer_config["jobstore"] == "memory" assert data_transfer_config["trigger"] is run_scheduler.DATA_TRANSFER_SYNC_INTERVAL + repo_staging_func, repo_staging_config = repo_staging_job + assert repo_staging_func is run_scheduler.sync_workflow_repo_staging + assert repo_staging_config["kwargs"] == {"dry_run": True} + assert repo_staging_config["jobstore"] == "memory" + assert repo_staging_config["trigger"] is run_scheduler.REPO_STAGING_SYNC_INTERVAL + refresh_func, refresh_config = refresh_job assert refresh_func is run_scheduler.refresh_user_credits assert refresh_config["jobstore"] == "db" @@ -77,6 +90,7 @@ def test_main_adds_expected_jobs_and_starts_scheduler(monkeypatch): ("add_job", "submit_pending_jobs"), ("add_job", "sync_completed_workflow_runs"), ("add_job", "sync_data_transfers"), + ("add_job", "sync_workflow_repo_staging"), ("add_job", "refresh_user_credits"), ("start", None), ("shutdown", None), diff --git a/tests/test_routes_workflows.py b/tests/test_routes_workflows.py index 32066817..41c44969 100644 --- a/tests/test_routes_workflows.py +++ b/tests/test_routes_workflows.py @@ -148,7 +148,7 @@ def test_launch_success_without_dataset( assert "submitTime" in data launch_form_arg = mock_prepare.call_args.args[0] assert launch_form_arg.tool == "bindcraft" - assert mock_prepare.call_args.kwargs["pipeline"] == "https://github.com/test/repo" + assert mock_prepare.call_args.kwargs["pipeline"] == "file:/staged/workflow-repo/path" assert mock_prepare.call_args.kwargs["revision"] == "dev" assert mock_prepare.call_args.kwargs["output_id"] == data["runId"] @@ -216,6 +216,49 @@ def test_launch_success_without_dataset( ) +@patch("app.routes.workflows.upload_csv_to_s3") +@patch("app.routes.workflows.read_csv_from_s3") +@patch("app.routes.workflows.prepare_bindflow_workflow", side_effect=_queue_job_for_route_prepare) +def test_launch_bindcraft_fills_in_default_settings_assets_in_samplesheet( + mock_prepare, mock_read_csv, mock_upload_csv, client: TestClient +): + """settings_filters/settings_advanced samplesheet columns are left empty + by the frontend (see sbp-portal's de-novo-design.ts) - confirmed in + production via a real staged samplesheet. _rewrite_bindflow_settings_ + asset_columns must fill them in with the local Gadi path to bindflow's + bundled default JSON files before the samplesheet is staged to Gadi.""" + mock_read_csv.return_value = [ + { + "starting_pdb": "s3://test-bucket/pdb/target.pdb", + "settings_filters": "", + "settings_advanced": "", + } + ] + mock_upload_csv.return_value = S3UploadResult( + success=True, file_key="inputs/samplesheets/corrected.csv", bucket="test-bucket" + ) + + payload = { + "launch": {"workflow": "de-novo-design", "tool": "bindcraft", "runName": "test-run"}, + "s3InputKey": "inputs/samplesheets/test.csv", + "formData": {"workflow": "de-novo-design", "tool": "bindcraft"}, + } + + response = client.post("/api/workflows/launch", json=payload) + + assert response.status_code == 201 + # First call corrects starting_pdb (_stage_referenced_samplesheet_file), + # second call rewrites the settings_* columns - both re-upload the row. + assert mock_upload_csv.call_count == 2 + rewritten_row = mock_upload_csv.call_args_list[1].args[0] + assert rewritten_row["settings_filters"] == ( + "/staged/workflow-repo/assets/assets/bindcraft/default_filters.json" + ) + assert rewritten_row["settings_advanced"] == ( + "/staged/workflow-repo/assets/assets/bindcraft/default_4stage_multimer.json" + ) + + @patch("app.routes.workflows.upload_csv_to_s3") @patch("app.routes.workflows.read_csv_from_s3") @patch("app.routes.workflows.prepare_bindflow_workflow") @@ -351,9 +394,7 @@ def test_launch_de_novo_design_rfdiffusion_routes_to_proteindj( assert data["status"] == "staging" mock_prepare_proteindj.assert_called_once() mock_prepare_bindflow.assert_not_called() - assert ( - mock_prepare_proteindj.call_args.kwargs["pipeline"] == "https://github.com/test/proteindj" - ) + assert mock_prepare_proteindj.call_args.kwargs["pipeline"] == "file:/staged/workflow-repo/path" assert mock_prepare_proteindj.call_args.kwargs["output_id"] == data["runId"] # rfdiffusion's s3InputKey is the starting PDB's own URI (no samplesheet exists # for it), and prepare_proteindj_workflow stages that file itself - so the @@ -704,7 +745,7 @@ def test_launch_proteinfold_success( assert data["status"] == "staging" run_id = UUID(data["runId"]) mock_prepare.assert_called_once() - assert mock_prepare.call_args.kwargs["pipeline"] == "https://github.com/nf-core/proteinfold" + assert mock_prepare.call_args.kwargs["pipeline"] == "file:/staged/workflow-repo/path" assert mock_prepare.call_args.kwargs["revision"] == "dev" assert mock_prepare.call_args.kwargs["output_id"] == str(run_id) with Session(test_engine) as db: @@ -1059,7 +1100,7 @@ def test_launch_interaction_screening_success(mock_prepare, wisps_client: TestCl call_kwargs = mock_prepare.call_args.kwargs assert call_kwargs["form_data"].fastaS3Uri == "s3://bucket/test.fasta" assert call_kwargs["form_data"].splitOutputDir == "/data/split" - assert call_kwargs["pipeline"] == "https://github.com/test/wisps" + assert call_kwargs["pipeline"] == "file:/staged/workflow-repo/path" assert call_kwargs["revision"] in {"dev", "main"} assert call_kwargs["output_id"] == str(run_id) diff --git a/tests/test_services_globus_transfer.py b/tests/test_services_globus_transfer.py index 763ece95..2af6c656 100644 --- a/tests/test_services_globus_transfer.py +++ b/tests/test_services_globus_transfer.py @@ -21,7 +21,23 @@ submit_pending_transfer, sync_data_transfers, ) -from tests.datagen import DataTransferFactory, QueuedJobFactory, WorkflowRunFactory +from tests.datagen import ( + DataTransferFactory, + QueuedJobFactory, + WorkflowFactory, + WorkflowRunFactory, +) + + +def _workflow_run_without_repo_staging(): + """A WorkflowRun whose Workflow has no repo-staging requirement (None) - + QueuedJobFactory/WorkflowRunFactory don't create a real related Workflow + row by default (__set_relationships__ = False), and _try_promote_staging_job + now dereferences queued_job.workflow, so tests exercising only the + input-transfer gate need an explicit, real Workflow with + repo_staging_status=None (not a random value) so it acts as a no-op gate.""" + workflow = WorkflowFactory.create_sync(repo_staging_status=None) + return WorkflowRunFactory.create_sync(workflow=workflow) def _globus_api_error(status_code: int, json_body: dict) -> Exception: @@ -140,7 +156,7 @@ def test_submit_pending_transfer_success(test_db, persistent_models, mock_transf mock_transfer_client.get_submission_id.return_value = {"value": "sub-123"} mock_transfer_client.submit_transfer.return_value = {"task_id": "task-abc"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -173,7 +189,7 @@ def test_submit_pending_transfer_reuses_existing_submission_id( not mint a fresh submission_id on retry - Globus dedupes on this id.""" mock_transfer_client.submit_transfer.return_value = {"task_id": "task-abc"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -201,7 +217,7 @@ def test_submit_pending_transfer_api_error_marks_failed( 400, {"code": "UNKNOWN_SCOPE_ERROR", "message": "requested unknown scopes"} ) - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -225,7 +241,7 @@ def test_submit_pending_transfer_api_error_marks_failed( def test_poll_transfer_no_transfer_id_raises(test_db, persistent_models): - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", transfer_id=None ) @@ -239,7 +255,7 @@ def test_poll_transfer_api_error_records_message_without_changing_status( mock_transfer_client.get_task.side_effect = _globus_api_error( 401, {"code": "AuthenticationFailed", "message": "token expired"} ) - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", transfer_id="task-abc" ) @@ -255,7 +271,7 @@ def test_poll_transfer_api_error_records_message_without_changing_status( def test_poll_transfer_succeeded_marks_completed(test_db, persistent_models, mock_transfer_client): mock_transfer_client.get_task.return_value = {"status": "SUCCEEDED"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", transfer_id="task-abc" ) @@ -270,7 +286,7 @@ def test_poll_transfer_failed_records_fatal_error(test_db, persistent_models, mo "status": "FAILED", "fatal_error": {"description": "no such file"}, } - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", transfer_id="task-abc" ) @@ -283,7 +299,7 @@ def test_poll_transfer_failed_records_fatal_error(test_db, persistent_models, mo def test_poll_transfer_active_stays_in_progress(test_db, persistent_models, mock_transfer_client): mock_transfer_client.get_task.return_value = {"status": "ACTIVE"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", transfer_id="task-abc" ) @@ -297,7 +313,7 @@ def test_poll_transfer_inactive_recent_stays_in_progress( test_db, persistent_models, mock_transfer_client ): mock_transfer_client.get_task.return_value = {"status": "INACTIVE"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, status="in_progress", @@ -316,7 +332,7 @@ def test_poll_transfer_inactive_stale_marks_failed( test_db, persistent_models, mock_transfer_client ): mock_transfer_client.get_task.return_value = {"status": "INACTIVE"} - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() stale_created_at = datetime.now(UTC) - STALE_TRANSFER_TIMEOUT - timedelta(minutes=1) data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, @@ -337,8 +353,10 @@ def test_poll_transfer_inactive_stale_marks_failed( def test_notify_launcher_ignores_output_direction(test_db, persistent_models): - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="output", status="completed" ) @@ -350,8 +368,10 @@ def test_notify_launcher_ignores_output_direction(test_db, persistent_models): def test_notify_launcher_ignores_non_staging_queued_job(test_db, persistent_models): - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="pending") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="pending" + ) data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", status="completed" ) @@ -363,8 +383,10 @@ def test_notify_launcher_ignores_non_staging_queued_job(test_db, persistent_mode def test_notify_launcher_failed_transfer_fails_queued_job(test_db, persistent_models): - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) data_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -382,8 +404,10 @@ def test_notify_launcher_failed_transfer_fails_queued_job(test_db, persistent_mo def test_notify_launcher_waits_for_all_input_transfers(test_db, persistent_models): """A run with two input transfers (samplesheet + pdb) must not launch until both have completed.""" - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) completed_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", status="completed" ) @@ -400,8 +424,10 @@ def test_notify_launcher_waits_for_all_input_transfers(test_db, persistent_model def test_notify_launcher_flips_to_pending_once_all_input_transfers_complete( test_db, persistent_models ): - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) first_transfer = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", status="completed" ) @@ -425,8 +451,10 @@ def test_sync_data_transfers_submits_and_notifies(test_db, persistent_models, mo mock_transfer_client.get_submission_id.return_value = {"value": "sub-1"} mock_transfer_client.submit_transfer.return_value = {"task_id": "task-1"} - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -449,8 +477,10 @@ def test_sync_data_transfers_submits_and_notifies(test_db, persistent_models, mo def test_sync_data_transfers_polls_and_completes(test_db, persistent_models, mock_transfer_client): mock_transfer_client.get_task.return_value = {"status": "SUCCEEDED"} - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -470,7 +500,7 @@ def test_sync_data_transfers_polls_and_completes(test_db, persistent_models, moc def test_sync_data_transfers_ignores_non_globus_provider( test_db, persistent_models, mock_transfer_client ): - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() DataTransferFactory.create_sync( workflow_run=workflow_run, provider="s3", @@ -495,8 +525,10 @@ def test_sync_data_transfers_counts_soft_submission_failure( 400, {"code": "UNKNOWN_SCOPE_ERROR", "message": "requested unknown scopes"} ) - workflow_run = WorkflowRunFactory.create_sync() - QueuedJobFactory.create_sync(workflow_run=workflow_run, status="staging") + workflow_run = _workflow_run_without_repo_staging() + QueuedJobFactory.create_sync( + workflow=workflow_run.workflow, workflow_run=workflow_run, status="staging" + ) DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -522,7 +554,7 @@ def test_sync_data_transfers_continues_after_unexpected_error( ): mock_transfer_client.get_task.side_effect = RuntimeError("boom") - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", @@ -550,7 +582,7 @@ def get_task(transfer_id): mock_transfer_client.get_task.side_effect = get_task - workflow_run = WorkflowRunFactory.create_sync() + workflow_run = _workflow_run_without_repo_staging() broken = DataTransferFactory.create_sync( workflow_run=workflow_run, direction="input", diff --git a/tests/test_services_seqera.py b/tests/test_services_seqera.py index db853904..e03813b6 100644 --- a/tests/test_services_seqera.py +++ b/tests/test_services_seqera.py @@ -152,8 +152,10 @@ async def test_prepare_bindflow_workflow_writes_expected_queued_job( config_path=_CONFIG_PATH, revision="main", output_id="run-output-id", + form_data=_empty_form_data(), user_details=_USER_DETAILS, staged_input_location="/test/input/de-novo-design/run-id/test.csv", + repo_assets_path="/test/workflow_repos/test-repo/abc123", ) queued_job = test_db.scalar( @@ -184,6 +186,105 @@ async def test_prepare_bindflow_workflow_writes_expected_queued_job( assert "custom_param: value" in queued_job.launch_payload["paramsText"] +@pytest.mark.asyncio +async def test_prepare_bindflow_workflow_fills_in_default_when_settings_unset( + test_db, persistent_models, mock_settings +): + """The frontend no longer sends any value for settings_filters/ + settings_advanced (see sbp-portal's de-novo-design.ts) - an empty/missing + value must still resolve to the bindflow repo's bundled default file, + not be left blank.""" + user = AppUserFactory.create_sync() + workflow = WorkflowFactory.create_sync() + workflow_run = WorkflowRunFactory.create_sync(workflow=workflow, owner=user) + + form = WorkflowLaunchForm(workflow="de-novo-design", tool="bindcraft", runName="run-1") + + with ( + patch("app.services.bindflow_executor.get_bindflow_config_profiles", return_value=["gadi"]), + patch( + "app.services.bindflow_executor.get_bindflow_config_text", return_value="config_text" + ), + ): + await prepare_bindflow_workflow( + form=form, + settings=mock_settings, + db_session=test_db, + workflow_run=workflow_run, + pipeline="file:/g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git", + config_path=_CONFIG_PATH, + revision="dev", + output_id="run-output-id", + form_data=_empty_form_data(), + user_details=_USER_DETAILS, + staged_input_location="/test/input/de-novo-design/run-id/test.csv", + repo_assets_path="/g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git", + ) + + queued_job = test_db.scalar( + select(QueuedJob).where(QueuedJob.workflow_run_id == workflow_run.id) + ) + assert queued_job is not None + params_text = queued_job.launch_payload["paramsText"] + assert ( + "settings_filters: /g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git/" + "assets/bindcraft/default_filters.json" in params_text + ) + assert ( + "settings_advanced: /g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git/" + "assets/bindcraft/default_4stage_multimer.json" in params_text + ) + + +@pytest.mark.asyncio +async def test_prepare_bindflow_workflow_passes_through_custom_settings_value( + test_db, persistent_models, mock_settings +): + """A value that doesn't match the known default bindflow-repo URL is left + unchanged - there's no support for staging arbitrary user-supplied + settings files yet, so it's passed through as-is rather than guessed at.""" + user = AppUserFactory.create_sync() + workflow = WorkflowFactory.create_sync() + workflow_run = WorkflowRunFactory.create_sync(workflow=workflow, owner=user) + + form = WorkflowLaunchForm(workflow="de-novo-design", tool="bindcraft", runName="run-1") + form_data = WorkflowFormData( + workflow="de-novo-design", + tool="bindcraft", + settings_filters="https://example.com/my-custom-filters.json", + settings_advanced="https://example.com/my-custom-advanced.json", + ) + + with ( + patch("app.services.bindflow_executor.get_bindflow_config_profiles", return_value=["gadi"]), + patch( + "app.services.bindflow_executor.get_bindflow_config_text", return_value="config_text" + ), + ): + await prepare_bindflow_workflow( + form=form, + settings=mock_settings, + db_session=test_db, + workflow_run=workflow_run, + pipeline="file:/g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git", + config_path=_CONFIG_PATH, + revision="dev", + output_id="run-output-id", + form_data=form_data, + user_details=_USER_DETAILS, + staged_input_location="/test/input/de-novo-design/run-id/test.csv", + repo_assets_path="/g/data/yz52/sbp_data/workflow_repos/x-bindflow/abc123.git", + ) + + queued_job = test_db.scalar( + select(QueuedJob).where(QueuedJob.workflow_run_id == workflow_run.id) + ) + assert queued_job is not None + params_text = queued_job.launch_payload["paramsText"] + assert "settings_filters: https://example.com/my-custom-filters.json" in params_text + assert "settings_advanced: https://example.com/my-custom-advanced.json" in params_text + + @pytest.mark.asyncio @respx.mock async def test_launch_success_with_all_params(persistent_models): diff --git a/tests/test_services_workflow_repo_staging.py b/tests/test_services_workflow_repo_staging.py new file mode 100644 index 00000000..8f757764 --- /dev/null +++ b/tests/test_services_workflow_repo_staging.py @@ -0,0 +1,680 @@ +"""Tests for staging a workflow's GitHub repo onto Gadi via S3 + Globus.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import requests +from github import GithubException + +from app.config import GlobusSettings +from app.services.globus_errors import GlobusTransferError +from app.services.workflow_repo_staging import ( + RepoStagingError, + build_repo_gadi_path, + build_repo_s3_prefix, + ensure_repo_staging_requested, + parse_github_repo, + poll_repo_staging, + resolve_latest_commit_sha, + stage_pending_repo, + sync_workflow_repo_staging, +) +from tests.datagen import QueuedJobFactory, WorkflowFactory + + +def _mock_github_client(*, sha: str | None = None, raises: Exception | None = None) -> MagicMock: + """PyGithub client double - resolve_latest_commit_sha only ever calls + client.get_repo(...).get_commit(...).sha, so that's all this needs to + fake. PyGithub uses `requests` internally (not httpx), so respx can't + intercept its calls - mocking at this boundary is simpler than pulling in + another HTTP-mocking library just for these few tests.""" + client = MagicMock() + if raises is not None: + client.get_repo.return_value.get_commit.side_effect = raises + else: + commit = MagicMock() + commit.sha = sha + client.get_repo.return_value.get_commit.return_value = commit + return client + + +def _globus_api_error(status_code: int, json_body: dict) -> Exception: + """Same helper as test_services_globus_transfer.py - builds a real + globus_sdk.TransferAPIError, which needs a full requests.Response.""" + import globus_sdk + + response = requests.Response() + response.status_code = status_code + response._content = json.dumps(json_body).encode("utf-8") + response.headers["Content-Type"] = "application/json" + response.request = requests.PreparedRequest() + response.request.prepare(method="GET", url="https://transfer.api.globus.org/v0.10/x") + return globus_sdk.TransferAPIError(response) + + +def _make_local_git_repo(tmp_path: Path, files: dict[str, str]) -> tuple[str, str]: + """Create a real local git repo (used as a stand-in "GitHub" remote so + _clone_and_upload_repo's git commands run for real, not mocked) and return + (repo_path, commit_sha).""" + repo_dir = tmp_path / "origin" + repo_dir.mkdir() + subprocess.run(["git", "init", "-q", "-b", "main"], cwd=repo_dir, check=True) + subprocess.run(["git", "config", "user.email", "test@test.com"], cwd=repo_dir, check=True) + subprocess.run(["git", "config", "user.name", "test"], cwd=repo_dir, check=True) + for relative_path, content in files.items(): + file_path = repo_dir / relative_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content) + subprocess.run(["git", "add", relative_path], cwd=repo_dir, check=True) + subprocess.run(["git", "commit", "-q", "-m", "init"], cwd=repo_dir, check=True) + commit_sha = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo_dir, check=True, capture_output=True, text=True + ).stdout.strip() + return str(repo_dir), commit_sha + + +@pytest.fixture +def mock_transfer_client(): + client = MagicMock() + with patch("app.services.workflow_repo_staging.get_transfer_client", return_value=client): + yield client + + +@pytest.fixture +def globus_settings(): + return GlobusSettings( + client_id="test-globus-client-id", + client_secret="test-globus-client-secret", + gadi_collection_id="test-gadi-collection-id", + s3_collection_id="test-s3-collection-id", + gadi_collection_root="/test", + input_dir="/test/input", + output_dir="/test/output", + ) + + +# ============================================================================ +# parse_github_repo / build_repo_* path helpers +# ============================================================================ + + +def test_parse_github_repo_extracts_owner_and_repo(): + assert parse_github_repo("https://github.com/nf-core/proteinfold") == ( + "nf-core", + "proteinfold", + ) + + +def test_parse_github_repo_strips_git_suffix(): + assert parse_github_repo("https://github.com/nf-core/proteinfold.git") == ( + "nf-core", + "proteinfold", + ) + + +def test_parse_github_repo_rejects_non_github_host(): + with pytest.raises(RepoStagingError, match="Only github.com"): + parse_github_repo("https://gitlab.com/nf-core/proteinfold") + + +def test_parse_github_repo_rejects_missing_repo(): + with pytest.raises(RepoStagingError, match="Could not parse"): + parse_github_repo("https://github.com/nf-core") + + +def test_build_repo_s3_prefix(): + assert ( + build_repo_s3_prefix("nf-core", "proteinfold", "abc123") + == "workflow-repos/nf-core-proteinfold/abc123" + ) + + +def test_build_repo_gadi_path(globus_settings): + path = build_repo_gadi_path("nf-core", "proteinfold", "abc123", globus_settings=globus_settings) + assert path == "/test/workflow_repos/nf-core-proteinfold/abc123.git" + + +# ============================================================================ +# resolve_latest_commit_sha +# ============================================================================ + + +def test_resolve_latest_commit_sha_success(mock_settings): + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha="abc123"), + ): + assert ( + resolve_latest_commit_sha( + "https://github.com/nf-core/proteinfold", "dev", settings=mock_settings + ) + == "abc123" + ) + + +def test_resolve_latest_commit_sha_http_error_raises(mock_settings): + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(raises=GithubException(404, {"message": "Not Found"}, {})), + ): + with pytest.raises(RepoStagingError, match="Failed to resolve commit"): + resolve_latest_commit_sha( + "https://github.com/nf-core/proteinfold", "dev", settings=mock_settings + ) + + +def test_resolve_latest_commit_sha_missing_sha_raises(mock_settings): + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha=None), + ): + with pytest.raises(RepoStagingError, match="no commit sha"): + resolve_latest_commit_sha( + "https://github.com/nf-core/proteinfold", "dev", settings=mock_settings + ) + + +# ============================================================================ +# ensure_repo_staging_requested +# ============================================================================ + + +def test_ensure_repo_staging_requested_marks_pending_on_cache_miss( + test_db, persistent_models, mock_settings +): + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha=None, + repo_staging_status=None, + ) + + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha="newsha"), + ): + locations = ensure_repo_staging_requested(test_db, workflow, settings=mock_settings) + + assert locations.gadi_path == workflow.repo_gadi_path + assert locations.assets_gadi_path == "/test/workflow_repos/test-repo/newsha" + assert workflow.repo_staged_commit_sha == "newsha" + assert workflow.repo_staging_status == "pending" + + +def test_ensure_repo_staging_requested_resets_on_commit_change( + test_db, persistent_models, mock_settings +): + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="oldsha", + repo_staging_status="completed", + repo_gadi_path="/test/workflow_repos/test-repo/oldsha", + repo_staging_transfer_id="old-task-id", + repo_staging_error_message=None, + ) + + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha="newsha"), + ): + ensure_repo_staging_requested(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staged_commit_sha == "newsha" + assert workflow.repo_staging_status == "pending" + assert workflow.repo_staging_transfer_id is None + + +def test_ensure_repo_staging_requested_reuses_cache_hit(test_db, persistent_models, mock_settings): + """Same commit, already completed - must not reset back to pending (that + would re-trigger staging for no reason).""" + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="samesha", + repo_staging_status="completed", + repo_gadi_path="/test/workflow_repos/test-repo/samesha.git", + ) + + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha="samesha"), + ): + locations = ensure_repo_staging_requested(test_db, workflow, settings=mock_settings) + + assert locations.gadi_path == "/test/workflow_repos/test-repo/samesha.git" + assert locations.assets_gadi_path == "/test/workflow_repos/test-repo/samesha" + assert workflow.repo_staging_status == "completed" + + +def test_ensure_repo_staging_requested_retries_after_failure( + test_db, persistent_models, mock_settings +): + """Same commit but previously failed - must re-request staging, not treat + the failure as a permanent cache entry.""" + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="samesha", + repo_staging_status="failed", + repo_staging_error_message="boom", + ) + + with patch( + "app.services.workflow_repo_staging._get_github_client", + return_value=_mock_github_client(sha="samesha"), + ): + ensure_repo_staging_requested(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "pending" + assert workflow.repo_staging_error_message is None + + +# ============================================================================ +# stage_pending_repo +# ============================================================================ + + +def test_stage_pending_repo_uses_collection_relative_destination_path( + test_db, persistent_models, mock_transfer_client, mock_settings +): + """Regression test: add_item must receive the Gadi path relative to the + collection root, not the raw absolute Gadi filesystem path - sending the + absolute path double-nests it under the collection root on the real + filesystem (confirmed in production: the staged repo was unreachable at + its expected path).""" + mock_transfer_client.get_submission_id.return_value = {"value": "sub-1"} + mock_transfer_client.submit_transfer.return_value = {"task_id": "task-1"} + + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="abc123", + repo_staging_status="pending", + repo_gadi_path="/test/workflow_repos/test-repo/abc123", + repo_staging_transfer_id=None, + ) + + with patch("app.services.workflow_repo_staging._clone_and_upload_repo") as mock_clone: + stage_pending_repo(test_db, workflow, settings=mock_settings) + + mock_clone.assert_called_once() + submitted = mock_transfer_client.submit_transfer.call_args[0][0] + assert submitted["source_endpoint"] == "test-s3-collection-id" + assert submitted["destination_endpoint"] == "test-gadi-collection-id" + # NOT "/test/workflow_repos/test-repo/abc123" - that's the absolute path, + # which would double-nest under the "/test" collection root on Gadi. + assert submitted["DATA"][0]["destination_path"] == "/workflow_repos/test-repo/abc123" + assert submitted["DATA"][0]["source_path"] == "/workflow-repos/test-repo/abc123" + assert submitted["DATA"][0]["recursive"] is True + # Second item: the plain-checkout companion for pipeline-bundled assets + # (see build_repo_assets_gadi_path) - a sibling directory, no ".git" suffix. + assert submitted["DATA"][1]["destination_path"] == "/workflow_repos/test-repo/abc123" + assert submitted["DATA"][1]["source_path"] == "/workflow-repos-assets/test-repo/abc123" + assert submitted["DATA"][1]["recursive"] is True + + assert workflow.repo_staging_status == "in_progress" + assert workflow.repo_staging_transfer_id == "task-1" + + +def test_stage_pending_repo_no_commit_sha_raises(test_db, persistent_models, mock_settings): + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + repo_staged_commit_sha=None, + ) + with pytest.raises(RepoStagingError, match="no commit sha"): + stage_pending_repo(test_db, workflow, settings=mock_settings) + + +def test_stage_pending_repo_download_failure_marks_failed( + test_db, persistent_models, mock_transfer_client, mock_settings +): + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="abc123", + repo_staging_status="pending", + ) + + with patch( + "app.services.workflow_repo_staging._clone_and_upload_repo", + side_effect=RepoStagingError("clone failed"), + ): + stage_pending_repo(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "failed" + assert "clone failed" in workflow.repo_staging_error_message + mock_transfer_client.submit_transfer.assert_not_called() + + +def test_stage_pending_repo_submission_api_error_marks_failed( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_submission_id.return_value = {"value": "sub-1"} + mock_transfer_client.submit_transfer.side_effect = _globus_api_error( + 400, {"code": "UNKNOWN_SCOPE_ERROR", "message": "bad scope"} + ) + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="abc123", + repo_staging_status="pending", + repo_gadi_path="/test/workflow_repos/test-repo/abc123", + ) + + with patch("app.services.workflow_repo_staging._clone_and_upload_repo"): + stage_pending_repo(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "failed" + assert "UNKNOWN_SCOPE_ERROR" in workflow.repo_staging_error_message + + +def test_stage_pending_repo_reuses_existing_submission_id( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.submit_transfer.return_value = {"task_id": "task-1"} + workflow = WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="abc123", + repo_staging_status="pending", + repo_gadi_path="/test/workflow_repos/test-repo/abc123", + repo_staging_transfer_id="already-committed-sub-id", + ) + + with patch("app.services.workflow_repo_staging._clone_and_upload_repo"): + stage_pending_repo(test_db, workflow, settings=mock_settings) + + mock_transfer_client.get_submission_id.assert_not_called() + submitted = mock_transfer_client.submit_transfer.call_args[0][0] + assert submitted["submission_id"] == "already-committed-sub-id" + + +# ============================================================================ +# _clone_and_upload_repo +# ============================================================================ + + +def test_clone_and_upload_repo_uploads_bare_repo_structure(tmp_path, mock_settings): + """The staged repo must be bare - HEAD/config/objects directly at the + prefix root, not nested under a .git/ subdirectory, and no plain working- + tree files like main.nf (see build_repo_gadi_path's docstring: Nextflow + resolves the ".git"-suffixed path as a git-dir directly, so a non-bare + checkout with a *nested* .git/ put those one level too deep to be found - + confirmed in production as "fatal: not a git repository: ''" even + though the nested .git was fully and correctly staged).""" + from app.services.workflow_repo_staging import _clone_and_upload_repo + + revision = "dev" + repo_path, commit_sha = _make_local_git_repo( + tmp_path, {"main.nf": "process {}", "nextflow.config": "params {}"} + ) + mock_s3_client = MagicMock() + prefix = "workflow-repos/test-repo/abc123" + # _clone_and_upload_repo's source files live under a TemporaryDirectory + # that's cleaned up before the function returns - copy each file's bytes + # out at upload time (mocked here in place of a real S3 PUT) so they can + # be reassembled and verified with a real `git show` after the fact. + staged_dir = tmp_path / "staged-verify" + + def _capture_upload(source_path: str, _bucket: str, key: str) -> None: + # _clone_and_upload_repo also uploads a plain-checkout companion + # under a different prefix (see test_clone_and_upload_repo_uploads_ + # plain_checkout_assets below) - ignore those keys here, this test + # only cares about the bare repo's structure. + if not key.startswith(f"{prefix}/"): + return + dest = staged_dir / key[len(prefix) + 1 :] + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(Path(source_path).read_bytes()) + + mock_s3_client.upload_file.side_effect = _capture_upload + + with patch("app.services.workflow_repo_staging.get_s3_client", return_value=mock_s3_client): + _clone_and_upload_repo( + "test", + "repo", + commit_sha, + repo_path, + prefix, + revision=revision, + settings=mock_settings, + ) + + uploaded_keys = {call.args[2] for call in mock_s3_client.upload_file.call_args_list} + assert f"{prefix}/HEAD" in uploaded_keys + assert f"{prefix}/config" in uploaded_keys + assert any(key.startswith(f"{prefix}/objects/") for key in uploaded_keys) + # No nested .git/ - and no plain working-tree files, since a bare repo has + # no working tree; main.nf only exists as a blob inside the packed objects. + assert not any(".git/" in key for key in uploaded_keys) + assert f"{prefix}/main.nf" not in uploaded_keys + # The branch ref must be a real loose file (not packed away) so it + # survives the S3/Globus round-trip, which only preserves actual files. + assert f"{prefix}/refs/heads/{revision}" in uploaded_keys + + # revision must actually resolve to the fetched commit and read back the + # real file content - this is what Nextflow's own checkout will do once + # it resolves the pipeline. + shown = subprocess.run( + ["git", f"--git-dir={staged_dir}", "show", f"{revision}:main.nf"], + capture_output=True, + text=True, + check=True, + ) + assert shown.stdout == "process {}" + + +def test_clone_and_upload_repo_uploads_plain_checkout_assets(tmp_path, mock_settings): + """Alongside the bare repo, a plain checkout of the same commit must be + uploaded under build_repo_assets_s3_prefix's prefix - this is what makes + pipeline-bundled asset files (e.g. bindcraft's default settings JSON, + see bindflow_executor.py) readable as plain files on Gadi, since the bare + repo has no working tree at all.""" + from app.services.workflow_repo_staging import ( + _clone_and_upload_repo, + build_repo_assets_s3_prefix, + ) + + repo_path, commit_sha = _make_local_git_repo( + tmp_path, + { + "main.nf": "process {}", + "assets/bindcraft/default_filters.json": '{"filter": true}', + }, + ) + mock_s3_client = MagicMock() + prefix = "workflow-repos/test-repo/abc123" + assets_prefix = build_repo_assets_s3_prefix("test", "repo", commit_sha) + # Source files live under a TemporaryDirectory cleaned up before the + # function returns - capture content at upload time, mocked here in + # place of a real S3 PUT. + uploaded_content: dict[str, bytes] = {} + + def _capture_upload(source_path: str, _bucket: str, key: str) -> None: + uploaded_content[key] = Path(source_path).read_bytes() + + mock_s3_client.upload_file.side_effect = _capture_upload + + with patch("app.services.workflow_repo_staging.get_s3_client", return_value=mock_s3_client): + _clone_and_upload_repo( + "test", + "repo", + commit_sha, + repo_path, + prefix, + revision="dev", + settings=mock_settings, + ) + + assert f"{assets_prefix}/main.nf" in uploaded_content + assert f"{assets_prefix}/assets/bindcraft/default_filters.json" in uploaded_content + # Plain files, readable directly - not blobs needing `git show` to extract. + assert uploaded_content[f"{assets_prefix}/main.nf"] == b"process {}" + assert ( + uploaded_content[f"{assets_prefix}/assets/bindcraft/default_filters.json"] + == b'{"filter": true}' + ) + # No .git internals in the plain checkout - that's what the bare repo + # (uploaded separately, under `prefix`) is for. + assert not any(key.startswith(f"{assets_prefix}/.git") for key in uploaded_content) + + +def test_clone_and_upload_repo_clone_failure_raises(tmp_path, mock_settings): + from app.services.workflow_repo_staging import _clone_and_upload_repo + + nonexistent_repo_path = str(tmp_path / "does-not-exist") + + with pytest.raises(RepoStagingError, match="Failed to clone"): + _clone_and_upload_repo( + "test", + "repo", + "abc123", + nonexistent_repo_path, + "workflow-repos/test-repo/abc123", + revision="dev", + settings=mock_settings, + ) + + +# ============================================================================ +# poll_repo_staging +# ============================================================================ + + +def test_poll_repo_staging_no_transfer_id_raises(test_db, persistent_models): + workflow = WorkflowFactory.create_sync(repo_staging_transfer_id=None) + with pytest.raises(GlobusTransferError, match="no transfer_id"): + poll_repo_staging(test_db, workflow) + + +def test_poll_repo_staging_succeeded_marks_completed_and_promotes_queued_jobs( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.return_value = {"status": "SUCCEEDED"} + workflow = WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + QueuedJobFactory.create_sync(workflow=workflow, status="staging") + + poll_repo_staging(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "completed" + + +def test_poll_repo_staging_failed_records_fatal_error( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.return_value = { + "status": "FAILED", + "fatal_error": {"description": "permission denied"}, + } + workflow = WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + + poll_repo_staging(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "failed" + assert workflow.repo_staging_error_message == "permission denied" + + +def test_poll_repo_staging_active_leaves_status_unchanged( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.return_value = {"status": "ACTIVE"} + workflow = WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + + poll_repo_staging(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "in_progress" + + +def test_poll_repo_staging_api_error_records_message_without_changing_status( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.side_effect = _globus_api_error( + 401, {"code": "AuthenticationFailed", "message": "token expired"} + ) + workflow = WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + + with pytest.raises(GlobusTransferError, match="Failed to poll"): + poll_repo_staging(test_db, workflow, settings=mock_settings) + + assert workflow.repo_staging_status == "in_progress" + assert "Poll failed" in workflow.repo_staging_error_message + + +# ============================================================================ +# sync_workflow_repo_staging +# ============================================================================ + + +def test_sync_workflow_repo_staging_submits_pending( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_submission_id.return_value = {"value": "sub-1"} + mock_transfer_client.submit_transfer.return_value = {"task_id": "task-1"} + WorkflowFactory.create_sync( + repo_url="https://github.com/test/repo", + default_revision="dev", + repo_staged_commit_sha="abc123", + repo_staging_status="pending", + repo_gadi_path="/test/workflow_repos/test-repo/abc123", + repo_staging_transfer_id=None, + ) + + with patch("app.services.workflow_repo_staging._clone_and_upload_repo"): + result = sync_workflow_repo_staging(test_db, settings=mock_settings) + + assert result.checked == 1 + assert result.submitted == 1 + + +def test_sync_workflow_repo_staging_polls_in_progress( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.return_value = {"status": "SUCCEEDED"} + WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + + result = sync_workflow_repo_staging(test_db, settings=mock_settings) + + assert result.checked == 1 + assert result.completed == 1 + + +def test_sync_workflow_repo_staging_ignores_workflows_not_staging( + test_db, persistent_models, mock_transfer_client, mock_settings +): + WorkflowFactory.create_sync(repo_staging_status="completed") + WorkflowFactory.create_sync(repo_staging_status=None) + + result = sync_workflow_repo_staging(test_db, settings=mock_settings) + + assert result.checked == 0 + mock_transfer_client.get_task.assert_not_called() + + +def test_sync_workflow_repo_staging_continues_after_unexpected_error( + test_db, persistent_models, mock_transfer_client, mock_settings +): + mock_transfer_client.get_task.side_effect = RuntimeError("boom") + WorkflowFactory.create_sync( + repo_staging_transfer_id="task-1", repo_staging_status="in_progress" + ) + + result = sync_workflow_repo_staging(test_db, settings=mock_settings) + + assert result.checked == 1 + assert result.errored == 1 diff --git a/uv.lock b/uv.lock index 877c8124..80da2213 100644 --- a/uv.lock +++ b/uv.lock @@ -1022,6 +1022,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/30/a4/2bffa9f8e804325a09867f0e9d30795c80ea9f8d62560bd1b6ad6220eb2f/pydantic_settings-2.15.0-py3-none-any.whl", hash = "sha256:0ba092c291c94baceb5eff768aa0d56400a457585bc0175925a5a5510303da42", size = 69413, upload-time = "2026-08-07T09:24:55.839Z" }, ] +[[package]] +name = "pygithub" +version = "2.10.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyjwt", extra = ["crypto"] }, + { name = "pynacl" }, + { name = "requests" }, + { name = "typing-extensions" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e1/9b/195603d5371861005a3467c5e4afd02fd0698795a2aa36dc41498b9d879d/pygithub-2.10.0.tar.gz", hash = "sha256:90ff24ef1cd1bd57124c2a3869cafee9d7b066909129ecdaba2c2d1903bc118d", size = 2750952, upload-time = "2026-08-20T10:05:08.327Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/91/71/f314841697a1d52af3e1ea7c5e1c3f09685b64ae4c58ff16b605a866255d/pygithub-2.10.0-py3-none-any.whl", hash = "sha256:192ada2a76e4afc7d6b37e500c9bfeba1731e6506697445a5ba1c4af8bf0b924", size = 455554, upload-time = "2026-08-20T10:05:06.991Z" }, +] + [[package]] name = "pygments" version = "2.20.0" @@ -1045,6 +1061,41 @@ crypto = [ { name = "cryptography" }, ] +[[package]] +name = "pynacl" +version = "1.6.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cffi", marker = "platform_python_implementation != 'PyPy'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d9/9a/4019b524b03a13438637b11538c82781a5eda427394380381af8f04f467a/pynacl-1.6.2.tar.gz", hash = "sha256:018494d6d696ae03c7e656e5e74cdfd8ea1326962cc401bcf018f1ed8436811c", size = 3511692, upload-time = "2026-01-01T17:48:10.851Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4b/79/0e3c34dc3c4671f67d251c07aa8eb100916f250ee470df230b0ab89551b4/pynacl-1.6.2-cp314-cp314t-macosx_10_10_universal2.whl", hash = "sha256:622d7b07cc5c02c666795792931b50c91f3ce3c2649762efb1ef0d5684c81594", size = 390064, upload-time = "2026-01-01T17:31:57.264Z" }, + { url = "https://files.pythonhosted.org/packages/eb/1c/23a26e931736e13b16483795c8a6b2f641bf6a3d5238c22b070a5112722c/pynacl-1.6.2-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:d071c6a9a4c94d79eb665db4ce5cedc537faf74f2355e4d502591d850d3913c0", size = 809370, upload-time = "2026-01-01T17:31:59.198Z" }, + { url = "https://files.pythonhosted.org/packages/87/74/8d4b718f8a22aea9e8dcc8b95deb76d4aae380e2f5b570cc70b5fd0a852d/pynacl-1.6.2-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:fe9847ca47d287af41e82be1dd5e23023d3c31a951da134121ab02e42ac218c9", size = 1408304, upload-time = "2026-01-01T17:32:01.162Z" }, + { url = "https://files.pythonhosted.org/packages/fd/73/be4fdd3a6a87fe8a4553380c2b47fbd1f7f58292eb820902f5c8ac7de7b0/pynacl-1.6.2-cp314-cp314t-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:04316d1fc625d860b6c162fff704eb8426b1a8bcd3abacea11142cbd99a6b574", size = 844871, upload-time = "2026-01-01T17:32:02.824Z" }, + { url = "https://files.pythonhosted.org/packages/55/ad/6efc57ab75ee4422e96b5f2697d51bbcf6cdcc091e66310df91fbdc144a8/pynacl-1.6.2-cp314-cp314t-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:44081faff368d6c5553ccf55322ef2819abb40e25afaec7e740f159f74813634", size = 1446356, upload-time = "2026-01-01T17:32:04.452Z" }, + { url = "https://files.pythonhosted.org/packages/78/b7/928ee9c4779caa0a915844311ab9fb5f99585621c5d6e4574538a17dca07/pynacl-1.6.2-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:a9f9932d8d2811ce1a8ffa79dcbdf3970e7355b5c8eb0c1a881a57e7f7d96e88", size = 826814, upload-time = "2026-01-01T17:32:06.078Z" }, + { url = "https://files.pythonhosted.org/packages/f7/a9/1bdba746a2be20f8809fee75c10e3159d75864ef69c6b0dd168fc60e485d/pynacl-1.6.2-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:bc4a36b28dd72fb4845e5d8f9760610588a96d5a51f01d84d8c6ff9849968c14", size = 1411742, upload-time = "2026-01-01T17:32:07.651Z" }, + { url = "https://files.pythonhosted.org/packages/f3/2f/5e7ea8d85f9f3ea5b6b87db1d8388daa3587eed181bdeb0306816fdbbe79/pynacl-1.6.2-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:3bffb6d0f6becacb6526f8f42adfb5efb26337056ee0831fb9a7044d1a964444", size = 801714, upload-time = "2026-01-01T17:32:09.558Z" }, + { url = "https://files.pythonhosted.org/packages/06/ea/43fe2f7eab5f200e40fb10d305bf6f87ea31b3bbc83443eac37cd34a9e1e/pynacl-1.6.2-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:2fef529ef3ee487ad8113d287a593fa26f48ee3620d92ecc6f1d09ea38e0709b", size = 1372257, upload-time = "2026-01-01T17:32:11.026Z" }, + { url = "https://files.pythonhosted.org/packages/4d/54/c9ea116412788629b1347e415f72195c25eb2f3809b2d3e7b25f5c79f13a/pynacl-1.6.2-cp314-cp314t-win32.whl", hash = "sha256:a84bf1c20339d06dc0c85d9aea9637a24f718f375d861b2668b2f9f96fa51145", size = 231319, upload-time = "2026-01-01T17:32:12.46Z" }, + { url = "https://files.pythonhosted.org/packages/ce/04/64e9d76646abac2dccf904fccba352a86e7d172647557f35b9fe2a5ee4a1/pynacl-1.6.2-cp314-cp314t-win_amd64.whl", hash = "sha256:320ef68a41c87547c91a8b58903c9caa641ab01e8512ce291085b5fe2fcb7590", size = 244044, upload-time = "2026-01-01T17:32:13.781Z" }, + { url = "https://files.pythonhosted.org/packages/33/33/7873dc161c6a06f43cda13dec67b6fe152cb2f982581151956fa5e5cdb47/pynacl-1.6.2-cp314-cp314t-win_arm64.whl", hash = "sha256:d29bfe37e20e015a7d8b23cfc8bd6aa7909c92a1b8f41ee416bbb3e79ef182b2", size = 188740, upload-time = "2026-01-01T17:32:15.083Z" }, + { url = "https://files.pythonhosted.org/packages/be/7b/4845bbf88e94586ec47a432da4e9107e3fc3ce37eb412b1398630a37f7dd/pynacl-1.6.2-cp38-abi3-macosx_10_10_universal2.whl", hash = "sha256:c949ea47e4206af7c8f604b8278093b674f7c79ed0d4719cc836902bf4517465", size = 388458, upload-time = "2026-01-01T17:32:16.829Z" }, + { url = "https://files.pythonhosted.org/packages/1e/b4/e927e0653ba63b02a4ca5b4d852a8d1d678afbf69b3dbf9c4d0785ac905c/pynacl-1.6.2-cp38-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:8845c0631c0be43abdd865511c41eab235e0be69c81dc66a50911594198679b0", size = 800020, upload-time = "2026-01-01T17:32:18.34Z" }, + { url = "https://files.pythonhosted.org/packages/7f/81/d60984052df5c97b1d24365bc1e30024379b42c4edcd79d2436b1b9806f2/pynacl-1.6.2-cp38-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:22de65bb9010a725b0dac248f353bb072969c94fa8d6b1f34b87d7953cf7bbe4", size = 1399174, upload-time = "2026-01-01T17:32:20.239Z" }, + { url = "https://files.pythonhosted.org/packages/68/f7/322f2f9915c4ef27d140101dd0ed26b479f7e6f5f183590fd32dfc48c4d3/pynacl-1.6.2-cp38-abi3-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:46065496ab748469cdd999246d17e301b2c24ae2fdf739132e580a0e94c94a87", size = 835085, upload-time = "2026-01-01T17:32:22.24Z" }, + { url = "https://files.pythonhosted.org/packages/3e/d0/f301f83ac8dbe53442c5a43f6a39016f94f754d7a9815a875b65e218a307/pynacl-1.6.2-cp38-abi3-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8a66d6fb6ae7661c58995f9c6435bda2b1e68b54b598a6a10247bfcdadac996c", size = 1437614, upload-time = "2026-01-01T17:32:23.766Z" }, + { url = "https://files.pythonhosted.org/packages/c4/58/fc6e649762b029315325ace1a8c6be66125e42f67416d3dbd47b69563d61/pynacl-1.6.2-cp38-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:26bfcd00dcf2cf160f122186af731ae30ab120c18e8375684ec2670dccd28130", size = 818251, upload-time = "2026-01-01T17:32:25.69Z" }, + { url = "https://files.pythonhosted.org/packages/c9/a8/b917096b1accc9acd878819a49d3d84875731a41eb665f6ebc826b1af99e/pynacl-1.6.2-cp38-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:c8a231e36ec2cab018c4ad4358c386e36eede0319a0c41fed24f840b1dac59f6", size = 1402859, upload-time = "2026-01-01T17:32:27.215Z" }, + { url = "https://files.pythonhosted.org/packages/85/42/fe60b5f4473e12c72f977548e4028156f4d340b884c635ec6b063fe7e9a5/pynacl-1.6.2-cp38-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:68be3a09455743ff9505491220b64440ced8973fe930f270c8e07ccfa25b1f9e", size = 791926, upload-time = "2026-01-01T17:32:29.314Z" }, + { url = "https://files.pythonhosted.org/packages/fa/f9/e40e318c604259301cc091a2a63f237d9e7b424c4851cafaea4ea7c4834e/pynacl-1.6.2-cp38-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:8b097553b380236d51ed11356c953bf8ce36a29a3e596e934ecabe76c985a577", size = 1363101, upload-time = "2026-01-01T17:32:31.263Z" }, + { url = "https://files.pythonhosted.org/packages/48/47/e761c254f410c023a469284a9bc210933e18588ca87706ae93002c05114c/pynacl-1.6.2-cp38-abi3-win32.whl", hash = "sha256:5811c72b473b2f38f7e2a3dc4f8642e3a3e9b5e7317266e4ced1fba85cae41aa", size = 227421, upload-time = "2026-01-01T17:32:33.076Z" }, + { url = "https://files.pythonhosted.org/packages/41/ad/334600e8cacc7d86587fe5f565480fde569dfb487389c8e1be56ac21d8ac/pynacl-1.6.2-cp38-abi3-win_amd64.whl", hash = "sha256:62985f233210dee6548c223301b6c25440852e13d59a8b81490203c3227c5ba0", size = 239754, upload-time = "2026-01-01T17:32:34.557Z" }, + { url = "https://files.pythonhosted.org/packages/29/7d/5945b5af29534641820d3bd7b00962abbbdfee84ec7e19f0d5b3175f9a31/pynacl-1.6.2-cp38-abi3-win_arm64.whl", hash = "sha256:834a43af110f743a754448463e8fd61259cd4ab5bbedcf70f9dabad1d28a394c", size = 184801, upload-time = "2026-01-01T17:32:36.309Z" }, +] + [[package]] name = "pytest" version = "8.4.2" @@ -1325,6 +1376,7 @@ dependencies = [ { name = "psycopg", extra = ["binary"] }, { name = "pydantic" }, { name = "pydantic-settings" }, + { name = "pygithub" }, { name = "python-dotenv" }, { name = "python-jose", extra = ["cryptography"] }, { name = "python-multipart" }, @@ -1372,6 +1424,7 @@ requires-dist = [ { name = "psycopg", extras = ["binary"], specifier = "~=3.1" }, { name = "pydantic", specifier = "~=2.13" }, { name = "pydantic-settings", specifier = "~=2.15" }, + { name = "pygithub", specifier = ">=2.10.0" }, { name = "python-dotenv", specifier = "~=1.2" }, { name = "python-jose", extras = ["cryptography"], specifier = "~=3.5" }, { name = "python-multipart", specifier = "~=0.0.9" },