From 2c88f3c45622405c4e527d01694b661e3af51ce4 Mon Sep 17 00:00:00 2001 From: maggiecbms Date: Mon, 26 Oct 2026 11:32:25 -0400 Subject: [PATCH 1/2] add copy workflow for ngs360 platform --- CHANGELOG.md | 10 +++ src/cwl_platform/ngs360_platform.py | 81 +++++++++++++++++++- tests/test_ngs360_ga4gh_platform.py | 110 +++++++++++++++++++++++++++- 3 files changed, 194 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 63765871..204810f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Changed + +- `NGS360Platform.copy_workflow` now resolves the workflow version the project + uses and returns `:`, instead of returning the workflow id + unchanged. A project therefore keeps running one version of a workflow even + after the `prod` alias moves on. An id that already names an alias or version + (`:3`) is still returned untouched. Requires an NGS360 API server with + `/projects/{id}/workflow-versions`. SevenBridges and Arvados are unaffected, + and callers need no change. + ## [v0.6.1](https://github.com/NGS360/PAML/releases/tag/v0.6.1) - 2026-09-21 ### Changed diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index ffe52e9b..f8d126f1 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -13,6 +13,9 @@ from .base_platform import Platform +OMICS_ENGINE = "AWSHealthOmics (us-east)" +PROD_ALIAS = "prod" + class WESTask: """ @@ -362,13 +365,85 @@ def upload_file( return None # Task/Workflow methods + def _resolve_workflow_version(self, project_id, workflow_id): + """ + Ask NGS360 which version of a workflow this project uses, recording the + version the `prod` alias points at if the project has none yet. + + :param project_id: NGS360 project business key, e.g. P-00000000-0001 + :param workflow_id: NGS360 workflow UUID + :return: The resolved record as a dict, including `version` + :raises RuntimeError: If NGS360 will not resolve a version + """ + response = requests.post( + f"{self.ngs360_endpoint}/api/v1/projects/{project_id}" + f"/workflow-versions/{workflow_id}/resolve", + json={"engine": OMICS_ENGINE, "alias": PROD_ALIAS}, + headers=self._auth_headers(), + timeout=30 + ) + if response.status_code not in (200, 201): + # Deliberately fatal. Raising error instead of returning the bare + # workflow id, which would leave WES to pick the highest version. + try: + detail = response.json().get("detail", response.text) + except ValueError: + detail = response.text + self.logger.error( + "NGS360 would not resolve a version of workflow %s for project %s: %s", + workflow_id, project_id, detail + ) + raise RuntimeError( + f"Could not resolve a version of workflow {workflow_id} for " + f"project {project_id} on {OMICS_ENGINE}: {detail}" + ) + return response.json() + def copy_workflow(self, src_workflow, destination_project): """ - Copy a workflow from one project to another + Return the workflow identifier this project should submit with. + + Nothing is copied: NGS360 has no per-project copy of a workflow. Instead + the project's recorded version is resolved and returned as + ":", so every run of a project uses the same version even + after the `prod` alias moves on. + + An identifier that already carries an alias or version (":prod", + ":3") is returned untouched, and the project's recorded version is + neither read nor written -- a one-off run must not change what the + project uses from then on. - Note: WES API doesn't have a direct concept of projects, so this returns the workflow + :param src_workflow: NGS360 workflow UUID, optionally ":" + :param destination_project: The project the workflow will run for + :return: The workflow identifier to submit + :raises RuntimeError: If NGS360 will not resolve a version """ - return src_workflow + if ":" in src_workflow: + self.logger.info( + "Workflow %s names a version explicitly; using it unchanged", + src_workflow + ) + return src_workflow + + project_id = (destination_project or {}).get("project_id") + if not project_id: + raise RuntimeError( + f"A project with a project_id is required to resolve a version of " + f"workflow {src_workflow}; got {destination_project!r}" + ) + + record = self._resolve_workflow_version(project_id, src_workflow) + version = record["version"] + self.logger.info( + "Workflow %s resolved to version %s for project %s (%s)", + src_workflow, version, project_id, + "newly recorded" if record.get("created") else "already recorded" + ) + if record.get("message"): + # Set when the recorded version is not what this call asked for, + # i.e. the alias has moved since the project first ran. + self.logger.info("NGS360: %s", record["message"]) + return f"{src_workflow}:{version}" def copy_workflows(self, reference_project, destination_project): """ diff --git a/tests/test_ngs360_ga4gh_platform.py b/tests/test_ngs360_ga4gh_platform.py index fb700dbf..365f8288 100644 --- a/tests/test_ngs360_ga4gh_platform.py +++ b/tests/test_ngs360_ga4gh_platform.py @@ -915,10 +915,6 @@ def test_workflow_methods(self): ''' project = {'project_id': 'test_project'} - # Test copy_workflow - result = self.platform.copy_workflow('src_workflow', project) - self.assertEqual(result, 'src_workflow') - # Test copy_workflows result = self.platform.copy_workflows('ref_project', project) self.assertEqual(result, []) @@ -1044,6 +1040,112 @@ def test_current_task_and_users(self): result = self.platform.get_project_users(project) self.assertEqual(result, []) + # --- copy_workflow: resolving the version a project uses ----------------- + + PROJECT = {'project_id': 'P-00000000-0001'} + WORKFLOW_ID = 'b6fd2ade-0000-0000-0000-000000000001' + RESOLVE_URL = ( + 'https://ngs360.example.com/api/v1/projects/P-00000000-0001' + '/workflow-versions/b6fd2ade-0000-0000-0000-000000000001/resolve' + ) + + @staticmethod + def _resolve_response(status_code=200, **body): + '''Build a mock NGS360 resolve response''' + response = MagicMock() + response.status_code = status_code + response.json.return_value = body + return response + + @patch('requests.post') + def test_copy_workflow_records_version_on_first_use(self, mock_post): + ''' + A bare workflow id resolves to the project's version, and is returned + suffixed with it so WES runs that exact version + ''' + mock_post.return_value = self._resolve_response(201, version=2, created=True) + + result = self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT) + + self.assertEqual(result, f'{self.WORKFLOW_ID}:2') + mock_post.assert_called_once_with( + self.RESOLVE_URL, + json={'engine': 'AWSHealthOmics (us-east)', 'alias': 'prod'}, + headers={'Authorization': 'Bearer test_token'}, + timeout=30 + ) + + @patch('requests.post') + def test_copy_workflow_reuses_recorded_version(self, mock_post): + ''' + The recorded version is used even once the alias has moved past it, + which is the whole point of recording it + ''' + mock_post.return_value = self._resolve_response( + 200, version=2, created=False, + message='Project already records version 2; returned it instead of alias \'prod\'.' + ) + + result = self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT) + + self.assertEqual(result, f'{self.WORKFLOW_ID}:2') + + @patch('requests.post') + def test_copy_workflow_leaves_explicit_versions_alone(self, mock_post): + ''' + An id that already names an alias or version is submitted as given, and + the project's recorded version is neither read nor written + ''' + for explicit in (f'{self.WORKFLOW_ID}:3', f'{self.WORKFLOW_ID}:prod'): + result = self.platform.copy_workflow(explicit, self.PROJECT) + self.assertEqual(result, explicit) + + mock_post.assert_not_called() + + @patch('requests.post') + def test_copy_workflow_raises_when_no_version_can_be_resolved(self, mock_post): + ''' + A refusal from NGS360 stops the caller rather than falling back to the + bare id, which WES would resolve to the highest version + ''' + for status_code in (404, 422): + mock_post.return_value = self._resolve_response( + status_code, detail='No deployment on that engine' + ) + with self.assertRaises(RuntimeError) as raised: + self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT) + self.assertIn('No deployment on that engine', str(raised.exception)) + + @patch('requests.post') + def test_copy_workflow_requires_a_project(self, mock_post): + ''' + Without a project there is nothing to resolve a version for, and the + error should say so rather than failing on a missing key + ''' + for no_project in (None, {}, {'name': 'Test Project'}): + with self.assertRaises(RuntimeError) as raised: + self.platform.copy_workflow(self.WORKFLOW_ID, no_project) + self.assertIn('project_id', str(raised.exception)) + + mock_post.assert_not_called() + + @patch('requests.request') + def test_task_reuse_ignores_the_version_suffix(self, mock_request): + ''' + Tasks are matched on tags, so the change from "" to ":N" + does not stop a rerun reusing completed tasks + ''' + mock_response = MagicMock() + mock_response.content = b'{"runs": []}' + mock_response.json.return_value = {'runs': []} + mock_request.return_value = mock_response + + self.platform.get_tasks_by_name(self.PROJECT, task_name='Sample 1 - alignment') + + _, kwargs = mock_request.call_args + filters = json.loads(kwargs['params']['filters']) + self.assertNotIn('workflow_url', filters) + def test_detect_method(self): ''' Test detect class method From 2a94b8bfa0a8c74d7c0969a3bdf222b87aab6063 Mon Sep 17 00:00:00 2001 From: maggiecbms Date: Mon, 26 Oct 2026 13:11:33 -0400 Subject: [PATCH 2/2] remove redudant log --- src/cwl_platform/ngs360_platform.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index f8d126f1..a83f4bdd 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -439,10 +439,6 @@ def copy_workflow(self, src_workflow, destination_project): src_workflow, version, project_id, "newly recorded" if record.get("created") else "already recorded" ) - if record.get("message"): - # Set when the recorded version is not what this call asked for, - # i.e. the alias has moved since the project first ran. - self.logger.info("NGS360: %s", record["message"]) return f"{src_workflow}:{version}" def copy_workflows(self, reference_project, destination_project):