diff --git a/CHANGELOG.md b/CHANGELOG.md index 72f965b8..a3b7200d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ ## lifebit-ai/cloudos-cli: changelog +## v2.94.0 (2026-06-23) + +### Feat: + +- Adds `cloudos queue create` with preset templates (e.g. `standard-stable`, `standard-gpu`) +- Adds `--set-default` to mark a new queue as the workspace default + ## v2.93.1 (2026-06-12) ### Fix: diff --git a/README.md b/README.md index b311d1e2..c4fdb3f6 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ Python package for interacting with Lifebit Platform - [Create Projects](#create-projects) - [Queue](#queue) - [List Queues](#list-queues) + - [Create Queue](#create-queue) - [Workflow](#workflow) - [List All Available Workflows](#list-all-available-workflows) - [Import a Nextflow Workflow](#import-a-nextflow-workflow) @@ -441,6 +442,35 @@ cloudos queue list --profile my_profile --output-format csv > NOTE: The queue name that is visible in Lifebit Platform and must be used with the `--job-queue` parameter is the one in the `label` field. +#### Create Queue + +You can create a new AWS batch job queue in your Lifebit Platform workspace using the `queue create` command. By default the queue is built from a preset template, selected with the `--preset` option. Available presets are: + +- **standard-stable** (default): On-demand stable instances +- **standard-cost-saving**: Spot instances (up to ~80% cheaper, at risk of premature termination) +- **read-write-optimised**: On-demand instances with increased disk I/O performance +- **standard-gpu**: On-demand standard instances plus GPU instances + +To create a queue from a preset: + +```bash +cloudos queue create --profile my_profile --label "my-new-queue" --description "Queue for RNA-seq jobs" --preset standard-stable +``` + +Before creating the queue, a summary is shown and confirmation is requested. To skip the confirmation prompt, add the `-y`/`--yes` flag. You can also set the new queue as the workspace default with `--set-default`. + +The expected output is something similar to: + +```console +Executing queue create... + Queue "my-new-queue" created successfully. + Queue ID : 64f1a23b8e4c9d001234abcd + View at : https://cloudos.lifebit.ai/app/job-queues/64f1a23b8e4c9d001234abcd +``` + +> [!NOTE] +> **Azure Platform**: Batch job queues are an AWS-only feature and are not available in Azure or HPC workspaces. + **Job queues for platform workflows** Platform workflows (those provided by Lifebit Platform in your workspace as modules) run on separate and specific AWS batch queues (system queues). Therefore, Lifebit Platform will automatically assign the valid queue and you should not specify any queue using the `--job-queue` parameter. Any attempt to use this parameter will be ignored. Examples of such platform workflows are "System Tools" and "Data Factory" workflows. diff --git a/cloudos_cli/_version.py b/cloudos_cli/_version.py index aaeaf336..51e998d4 100644 --- a/cloudos_cli/_version.py +++ b/cloudos_cli/_version.py @@ -1 +1 @@ -__version__ = '2.93.1' +__version__ = '2.94.0' diff --git a/cloudos_cli/queue/__init__.py b/cloudos_cli/queue/__init__.py index 15a458fe..fa4a104d 100755 --- a/cloudos_cli/queue/__init__.py +++ b/cloudos_cli/queue/__init__.py @@ -5,4 +5,4 @@ from .queue import Queue -__all__ = ['queue'] +__all__ = ['Queue'] diff --git a/cloudos_cli/queue/cli.py b/cloudos_cli/queue/cli.py index 1bf9f1f6..40bd222e 100644 --- a/cloudos_cli/queue/cli.py +++ b/cloudos_cli/queue/cli.py @@ -1,14 +1,48 @@ """CLI commands for Lifebit Platform job queue management.""" +import sys import rich_click as click import json -from cloudos_cli.queue.queue import Queue +from rich.console import Console +from cloudos_cli.queue.queue import ( + Queue, + QUEUE_PRESETS, + MAX_WORKSPACE_COMPUTE_ENVS, + WORKSPACE_CE_LIMIT_REACHED_MESSAGE, +) from cloudos_cli.utils.resources import ssl_selector from cloudos_cli.configure.configure import with_profile_config, CLOUDOS_URL from cloudos_cli.utils.cli_helpers import pass_debug_to_subcommands from cloudos_cli.utils.details import create_queue_list_table +# Workflow executors supported for job queues. +_EXECUTORS = ["nextflow", "cromwell"] + + +def _check_workspace_ce_limit(console, j_queue, queues=None): + """Exit with a warning if the workspace has reached its CE limit. + + Parameters + ---------- + console : rich.console.Console + The console used for rich output. + j_queue : Queue + The queue client used to count compute environments. + queues : list or None, optional + A pre-fetched list of job queue dicts. If ``None``, the queues are + fetched. + """ + count = j_queue.count_workspace_compute_environments(queues=queues) + if count >= MAX_WORKSPACE_COMPUTE_ENVS: + console.print( + f"[yellow]Warning:[/yellow] {WORKSPACE_CE_LIMIT_REACHED_MESSAGE}" + ) + sys.exit(0) + + + + # Create the queue group @click.group(cls=pass_debug_to_subcommands()) def queue(): @@ -50,6 +84,10 @@ def queue(): @click.option('--exclude-system-queues', help='Exclude system job queues from the list.', is_flag=True) +@click.option('--execution-platform', + help='Name of the execution platform implemented in your Lifebit Platform. Default=aws.', + type=click.Choice(['aws', 'azure', 'hpc']), + default='aws') @click.option('--disable-ssl-verification', help=('Disable SSL certificate verification. Please, remember that this option is ' + 'not generally recommended for security reasons.'), @@ -67,6 +105,7 @@ def list_queues(ctx, output_format, all_fields, exclude_system_queues, + execution_platform, disable_ssl_verification, ssl_cert, profile): @@ -74,6 +113,16 @@ def list_queues(ctx, # apikey, cloudos_url, and workspace_id are now automatically resolved by the decorator verify_ssl = ssl_selector(disable_ssl_verification, ssl_cert) + + # Batch job queues are an AWS-only feature; they are not available in + # Azure or HPC workspaces. + if execution_platform in ('azure', 'hpc'): + Console().print( + '[yellow]Warning:[/yellow] Batch job queues are not available in ' + f'{execution_platform.upper()} workspaces.' + ) + sys.exit(0) + print('Executing list...') j_queue = Queue(cloudos_url, apikey, None, workspace_id, verify=verify_ssl) my_queues = j_queue.get_job_queues(exclude_system_queues=exclude_system_queues) @@ -93,3 +142,149 @@ def list_queues(ctx, o.write(json.dumps(my_queues)) print(f'\tJob queue list collected with a total of {len(my_queues)} queues.') print(f'\tJob queue list saved to {outfile}') + + +@queue.command('create') +@click.option('-k', + '--apikey', + help='Your Lifebit Platform API key', + required=True) +@click.option('-c', + '--cloudos-url', + help=(f'The Lifebit Platform url you are trying to access to. Default={CLOUDOS_URL}.'), + default=CLOUDOS_URL, + required=True) +@click.option('--workspace-id', + help='The specific Lifebit Platform workspace id.', + required=True) +@click.option('--label', + help='Name (label) for the new job queue.', + required=False, + default=None) +@click.option('--description', + help='Short description of the new job queue.', + default='', + required=False) +@click.option('--preset', + help=( + 'Preset template to use. Choices: ' + + ', '.join(QUEUE_PRESETS.keys()) + + '. Default=standard-stable.' + ), + type=click.Choice(list(QUEUE_PRESETS.keys()), case_sensitive=False), + default='standard-stable', + show_default=True, + required=False) +@click.option('--executor', + help='Workflow executor for the queue. Default=nextflow.', + type=click.Choice(_EXECUTORS, case_sensitive=False), + default='nextflow', + show_default=True, + required=False) +@click.option('-y', + '--yes', + 'skip_confirmation', + help='Skip the confirmation prompt and proceed immediately.', + is_flag=True) +@click.option('--set-default', + help='Set the new job queue as the workspace default. Default=False.', + is_flag=True) +@click.option('--execution-platform', + help='Name of the execution platform implemented in your Lifebit Platform. Default=aws.', + type=click.Choice(['aws', 'azure', 'hpc']), + default='aws') +@click.option('--disable-ssl-verification', + help=('Disable SSL certificate verification. Please, remember that this option is ' + 'not generally recommended for security reasons.'), + is_flag=True) +@click.option('--ssl-cert', + help='Path to your SSL certificate file.') +@click.option('--profile', help='Profile to use from the config file', default=None) +@click.pass_context +@with_profile_config(required_params=['apikey', 'workspace_id']) +def create_queue(ctx, + apikey, + cloudos_url, + workspace_id, + label, + description, + preset, + executor, + skip_confirmation, + set_default, + execution_platform, + disable_ssl_verification, + ssl_cert, + profile): + """Create a new job queue in a Lifebit Platform workspace. + + A preset template is used to create the queue. + """ + + verify_ssl = ssl_selector(disable_ssl_verification, ssl_cert) + + # Batch job queues are an AWS-only feature; they are not available in + # Azure or HPC workspaces. + if execution_platform in ('azure', 'hpc'): + Console().print( + '[yellow]Warning:[/yellow] Batch job queues are not available in ' + f'{execution_platform.upper()} workspaces.' + ) + sys.exit(0) + + # --description is required when creating a queue. + if not description: + raise click.UsageError('Missing option --description.') + + if label is None: + raise click.UsageError('Missing option --label.') + + # Resolve the preset to show the user what will be created + preset_info = QUEUE_PRESETS[preset] + ce_name = preset_info['computeEnvironmentName'] + cr = preset_info['computeResources'] + resource_type = cr.get('type', 'EC2') + max_vcpus_preset = cr.get('maxvCpus', 'N/A') + instance_count = len(cr.get('instanceTypes', [])) + template_name = preset_info['templateName'] + + j_queue = Queue(cloudos_url, apikey, None, workspace_id, verify=verify_ssl) + + # Creating a queue creates a compute environment; enforce the workspace limit. + _check_workspace_ce_limit(Console(), j_queue) + + if not skip_confirmation: + click.echo('\nYou are about to create the following job queue:') + click.echo(f' Label : {label}') + click.echo(f' Description : {description or "(none)"}') + click.echo(f' Preset : {template_name}') + click.echo(f' Compute env name : {ce_name}') + click.echo(f' Resource type : {resource_type}') + click.echo(f' Max vCPUs : {max_vcpus_preset}') + click.echo(f' Instance types : {instance_count} types') + click.echo(f' Executor : {executor}') + click.echo(f' Set as default : {set_default}') + click.echo(f' Workspace : {workspace_id}') + click.echo('') + if not click.confirm('Proceed with queue creation?'): + click.echo('Aborted.') + sys.exit(0) + + console = Console() + print('Executing queue create...') + + try: + queue_id = j_queue.create_job_queue( + label=label, + description=description, + preset_name=preset, + executor=executor, + is_default=set_default, + ) + console.print(f'\t[green]Queue "{label}" created successfully.[/green]') + print(f'\tQueue ID : {queue_id}') + print(f'\tView at : {cloudos_url}/app/job-queues/{queue_id}') + except Exception as e: + console.print(f'\t[red]Error creating queue:[/red] {str(e)}') + sys.exit(1) + diff --git a/cloudos_cli/queue/queue.py b/cloudos_cli/queue/queue.py index 3ecffe51..37a3a7a8 100644 --- a/cloudos_cli/queue/queue.py +++ b/cloudos_cli/queue/queue.py @@ -2,13 +2,139 @@ This is the main class to create job queues. """ -import requests import json +import copy import pandas as pd from dataclasses import dataclass from typing import Union from cloudos_cli.clos import Cloudos -from cloudos_cli.utils.errors import BadRequestException +from cloudos_cli.utils.errors import ( + BadRequestException, + NoJobQueuesAvailableException, +) +from cloudos_cli.utils.requests import retry_requests_get, retry_requests_post + + +# --------------------------------------------------------------------------- +# Preset templates (match the CloudOS Platform UI presets exactly) +# --------------------------------------------------------------------------- + +_STANDARD_INSTANCE_TYPES = [ + "optimal", + "c4.2xlarge", "c4.4xlarge", "c4.8xlarge", + "c5.xlarge", "c5.2xlarge", "c5.4xlarge", "c5.9xlarge", + "c5.12xlarge", "c5.18xlarge", "c5.24xlarge", "c5.metal", + "m4.xlarge", "m4.2xlarge", "m4.4xlarge", "m4.10xlarge", "m4.16xlarge", + "m5.xlarge", "m5.2xlarge", "m5.4xlarge", "m5.8xlarge", + "m5.12xlarge", "m5.16xlarge", "m5.24xlarge", "m5.metal", + "r4.xlarge", "r4.2xlarge", "r4.4xlarge", "r4.8xlarge", "r4.16xlarge", + "r5.xlarge", "r5.2xlarge", "r5.4xlarge", "r5.8xlarge", + "r5.12xlarge", "r5.16xlarge", "r5.24xlarge", "r5.metal", +] + +_GPU_INSTANCE_TYPES = [ + "optimal", + "c4.2xlarge", "c4.4xlarge", "c4.8xlarge", + "c5.xlarge", "c5.2xlarge", "c5.4xlarge", "c5.9xlarge", + "c5.12xlarge", "c5.18xlarge", "c5.24xlarge", "c5.metal", + "g4dn.xlarge", "g4dn.2xlarge", "g4dn.4xlarge", "g4dn.8xlarge", + "g4dn.12xlarge", "g4dn.16xlarge", "g4dn.metal", + "m4.xlarge", "m4.2xlarge", "m4.4xlarge", "m4.10xlarge", "m4.16xlarge", + "m5.xlarge", "m5.2xlarge", "m5.4xlarge", "m5.8xlarge", + "m5.12xlarge", "m5.16xlarge", "m5.24xlarge", "m5.metal", + "p3.2xlarge", "p3.8xlarge", "p3.16xlarge", + "r4.xlarge", "r4.2xlarge", "r4.4xlarge", "r4.8xlarge", "r4.16xlarge", + "r5.xlarge", "r5.2xlarge", "r5.4xlarge", "r5.8xlarge", + "r5.12xlarge", "r5.16xlarge", "r5.24xlarge", "r5.metal", +] + +QUEUE_PRESETS = { + "standard-stable": { + "computeEnvironmentName": "OnDemandStandard", + "computeResources": { + "allocationStrategy": "BEST_FIT_PROGRESSIVE", + "instanceTypes": _STANDARD_INSTANCE_TYPES, + "maxvCpus": 512, + "type": "EC2", + "minvCpus": 0, + }, + "templateName": "Standard stable", + "templateDescription": ( + "Standard stable (on-demand) instances of all resource types from " + "c5, r5, m5, c4, r4, m4 instance families." + ), + }, + "standard-cost-saving": { + "computeEnvironmentName": "OnDemandSpot", + "computeResources": { + "allocationStrategy": "SPOT_CAPACITY_OPTIMIZED", + "instanceTypes": _STANDARD_INSTANCE_TYPES, + "maxvCpus": 512, + "type": "SPOT", + "minvCpus": 0, + "bidPercentage": 100, + }, + "templateName": "Standard cost-saving", + "templateDescription": ( + "Standard cost-saving (spot) instances of all resource types from " + "c5, r5, m5, c4, r4, m4 instance families. Spot instances allow to " + "save up to 80% cost compared to on-demand stable instances at a risk " + "of being prematurely terminated. Useful for short-running processes. " + "It is advised to use retry error strategy in the workflow for this job queue." + ), + }, + "read-write-optimised": { + "computeEnvironmentName": "OnDemandStandardHighDiskThroughput", + "computeResources": { + "allocationStrategy": "BEST_FIT_PROGRESSIVE", + "instanceTypes": _STANDARD_INSTANCE_TYPES, + "maxvCpus": 512, + "type": "EC2", + "minvCpus": 0, + "volume": { + "type": "gp3", + "size": {"usageQuantity": 1000, "usageUnit": "Gb"}, + "iops": 5000, + "throughput": 500, + "deviceName": "/dev/xvda", + "deleteOnTermination": False, + "encrypted": False, + }, + }, + "templateName": "Read/write optimised", + "templateDescription": ( + "Standard stable (on-demand) instances of all resource types from " + "c5, r5, m5, c4, r4, m4 instance families and increased disk I/O " + "performance. Useful for the jobs that require significant file read " + "and write activity. May increase the job cost." + ), + }, + "standard-gpu": { + "computeEnvironmentName": "OnDemandStandardGPUs", + "computeResources": { + "allocationStrategy": "BEST_FIT_PROGRESSIVE", + "instanceTypes": _GPU_INSTANCE_TYPES, + "maxvCpus": 512, + "type": "EC2", + "minvCpus": 0, + }, + "templateName": "Standard with GPUs", + "templateDescription": ( + "Standard stable (on-demand) instances as well as GPU instances of " + "p3 and/or g4dn families. On-demand GPU machines typically incur higher costs." + ), + }, +} + + +# Maximum number of compute environments a workspace can hold across all queues. +MAX_WORKSPACE_COMPUTE_ENVS = 10 + +# Message shown once a workspace reaches the compute environment limit. +WORKSPACE_CE_LIMIT_REACHED_MESSAGE = ( + "You have reached the limit for compute environments in your workspace. " + f"Workspaces can have up to {MAX_WORKSPACE_COMPUTE_ENVS} compute environments." +) @dataclass @@ -47,9 +173,9 @@ def get_job_queues(self, exclude_system_queues=False): A list of dicts, each corresponding to a job queue. """ headers = {"apikey": self.apikey} - r = requests.get("{}/api/v1/teams/aws/v2/job-queues?teamId={}".format(self.cloudos_url, - self.workspace_id), - headers=headers, verify=self.verify) + r = retry_requests_get("{}/api/v1/teams/aws/v2/job-queues?teamId={}".format(self.cloudos_url, + self.workspace_id), + headers=headers, verify=self.verify) if r.status_code >= 400: raise BadRequestException(r) queues = json.loads(r.content) @@ -69,9 +195,9 @@ def get_system_job_queues(self): A list of dicts, each corresponding to a system job queue. """ headers = {"apikey": self.apikey} - r = requests.get("{}/api/v1/teams/aws/v2/system-job-queues?teamId={}".format(self.cloudos_url, - self.workspace_id), - headers=headers, verify=self.verify) + r = retry_requests_get("{}/api/v1/teams/aws/v2/system-job-queues?teamId={}".format(self.cloudos_url, + self.workspace_id), + headers=headers, verify=self.verify) if r.status_code >= 400: raise BadRequestException(r) return json.loads(r.content) @@ -143,8 +269,7 @@ def fetch_job_queue_id(self, workflow_type, batch=True, job_queue=None): available_queues = [q for q in job_queues if q['status'] == 'Ready' and q['executor'] == workflow_type] if len(available_queues) == 0: - raise Exception(f'There are no available job queues for {workflow_type} ' + - 'workflows. Consider creating one using Lifebit Platform UI.') + raise NoJobQueuesAvailableException(workflow_type) default_queue = [q for q in available_queues if q.get('isDefault', False)] if len(default_queue) > 0: default_queue_id = default_queue[0]['id'] @@ -164,3 +289,138 @@ def fetch_job_queue_id(self, workflow_type, batch=True, job_queue=None): f'queue instead: {default_queue_name}.') return default_queue_id return selected_queue[0]['id'] + + @staticmethod + def get_preset_template(preset_name): + """Return the environment and template fields for a given preset name. + + Parameters + ---------- + preset_name : str + One of: 'standard-stable', 'standard-cost-saving', + 'read-write-optimised', 'standard-gpu'. + + Returns + ------- + template : dict + A dict with keys 'computeEnvironmentName', 'computeResources', + 'templateName', and 'templateDescription'. + + Raises + ------ + ValueError + If ``preset_name`` is not a recognised preset. + """ + if preset_name not in QUEUE_PRESETS: + valid = ', '.join(QUEUE_PRESETS.keys()) + raise ValueError( + f"Unknown preset '{preset_name}'. Valid presets are: {valid}" + ) + return copy.deepcopy(QUEUE_PRESETS[preset_name]) + + def create_job_queue(self, label, description, preset_name, executor="nextflow", + is_default=False): + """Create a new job queue in the workspace using a preset template. + + Parameters + ---------- + label : str + Human-readable name for the queue. + description : str + Short description of the queue's purpose. + preset_name : str + One of the supported preset keys (see ``QUEUE_PRESETS``). + executor : str, optional + Workflow executor. Defaults to ``'nextflow'``. + is_default : bool, optional + Whether to set the queue as the workspace default. Defaults to + ``False``. + + Returns + ------- + queue_id : str + The Lifebit Platform ID assigned to the newly created queue. + + Raises + ------ + BadRequestException + If the API returns a 4xx or 5xx response. + """ + preset = self.get_preset_template(preset_name) + payload = { + "id": "", + "label": label, + "description": description, + "executor": executor, + "status": "ToCreate", + "environment": { + "computeEnvironmentName": preset["computeEnvironmentName"], + "computeResources": preset["computeResources"], + }, + "templateName": preset["templateName"], + "templateDescription": preset["templateDescription"], + "isDefault": is_default, + } + return self._post_job_queue(payload) + + def _post_job_queue(self, payload): + """POST a job queue payload to the Lifebit Platform and return its ID. + + Parameters + ---------- + payload : dict + The fully-built job queue creation payload. + + Returns + ------- + queue_id : str + The Lifebit Platform ID assigned to the newly created queue. + + Raises + ------ + BadRequestException + If the API returns a 4xx or 5xx response. + """ + headers = { + "Content-Type": "application/json", + "apikey": self.apikey, + } + r = retry_requests_post( + "{}/api/v1/teams/aws/v2/job-queue?teamId={}".format( + self.cloudos_url, self.workspace_id + ), + headers=headers, + json=payload, + verify=self.verify, + ) + if r.status_code >= 400: + raise BadRequestException(r) + response_data = json.loads(r.content) + queue_id = response_data.get("id") or response_data.get("_id") + if not queue_id: + raise RuntimeError( + "Job queue creation succeeded but the server response did not " + "include a queue ID." + ) + return queue_id + + def count_workspace_compute_environments(self, queues=None): + """Count the total compute environments across all queues in the workspace. + + System job queues are not counted towards the workspace limit. + + Parameters + ---------- + queues : list or None, optional + A list of (non-system) job queue dicts as returned by + ``get_job_queues(exclude_system_queues=True)``. If ``None``, the + queues are fetched, excluding system queues. + + Returns + ------- + count : int + The total number of compute environments in the workspace. + """ + if queues is None: + queues = self.get_job_queues(exclude_system_queues=True) + return sum(len(q.get('computeEnvironments', [])) for q in queues) diff --git a/cloudos_cli/utils/errors.py b/cloudos_cli/utils/errors.py index 8bdd72e0..6ba23edb 100755 --- a/cloudos_cli/utils/errors.py +++ b/cloudos_cli/utils/errors.py @@ -80,6 +80,25 @@ def __init__(self, workspace_id): self.workspace_id = workspace_id +class NoJobQueuesAvailableException(Exception): + """Raised when no suitable job queues exist for a given workflow type. + + Parameters + ---------- + workflow_type : str + The workflow type (e.g. ``'nextflow'`` or ``'cromwell'``) for which no + ready job queue could be found. + """ + def __init__(self, workflow_type): + msg = ( + f"There are no available job queues for {workflow_type} workflows. " + "Consider creating one using 'cloudos queue create' or the Lifebit " + "Platform UI." + ) + super(NoJobQueuesAvailableException, self).__init__(msg) + self.workflow_type = workflow_type + + class JobAccessDeniedException(Exception): def __init__(self, job_id, job_owner_name=None, current_user_name=None): if job_owner_name and current_user_name: diff --git a/tests/test_data/queue/create_queue_response.json b/tests/test_data/queue/create_queue_response.json new file mode 100644 index 00000000..7b0c3181 --- /dev/null +++ b/tests/test_data/queue/create_queue_response.json @@ -0,0 +1,21 @@ +{ + "_id": "6a2039447e4338212900fe65", + "id": "6a2039447e4338212900fe65", + "label": "My Custom Job Queue", + "description": "Custom job queue for batch processing workloads", + "executor": "nextflow", + "status": "ToCreate", + "isDefault": false, + "templateName": "Standard stable", + "templateDescription": "Standard stable (on-demand) instances of all resource types from c5, r5, m5, c4, r4, m4 instance families.", + "environment": { + "computeEnvironmentName": "OnDemandStandard", + "computeResources": { + "allocationStrategy": "BEST_FIT_PROGRESSIVE", + "instanceTypes": ["optimal", "c5.xlarge"], + "maxvCpus": 512, + "type": "EC2", + "minvCpus": 0 + } + } +} diff --git a/tests/test_queue/test_create_queue.py b/tests/test_queue/test_create_queue.py new file mode 100644 index 00000000..08ed1690 --- /dev/null +++ b/tests/test_queue/test_create_queue.py @@ -0,0 +1,461 @@ +"""Tests for the queue create command and Queue.create_job_queue() method.""" + +import json +import re +import pytest +import responses +import requests_mock as requests_mock_module +from click.testing import CliRunner + +from cloudos_cli.queue.queue import Queue, QUEUE_PRESETS +from cloudos_cli.utils.errors import ( + BadRequestException, +) +from cloudos_cli.__main__ import run_cloudos_cli + +# rich_click renders usage errors in a styled panel and highlights option +# tokens (e.g. ``--description``) with ANSI escape codes. When colour output is +# enabled (as in CI), those escape codes are inserted between the surrounding +# text and the option token, breaking naive substring assertions. Strip ANSI +# escape sequences before asserting on the human-readable message. +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def _plain(text): + """Return ``text`` with ANSI escape sequences removed.""" + return _ANSI_RE.sub("", text) + +# --------------------------------------------------------------------------- +# Constants shared across tests +# --------------------------------------------------------------------------- + +APIKEY = 'vnoiweur89u2ongs' +CLOUDOS_URL = 'https://cloudos.lifebit.ai' +WORKSPACE_ID = 'lv89ufc838sdig' +CREATE_RESPONSE_FILE = 'tests/test_data/queue/create_queue_response.json' + +with open(CREATE_RESPONSE_FILE) as f: + CREATE_RESPONSE_JSON_STR = f.read() + CREATE_RESPONSE_JSON_DICT = json.loads(CREATE_RESPONSE_JSON_STR) + + +def _mock_get_queues_with_total_ces(m, total_ces, n_queues=1): + """Mock the GET job-queues endpoints with queues totalling ``total_ces`` CEs. + + The compute environments are spread across ``n_queues`` team queues. The + system-job-queues endpoint is mocked with an empty list. + """ + per_queue = [] + remaining = total_ces + for i in range(n_queues): + count = remaining if i == n_queues - 1 else remaining // (n_queues - i) + remaining -= count + per_queue.append(count) + queues = [ + { + 'id': f'q{i}', + 'name': f'queue-{i}', + 'label': f'queue-{i}', + 'description': '', + 'isDefault': False, + 'resourceType': '', + 'executor': 'nextflow', + 'computeEnvironments': [ + {'label': f'CE-{i}-{j}', 'environment': {}, 'status': 'Ready'} + for j in range(count) + ], + 'status': 'Ready', + } + for i, count in enumerate(per_queue) + ] + m.get( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queues?teamId={WORKSPACE_ID}", + text=json.dumps(queues), + status_code=200, + ) + m.get( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/system-job-queues?teamId={WORKSPACE_ID}", + text='[]', + status_code=200, + ) + + +# =========================================================================== +# Unit tests – Queue.get_preset_template() +# =========================================================================== + +class TestGetPresetTemplate: + def test_returns_dict_for_each_known_preset(self): + for preset_name in QUEUE_PRESETS: + result = Queue.get_preset_template(preset_name) + assert isinstance(result, dict) + assert 'computeEnvironmentName' in result + assert 'computeResources' in result + assert 'templateName' in result + assert 'templateDescription' in result + + def test_standard_stable_preset_uses_ec2(self): + preset = Queue.get_preset_template('standard-stable') + assert preset['computeResources']['type'] == 'EC2' + assert preset['computeResources']['allocationStrategy'] == 'BEST_FIT_PROGRESSIVE' + + def test_standard_cost_saving_preset_uses_spot(self): + preset = Queue.get_preset_template('standard-cost-saving') + assert preset['computeResources']['type'] == 'SPOT' + assert preset['computeResources']['allocationStrategy'] == 'SPOT_CAPACITY_OPTIMIZED' + assert preset['computeResources']['bidPercentage'] == 100 + + def test_read_write_optimised_has_volume(self): + preset = Queue.get_preset_template('read-write-optimised') + assert 'volume' in preset['computeResources'] + assert preset['computeResources']['volume']['type'] == 'gp3' + + def test_standard_gpu_includes_gpu_instances(self): + preset = Queue.get_preset_template('standard-gpu') + instance_types = preset['computeResources']['instanceTypes'] + gpu_instances = [i for i in instance_types if i.startswith(('g4dn', 'p3'))] + assert len(gpu_instances) > 0 + + def test_raises_value_error_for_unknown_preset(self): + with pytest.raises(ValueError, match="Unknown preset"): + Queue.get_preset_template('nonexistent-preset') + + def test_all_presets_have_optimal_instance_type(self): + for preset_name in QUEUE_PRESETS: + preset = Queue.get_preset_template(preset_name) + assert 'optimal' in preset['computeResources']['instanceTypes'] + + def test_all_presets_have_max_vcpus(self): + for preset_name in QUEUE_PRESETS: + preset = Queue.get_preset_template(preset_name) + assert preset['computeResources']['maxvCpus'] > 0 + + +# =========================================================================== +# Unit tests – Queue.create_job_queue() (API mocked via responses) +# =========================================================================== + +class TestCreateJobQueue: + def _make_queue(self): + return Queue( + cloudos_url=CLOUDOS_URL, + apikey=APIKEY, + cromwell_token=None, + workspace_id=WORKSPACE_ID, + ) + + @responses.activate + def test_create_job_queue_success_returns_id(self): + responses.add( + responses.POST, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + body=CREATE_RESPONSE_JSON_STR, + status=200, + content_type='application/json', + ) + q = self._make_queue() + queue_id = q.create_job_queue( + label='Test Queue', + description='A test queue', + preset_name='standard-stable', + ) + assert queue_id == CREATE_RESPONSE_JSON_DICT['_id'] + + @responses.activate + def test_create_job_queue_posts_correct_payload(self): + responses.add( + responses.POST, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + body=CREATE_RESPONSE_JSON_STR, + status=200, + content_type='application/json', + ) + q = self._make_queue() + q.create_job_queue( + label='My Queue', + description='desc', + preset_name='standard-cost-saving', + executor='nextflow', + ) + # Inspect what was sent + sent_payload = json.loads(responses.calls[0].request.body) + assert sent_payload['label'] == 'My Queue' + assert sent_payload['description'] == 'desc' + assert sent_payload['executor'] == 'nextflow' + assert sent_payload['status'] == 'ToCreate' + assert sent_payload['isDefault'] is False + assert sent_payload['id'] == '' + assert sent_payload['environment']['computeResources']['type'] == 'SPOT' + + @responses.activate + def test_create_job_queue_raises_on_400(self): + error_body = json.dumps({ + 'statusCode': 400, + 'code': 'BadRequest', + 'message': 'Bad Request.', + }) + responses.add( + responses.POST, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + body=error_body, + status=400, + content_type='application/json', + ) + q = self._make_queue() + with pytest.raises(BadRequestException): + q.create_job_queue( + label='Bad Queue', + description='', + preset_name='standard-stable', + ) + + @responses.activate + def test_create_job_queue_all_presets_succeed(self): + for preset_name in QUEUE_PRESETS: + responses.add( + responses.POST, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + body=CREATE_RESPONSE_JSON_STR, + status=200, + content_type='application/json', + ) + q = self._make_queue() + for preset_name in QUEUE_PRESETS: + queue_id = q.create_job_queue( + label=f'queue-{preset_name}', + description='test', + preset_name=preset_name, + ) + assert queue_id != '' + + def test_create_job_queue_raises_on_invalid_preset(self): + q = self._make_queue() + with pytest.raises(ValueError, match='Unknown preset'): + q.create_job_queue( + label='Queue', + description='', + preset_name='not-a-preset', + ) + + +# =========================================================================== +# CLI integration tests – `cloudos queue create` +# =========================================================================== + +class TestCreateQueueCLI: + def _base_args(self, extra=None): + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--label', 'Test Queue', + '--description', 'A test queue', + '--preset', 'standard-stable', + '--yes', + ] + if extra: + args.extend(extra) + return args + + def test_create_command_exists_in_help(self): + runner = CliRunner() + result = runner.invoke(run_cloudos_cli, ['queue', '--help']) + assert result.exit_code == 0 + assert 'create' in result.output + + def test_create_command_help(self): + runner = CliRunner() + result = runner.invoke(run_cloudos_cli, ['queue', 'create', '--help']) + assert result.exit_code == 0 + assert '--label' in result.output + assert '--preset' in result.output + assert '--yes' in result.output + assert '--description' in result.output + + def test_create_command_all_presets_in_help(self): + runner = CliRunner() + result = runner.invoke(run_cloudos_cli, ['queue', 'create', '--help']) + for preset_name in QUEUE_PRESETS: + assert preset_name in result.output + + def test_create_queue_success_with_yes_flag(self): + runner = CliRunner() + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 1) + m.post( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + text=CREATE_RESPONSE_JSON_STR, + status_code=200, + ) + result = runner.invoke(run_cloudos_cli, self._base_args()) + assert result.exit_code == 0 + assert 'created successfully' in result.output + assert CREATE_RESPONSE_JSON_DICT['_id'] in result.output + + def test_create_queue_shows_url_on_success(self): + runner = CliRunner() + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 1) + m.post( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + text=CREATE_RESPONSE_JSON_STR, + status_code=200, + ) + result = runner.invoke(run_cloudos_cli, self._base_args()) + assert CLOUDOS_URL in result.output + assert 'job-queues' in result.output + + def test_create_queue_aborted_on_confirmation_decline(self): + runner = CliRunner() + # Do NOT pass --yes; answer 'n' to the prompt + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--label', 'Test Queue', + '--description', 'A test queue', + '--preset', 'standard-stable', + ] + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 1) + result = runner.invoke(run_cloudos_cli, args, input='n\n') + assert result.exit_code == 0 + assert 'Aborted' in result.output + + def test_create_queue_proceeds_on_confirmation_accept(self): + runner = CliRunner() + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--label', 'Test Queue', + '--description', 'A test queue', + '--preset', 'standard-stable', + ] + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 1) + m.post( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + text=CREATE_RESPONSE_JSON_STR, + status_code=200, + ) + result = runner.invoke(run_cloudos_cli, args, input='y\n') + assert result.exit_code == 0 + assert 'created successfully' in result.output + + def test_create_queue_api_error_exits_nonzero(self): + runner = CliRunner() + error_body = json.dumps({'statusCode': 400, 'message': 'Bad Request.'}) + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 1) + m.post( + f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queue?teamId={WORKSPACE_ID}", + text=error_body, + status_code=400, + ) + result = runner.invoke(run_cloudos_cli, self._base_args()) + assert result.exit_code == 1 + assert 'Error' in result.output + + def test_create_queue_workspace_limit_reached_exits(self): + runner = CliRunner() + with requests_mock_module.Mocker() as m: + _mock_get_queues_with_total_ces(m, 10, n_queues=4) + result = runner.invoke(run_cloudos_cli, self._base_args()) + assert result.exit_code == 0 + assert 'reached the limit for compute environments in your workspace' \ + in result.output + + def test_create_queue_invalid_preset_rejected(self): + runner = CliRunner() + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--label', 'Test Queue', + '--description', 'A test queue', + '--preset', 'not-a-real-preset', + '--yes', + ] + result = runner.invoke(run_cloudos_cli, args) + assert result.exit_code != 0 + + def test_create_queue_missing_label_fails(self): + runner = CliRunner() + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--description', 'A test queue', + '--yes', + ] + result = runner.invoke(run_cloudos_cli, args) + assert result.exit_code != 0 + + def test_create_queue_missing_description_fails(self): + runner = CliRunner() + args = [ + 'queue', 'create', + '--apikey', APIKEY, + '--cloudos-url', CLOUDOS_URL, + '--workspace-id', WORKSPACE_ID, + '--label', 'Test Queue', + '--preset', 'standard-stable', + '--yes', + ] + result = runner.invoke(run_cloudos_cli, args) + assert result.exit_code != 0 + assert 'Missing option --description' in _plain(result.output) + + +# =========================================================================== +# Unit tests – Queue.count_workspace_compute_environments() +# =========================================================================== + +QUEUES_LIST_FILE = 'tests/test_data/queue/queues.json' + +with open(QUEUES_LIST_FILE) as f: + QUEUES_LIST_STR = f.read() + + +class TestCountWorkspaceComputeEnvironments: + def _make_queue(self): + return Queue( + cloudos_url=CLOUDOS_URL, + apikey=APIKEY, + cromwell_token=None, + workspace_id=WORKSPACE_ID, + ) + + def test_count_workspace_compute_environments_from_list(self): + q = self._make_queue() + queues = [ + {'computeEnvironments': [{}, {}]}, + {'computeEnvironments': [{}]}, + {}, + ] + assert q.count_workspace_compute_environments(queues=queues) == 3 + + @responses.activate + def test_count_workspace_compute_environments_fetches_queues(self): + # System queues are excluded from the workspace CE count, so only the + # team job-queues endpoint contributes. + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queues?teamId={WORKSPACE_ID}", + body=QUEUES_LIST_STR, + status=200, + content_type='application/json', + ) + q = self._make_queue() + team_queues = json.loads(QUEUES_LIST_STR) + expected = sum( + len(x.get('computeEnvironments', [])) for x in team_queues + ) + assert q.count_workspace_compute_environments() == expected + diff --git a/tests/test_queue/test_fetch_job_queue_id.py b/tests/test_queue/test_fetch_job_queue_id.py index 39a3cdc2..93aede0a 100644 --- a/tests/test_queue/test_fetch_job_queue_id.py +++ b/tests/test_queue/test_fetch_job_queue_id.py @@ -2,6 +2,7 @@ import pytest import responses from cloudos_cli.queue import Queue +from cloudos_cli.utils.errors import NoJobQueuesAvailableException from tests.functions_for_pytest import load_json_file INPUT = 'tests/test_data/queue/queues.json' @@ -147,3 +148,39 @@ def test_fetch_job_queue_id_batch_true_workflow_type_wrong(): with pytest.raises(ValueError) as error: j_queue.fetch_job_queue_id('wrong_workflow_type', batch=True) assert 'Only nextflow or cromwell workflows are allowed' in str(error) + + +@mock.patch('cloudos_cli.queue', mock.MagicMock()) +@responses.activate +def test_fetch_job_queue_id_no_available_queues(): + """ + Tests fetch_job_queue_id when batch=True but there are no ready job queues + for the requested workflow type, so a NoJobQueuesAvailableException is raised. + """ + header = { + "Accept": "application/json, text/plain, */*", + "Content-Type": "application/json;charset=UTF-8", + "apikey": APIKEY + } + # mock GET method for regular queues returning an empty list + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/job-queues?teamId={WORKSPACE_ID}", + body="[]", + headers=header, + status=200) + # mock GET method for system queues returning an empty list + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v1/teams/aws/v2/system-job-queues?teamId={WORKSPACE_ID}", + body="[]", + headers=header, + status=200) + # Initialise Queue + j_queue = Queue(cloudos_url=CLOUDOS_URL, apikey=APIKEY, cromwell_token=None, + workspace_id=WORKSPACE_ID) + # Raise NoJobQueuesAvailableException + with pytest.raises(NoJobQueuesAvailableException) as error: + j_queue.fetch_job_queue_id(WORKFLOW_TYPE, batch=True) + assert f'There are no available job queues for {WORKFLOW_TYPE} workflows' in str(error.value) + assert error.value.workflow_type == WORKFLOW_TYPE