Further improve command quoting

[GlassOfWhiskey]

This commit further improves command quoting in StreamFlow Command and Connector classes by properly using mslex and shlex utilities.

[codecov]

❌ 3 Tests Failed:

Tests completed	Failed	Passed	Skipped
1876	3	1873	8

View the top 3 failed test(s) by shortest run time

tests/test_remotepath.py::test_directory[kubernetes]

Stack Traces | 0.142s run time

context = <streamflow.core.context.StreamFlowContext object at 0x7fe950279d30>
connector = <streamflow.deployment.connector.kubernetes.KubernetesConnector object at 0x7fe94f95cd70>
location = <streamflow.core.deployment.ExecutionLocation object at 0x7fe94d2b4120>

    @pytest.mark.asyncio
    async def test_directory(
        context: StreamFlowContext, connector: Connector, location: ExecutionLocation
    ) -> None:
        """Test directory creation and deletion."""
        path = StreamFlowPath(
            tempfile.gettempdir() if location.local else "/tmp",
            utils.random_name(),
            context=context,
            location=location,
        )
        try:
            await path.mkdir(mode=0o777)
            assert await path.exists()
            assert await path.is_dir()
            # ./
            #   file1.txt
            #   file2.csv
            #   file with spaces.txt
            #   dir1/
            #   dir2/
            await (path / "dir1").mkdir(mode=0o777)
            await (path / "dir2").mkdir(mode=0o777)
            await (path / "file1.txt").write_text("StreamFlow")
            await (path / "file2.csv").write_text("StreamFlow")
>           await (path / "file with spaces.txt").write_text("StreamFlow")

tests/test_remotepath.py:55: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
streamflow/data/remotepath.py:889: in write_text
    async with await self.connector.get_stream_writer(
.../deployment/connector/kubernetes.py:123: in __aexit__
    await self.response.close()
streamflow/deployment/stream.py:25: in close
    await self._close()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <streamflow.deployment.connector.kubernetes.KubernetesResponseWriterWrapper object at 0x7fe94fe7ac40>

    async def _close(self) -> None:
        await self.stream.send_bytes(
            STREAM_CLOSE + bytes(chr(ws_client.STDIN_CHANNEL), "ascii")
        )
        while True:
            msg = await self.stream.receive()
            if payload := msg.data[1:]:
                result = json.loads(payload.decode("utf-8"))
                if result["status"] == "Success":
                    await self.stream.close()
                    break
                else:
>                   raise WorkflowExecutionException(
                        f"Kubernetes connection terminated with status {result['status']}: "
                        f"{result.get('reason', result.get( 'message', '' ))}"
                    )
E                   streamflow.core.exception.WorkflowExecutionException: Kubernetes connection terminated with status Failure: NonZeroExitCode

.../deployment/connector/kubernetes.py:189: WorkflowExecutionException

tests/test_remotepath.py::test_glob[kubernetes]

Stack Traces | 0.228s run time

context = <streamflow.core.context.StreamFlowContext object at 0x7fe950279d30>
connector = <streamflow.deployment.connector.kubernetes.KubernetesConnector object at 0x7fe94f95cd70>
location = <streamflow.core.deployment.ExecutionLocation object at 0x7fe94d2b4120>

    @pytest.mark.asyncio
    async def test_glob(
        context: StreamFlowContext, connector: Connector, location: ExecutionLocation
    ) -> None:
        """Test glob resolution."""
        path = StreamFlowPath(
            tempfile.gettempdir() if location.local else "/tmp",
            utils.random_name(),
            context=context,
            location=location,
        )
        try:
            await path.mkdir(mode=0o777)
            # ./
            #   file1.txt
            #   file2.csv
            #   dir1/
            #     file1.txt
            #     file2.csv
            #     file with spaces.txt
            #     dir2/
            #       file1.txt
            #       file2.csv
            await (path / "file1.txt").write_text("StreamFlow")
            await (path / "file2.csv").write_text("StreamFlow")
            await (path / "dir1" / "dir2").mkdir(mode=0o777, parents=True)
            await (path / "dir1" / "file1.txt").write_text("StreamFlow")
            await (path / "dir1" / "file2.csv").write_text("StreamFlow")
>           await (path / "dir1" / "file with spaces.txt").write_text("StreamFlow")

tests/test_remotepath.py:164: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
streamflow/data/remotepath.py:889: in write_text
    async with await self.connector.get_stream_writer(
.../deployment/connector/kubernetes.py:123: in __aexit__
    await self.response.close()
streamflow/deployment/stream.py:25: in close
    await self._close()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <streamflow.deployment.connector.kubernetes.KubernetesResponseWriterWrapper object at 0x7fe94d173020>

    async def _close(self) -> None:
        await self.stream.send_bytes(
            STREAM_CLOSE + bytes(chr(ws_client.STDIN_CHANNEL), "ascii")
        )
        while True:
            msg = await self.stream.receive()
            if payload := msg.data[1:]:
                result = json.loads(payload.decode("utf-8"))
                if result["status"] == "Success":
                    await self.stream.close()
                    break
                else:
>                   raise WorkflowExecutionException(
                        f"Kubernetes connection terminated with status {result['status']}: "
                        f"{result.get('reason', result.get( 'message', '' ))}"
                    )
E                   streamflow.core.exception.WorkflowExecutionException: Kubernetes connection terminated with status Failure: NonZeroExitCode

.../deployment/connector/kubernetes.py:189: WorkflowExecutionException

cwl-v1.3-6eddbfb03b8861885832b26e322fbd0fcb9a884d/conformance_tests.cwltest.yaml::conformance_tests::cwltest::yaml::output_secondaryfile_optional

Stack Traces | 2.46s run time

CWL test execution failed. 
Returned non-zero but it should be zero
Test: tool: 
  file:.../streamflow/streamflow/cwl-v1.3-6eddbfb03b8861885832b26e322fbd0fcb9a884d/tests/optional-output.cwl
job: 
  file:.../streamflow/streamflow/cwl-v1.3-6eddbfb03b8861885832b26e322fbd0fcb9a884d/tests/cat-job.json
output:
  optional_file:
  output_file:
    location: output.txt
    size: 13
    class: File
    checksum: sha1$47a013e660d408619d894b20806b1d5086aab03b
id: 
  file:.../streamflow/streamflow/cwl-v1.3-6eddbfb03b8861885832b26e322fbd0fcb9a884d/conformance_tests.cwltest.yaml#output_secondaryfile_optional
doc: Test optional output file and optional secondaryFile on output.
tags:
- docker
- command_line_tool
line: '860'

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}