MIDRC-1292 Enable Nextflow tests in Kind CI + test_pv_posix_ops

gen3-integration-tests/gen3_ci/scripts/setup_ci_env.sh

@@ -465,11 +465,6 @@ common_param_updates=(
   ".funnel.externalSecrets.dbcreds|${namespace}-funnel-creds"
   ".funnel.externalSecrets.funnelOidcClient|${namespace}-funnel-oidc-client"
   ".funnel.Kubernetes.JobsNamespace|workflow-pods-${namespace}"
-  # TODO: Remove the next line after funnel helm chart is completely removed as a dependent chart.
-  # The legacy chart still expects `funnel.funnel.Kubernetes.JobsNamespace`,
-  # while the standalone chart uses `funnel.Kubernetes.JobsNamespace`.
-  # Keep both values in sync until the migration is complete.
-  ".funnel.funnel.Kubernetes.JobsNamespace|workflow-pods-${namespace}"
 )
 
 for item in "${common_param_updates[@]}"; do

gen3-integration-tests/services/fence.py

@@ -83,7 +83,7 @@ def create_signed_url(
             response = requests.get(self.BASE_URL + url, auth={})
             status_code = response.status_code
             response = response.content.decode()
-        logger.info("Status code : " + str(status_code))
+        logger.info(f"Status code: {status_code} - {response}")
         assert (
             expected_status == status_code
         ), f"Expected response {expected_status}, but got {status_code}"

gen3-integration-tests/test_data/gen3_workflow/nextflow.config

@@ -3,6 +3,7 @@ plugins {
 }
 process {
 	executor = 'tes'
+	// Dockerfile location: https://github.com/uc-cdis/bio-nextflow/tree/c8f9fd595135078b1ac1b365aa3393641b5cacdb/nextflow_notebooks/containerized_cpu_workflows/midrc_batch_demo
 	container = 'quay.io/cdis/gen3-workflow:integration_tests_dicom_image'
 	// this test tends to fail intermittently; improve stability with retries for now:
 	process.errorStrategy = 'retry'

gen3-integration-tests/test_data/gen3_workflow/test_pv_posix_ops.py

@@ -0,0 +1,36 @@
+import os
+import zipfile
+
+from PIL import Image, ImageDraw
+
+print("SKIP_BROKEN:", os.environ.get("SKIP_BROKEN"))
+SKIP_BROKEN = os.environ.get("SKIP_BROKEN") == "YES"
+
+
+# Test sequential file writes
+with open("output.txt", "w") as f:
+    f.write("Initial sequential data\n")
+if not SKIP_BROKEN:
+    with open("output.txt", "a") as f:
+        f.write("Second sequential data\n")
+
+
+# Test multipage PDF creation
+page_one = Image.new("RGB", (100, 200), color="white")
+draw_tool = ImageDraw.Draw(page_one)
+draw_tool.rectangle([(10, 10), (40, 40)], fill="black")
+page_two = Image.new("RGB", (300, 400), color="lightgray")
+draw_tool = ImageDraw.Draw(page_two)
+draw_tool.rectangle([(20, 20), (40, 40)], fill=(255, 0, 0))
+page_one.save("output.pdf", "PDF", save_all=True, append_images=[page_two])
+
+
+# Test ZIP file creation
+files_to_zip = ["output.txt", "output.pdf"]
+if not SKIP_BROKEN:
+    with zipfile.ZipFile("output.zip", "w", zipfile.ZIP_DEFLATED) as f:
+        for file in files_to_zip:
+            f.write(file)
+else:
+    with open("output.zip", "w") as f:
+        f.write("")

gen3-integration-tests/tests/test_gen3_workflow.py

@@ -24,6 +24,7 @@
 import subprocess
 import tempfile
 import time
+import zipfile
 
 import jwt
 import pytest
@@ -1178,25 +1179,221 @@ def test_task_with_environment_variable(self):
             "SOMETHING=VALUE" in stdout
         ), f"Expected env var to be set, but `env` returned: {stdout}"
 
+    def test_pv_posix_ops(self):
+        """
+        Regression test for various issues caused by S3-CSI driver limitations. Fixed by switching
+        PVs to S3Files.
+        This is written as a single test to keep all the PV-related POSIX issues together, and to
+        run fewer TES tasks to speed up testing.
+
+        - Rename and move operations
+        - Incremental/append writes to files already flushed to S3
+          Note: unable to reproduce issue at this time
+        - Random-access writes - creation of binary formats requiring non sequential writes (eg
+          ZIP, PDF, FastQ, GZ)
+          Note: unable to reproduce PDF, FastQ and GZ issues at this time
+          MIDRC-1278: GZ files - It works when the input is `s3://{s3_path_prefix}/input.txt.gz`.
+          We may be able to reproduce the issue by using a presigned URL instead, but that case is
+          not trivial to write since existing tests assume the files they download are already
+          present in the S3 bucket, and here we need to upload a new GZ file to a bucket we have
+          access to (so, likely through Fence, in a bucket configured in Fence).
+        """
+        # temporarily used to disable tests that are broken in CI
+        # TODO remove this once we have switched to S3Files
+        SKIP_BROKEN = "YES"
 
-@pytest.mark.skipif(
-    "localhost" in pytest.hostname, reason="currently broken in Kind CI"
-)
-class TestGen3WorkflowNextflow(TestGen3Workflow):
-    """
-    Nextflow tests are currently broken in the Kind CI.
+        # run python script which includes some of the test cases
+        input_file_name = "test_pv_posix_ops.py"
+        self.gen3_workflow._perform_s3_action(
+            "upload_file",
+            filename=f"test_data/gen3_workflow/{input_file_name}",
+            object_path=f"{self.s3_storage_config.bucket_name}/{self.s3_folder_name}/{input_file_name}",
+            s3_storage_config=self.s3_storage_config,
+            expected_status=None,
+        )
+        s3_path_prefix = f"{self.s3_storage_config.bucket_name}/{self.s3_folder_name}"
+        task_response = self.gen3_workflow.create_tes_task(
+            request_body={
+                "name": "test_pv_posix_ops",
+                "description": "test_pv_posix_ops",
+                "inputs": [
+                    {
+                        "url": f"s3://{s3_path_prefix}/{input_file_name}",
+                        "path": f"/work/{input_file_name}",
+                    }
+                ],
+                "outputs": [
+                    {
+                        "path": "/work/output.txt",
+                        "url": f"s3://{s3_path_prefix}/output.txt",
+                        "type": "FILE",
+                    },
+                    {
+                        "path": (
+                            "/work/moved_output.pdf"
+                            if SKIP_BROKEN == "NO"
+                            else "/work/output.pdf"
+                        ),
+                        "url": f"s3://{s3_path_prefix}/output.pdf",
+                        "type": "FILE",
+                    },
+                ],
+                "executors": [
+                    {
+                        # this image includes python and `pillow`, used for PDF creation
+                        "image": "quay.io/cdis/gen3-workflow:integration_tests_dicom_image",
+                        "workdir": "/work",
+                        "command": [
+                            # test renaming+moving a file (if not SKIP_BROKEN)
+                            (
+                                f"SKIP_BROKEN={SKIP_BROKEN} python3 {input_file_name}"
+                                if SKIP_BROKEN == "YES"
+                                else f"SKIP_BROKEN={SKIP_BROKEN} python3 {input_file_name} && mv output.pdf moved_output.pdf"
+                            ),
+                        ],
+                    }
+                ],
+            },
+            user=self.valid_user,
+            expected_status=200,
+        )
+        task_id = task_response.get("id", None)
+        assert task_id, f"Expected 'id' in response, but got: {task_response}"
+        self.gen3_workflow.poll_until_task_reaches_expected_state(
+            task_id=task_id,
+            user=self.valid_user,
+            expected_final_state="COMPLETE",
+        )
 
-    - Nextflow logs:
-    nextflow.exception.AbortOperationException: Cannot create work-dir 's3://gen3wf-localhost-1/ga4gh-tes' -- Make sure you have write permissions or specify a different directory by using the `-w` command line option
+        # check the contents of the TXT output file
+        out_file_name = "output.txt"
+        response = self.gen3_workflow.get_bucket_object_with_boto3(
+            object_path=f"{s3_path_prefix}/{out_file_name}",
+            s3_storage_config=self.s3_storage_config,
+            user=self.valid_user,
+            expected_status=200,
+        )
+        try:
+            output_file_contents = response["Body"].read().decode("utf-8").strip()
+        except Exception as e:
+            logger.error(
+                f"Failed to read or decode content of {out_file_name} from S3. Error: {e}"
+            )
+            raise
+        if SKIP_BROKEN == "NO":
+            assert (
+                output_file_contents
+                == "Initial sequential data\nSecond sequential data\n"
+            )
 
-    - gen3-workflow logs:
-    Incoming S3 request from user '1': 'PUT gen3wf-localhost-1/ga4gh-tes/'
-    Outgoing S3 request: 'PUT http://minio.gen3-code-vigil-pr-561.svc.cluster.local:9000/gen3wf-localhost-1/ga4gh-tes/'
-    Error from S3: 403 <?xml version="1.0" encoding="UTF-8"?>
-    2026-06-01T23:00:33.9689129Z <Error><Code>SignatureDoesNotMatch</Code><Message>The request signature we calculated does not match the signature you provided. Check your key and signing method.</Message><Key>ga4gh-tes/</Key><BucketName>gen3wf-localhost-1</BucketName><Resource>/gen3wf-localhost-1/ga4gh-tes/</Resource><RequestId>18B5174696300C46</RequestId><HostId>dd9025bab4ad464b049177c95eb6ebf374d3b3fd1af9251148b658df7ac2e3e8</HostId></Error>
-    "PUT /s3/gen3wf-localhost-1/ga4gh-tes/ HTTP/1.0" 403
-    """
+        # check the contents of the PDF output file
+        out_file_name = "output.pdf"
+        response = self.gen3_workflow.get_bucket_object_with_boto3(
+            object_path=f"{s3_path_prefix}/{out_file_name}",
+            s3_storage_config=self.s3_storage_config,
+            user=self.valid_user,
+            expected_status=200,
+        )
+        try:
+            output_file_contents = response["Body"].read()
+        except Exception as e:
+            logger.error(
+                f"Failed to read or decode content of {out_file_name} from S3. Error: {e}"
+            )
+            raise
+        if (
+            SKIP_BROKEN == "NO" or "localhost" in pytest.hostname
+        ):  # Works in Kind, not in Dev env
+            assert b"/Width 100\n/Height 200" in output_file_contents  # page 1
+            assert b"/Width 300\n/Height 400" in output_file_contents  # page 2
+
+        # check the contents of the ZIP output file
+        if SKIP_BROKEN == "NO":
+            out_file_name = "output.zip"
+            response = self.gen3_workflow.get_bucket_object_with_boto3(
+                object_path=f"{s3_path_prefix}/{out_file_name}",
+                s3_storage_config=self.s3_storage_config,
+                user=self.valid_user,
+                expected_status=200,
+            )
+            try:
+                output_file_contents = response["Body"].read().decode("utf-8").strip()
+            except Exception as e:
+                logger.error(
+                    f"Failed to read or decode content of {out_file_name} from S3. Error: {e}"
+                )
+                raise
+            with zipfile.ZipFile(f"test_data/gen3_workflow/{out_file_name}", "r") as f:
+                assert f.namelist() == ["output.txt", "output.pdf"]
 
+        # MIDRC-1298: output files truncated at 8mb
+        if SKIP_BROKEN == "NO":
+            input_content = b"A" * (10 * 1024 * 1024)  # 10MB file
+            s3_path_prefix = (
+                f"{self.s3_storage_config.bucket_name}/{self.s3_folder_name}"
+            )
+            with tempfile.NamedTemporaryFile(delete=True) as file_to_upload:
+                file_to_upload.write(input_content)
+                file_to_upload.flush()
+                self.gen3_workflow._perform_s3_action(
+                    "upload_file",
+                    filename=file_to_upload.name,
+                    object_path=f"{s3_path_prefix}/10mb-file-in.txt",
+                    s3_storage_config=self.s3_storage_config,
+                    expected_status=None,
+                )
+            task_response = self.gen3_workflow.create_tes_task(
+                request_body={
+                    "name": "test_pv_posix_ops",
+                    "description": "test_pv_posix_ops",
+                    "inputs": [
+                        {
+                            "url": f"s3://{s3_path_prefix}/10mb-file-in.txt",
+                            "path": f"/work/10mb-file-in.txt",
+                        }
+                    ],
+                    "outputs": [
+                        {
+                            "path": "/work/10mb-file-out.txt",
+                            "url": f"s3://{s3_path_prefix}/10mb-file-out.txt",
+                            "type": "FILE",
+                        },
+                    ],
+                    "executors": [
+                        {
+                            "image": "public.ecr.aws/docker/library/alpine:latest",
+                            "workdir": "/work",
+                            "command": ["cp 10mb-file-in.txt 10mb-file-out.txt"],
+                        }
+                    ],
+                },
+                user=self.valid_user,
+                expected_status=200,
+            )
+            task_id = task_response.get("id", None)
+            assert task_id, f"Expected 'id' in response, but got: {task_response}"
+            self.gen3_workflow.poll_until_task_reaches_expected_state(
+                task_id=task_id,
+                user=self.valid_user,
+                expected_final_state="COMPLETE",
+            )
+            response = self.gen3_workflow.get_bucket_object_with_boto3(
+                object_path=f"{s3_path_prefix}/10mb-file-out.txt",
+                s3_storage_config=self.s3_storage_config,
+                user=self.valid_user,
+                expected_status=200,
+            )
+            try:
+                output_contents = response["Body"].read().decode("utf-8").strip()
+            except Exception as e:
+                logger.error(
+                    f"Failed to read or decode output file contents from S3. Error: {e}"
+                )
+                raise
+            assert len(input_content) == len(output_contents)
+
+
+class TestGen3WorkflowNextflow(TestGen3Workflow):
     def test_nextflow_workflow(self):
         """
         Test Case: Verify that a Nextflow workflow can be executed successfully.
@@ -1238,6 +1435,7 @@ def test_nextflow_workflow(self):
             assert (
                 task_category in expected_task_outputs
             ), f"Unexpected task name: {task_name}. Expected one of {list(expected_task_outputs.keys())}"
+            expected_task_outputs[task_category]["ran"] = True
 
             assert task["workDir"].startswith(
                 f"{self.s3_storage_config.bucket_name}/"
@@ -1363,6 +1561,11 @@ def get_nextflow_output_file_contents(test_file_name):
                     f"Actual content: `{file_contents}`"
                 }
 
+        for task_category in expected_task_outputs.keys():
+            assert (
+                expected_task_outputs[task_category].get("ran") == True
+            ), f"Expected to see completed '{task_category}' tasks"
+
     @pytest.mark.parametrize(
         "run_gpu_test",
         [
@@ -1384,12 +1587,17 @@ def test_nf_canary(self, run_gpu_test):
         - TEST_MV_FILE and TEST_MV_FOLDER_CONTENTS. Error:
             mv: cannot move 'test.txt' to 'output.txt': Operation not permitted
             -- they are not supported by S3 CSI mount (https://github.com/awslabs/mountpoint-s3/issues/506#issuecomment-1709952359)
+        Note: these work in the Kind CI, where we use Minio instead of AWS S3.
 
         Regression test for TES issues:
         - #40 (support Nextflow "publishDir" directive)
         - #60 (dynamic NodeSelector and Toleration configs to support GPU tasks)
         """
-        known_unsupported = ["TEST_MV_FILE", "TEST_MV_FOLDER_CONTENTS"]
+        known_unsupported = (
+            []
+            if "localhost" in pytest.hostname
+            else ["TEST_MV_FILE", "TEST_MV_FOLDER_CONTENTS"]
+        )
 
         # clone the tests repo
         directory = "test_data/gen3_workflow/nf-canary"

gen3-load-tests/gen3_ci/scripts/setup_ci_env.sh

@@ -153,11 +153,6 @@ yq eval ".ssjdispatcher.gen3Namespace = \"${namespace}\"" -i $manifest_values_ya
 yq eval ".funnel.externalSecrets.dbcreds = \"${namespace}-funnel-creds\"" -i $manifest_values_yaml
 yq eval ".funnel.externalSecrets.funnelOidcClient = \"${namespace}-funnel-oidc-client\"" -i $manifest_values_yaml
 yq eval ".funnel.Kubernetes.JobsNamespace = \"workflow-pods-${namespace}\"" -i $manifest_values_yaml
-# TODO: Remove the next line after funnel helm chart is completely removed as a dependent chart.
-# The legacy chart still expects `funnel.funnel.Kubernetes.JobsNamespace`,
-# while the standalone chart uses `funnel.Kubernetes.JobsNamespace`.
-# Keep both values in sync until the migration is complete.
-yq eval ".funnel.funnel.Kubernetes.JobsNamespace = \"workflow-pods-${namespace}\"" -i $manifest_values_yaml
 sed -i "s|FRAME_ANCESTORS: .*|FRAME_ANCESTORS: https://${HOSTNAME}|" $manifest_values_yaml
 
 # Remove aws-es-proxy block