fix

Author: etiennearnal

src/ansys/tools/common/cyberchannel.py

@@ -190,19 +190,16 @@ def create_insecure_channel(
 
 
 def create_uds_channel(
-    uds_fullpath: str | Path | None = None,
     uds_service: str | None = None,
     uds_dir: str | Path | None = None,
     uds_id: str | None = None,
     grpc_options: list[tuple[str, object]] | None = None,
+    uds_fullpath: str | Path | None = None,
 ) -> grpc.Channel:
     """Create a gRPC channel using Unix Domain Sockets (UDS).
 
     Parameters
     ----------
-    uds_fullpath : str | Path | None
-        Full path to the UDS socket file.
-        By default `None` and thus it will use the `uds_service`, `uds_dir` and `uds_id` parameters.
     uds_service : str
         Service name for the UDS socket.
     uds_dir : str | Path | None
@@ -216,6 +213,9 @@ def create_uds_channel(
         gRPC channel options to pass when creating the channel.
         Each option is a tuple of the form ("option_name", value).
         By default `None` and thus only the default authority option is added.
+    uds_fullpath : str | Path | None
+        Full path to the UDS socket file.
+        By default `None` and thus it will use the `uds_service`, `uds_dir` and `uds_id` parameters.
 
     Returns
     -------
@@ -489,10 +489,10 @@ def verify_transport_mode(transport_mode: str, mode: str | None = None) -> None:
 
 
 def verify_uds_socket(
-    uds_fullpath: str | Path | None = None,
     uds_service: str | None = None,
     uds_dir: Path | None = None,
     uds_id: str | None = None,
+    uds_fullpath: str | Path | None = None,
 ) -> bool:
     """Verify that the UDS socket file has been created.
 
@@ -507,6 +507,9 @@ def verify_uds_socket(
         Unique identifier for the UDS socket (optional).
         By default `None` and thus it will use "<uds_service>.sock".
         Otherwise, the socket filename will be "<uds_service>-<uds_id>.sock".
+    uds_fullpath : str | Path | None
+        Full path to the UDS socket file.
+        By default `None` and thus it will use the `uds_service`, `uds_dir` and `uds_id` parameters.
 
     Returns
     -------

tests/test_cyberchannel.py

@@ -0,0 +1,82 @@
+# Copyright (C) 2025 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+"""Tests for cyberchannel."""
+
+from pathlib import Path
+import tempfile
+
+import pytest
+
+from ansys.tools.common import cyberchannel
+
+
+def test_version_tuple():
+    """Test version tuple."""
+    assert cyberchannel.version_tuple("1.2.3") == (1, 2, 3)
+    assert cyberchannel.version_tuple("1.2.3.4") == (1, 2, 3, 4)
+    assert cyberchannel.version_tuple("1.0.0") == (1, 0, 0)
+
+
+def test_cyberchannel_functions():
+    """Test cyberchannel functions."""
+    assert cyberchannel.check_grpc_version()
+    assert cyberchannel.is_uds_supported()
+    uds_path = cyberchannel.determine_uds_folder()
+    uds_path.mkdir(parents=True, exist_ok=True)
+    assert uds_path.is_dir()
+    assert uds_path.exists()
+    uds_path.rmdir()
+    cyberchannel.verify_transport_mode(transport_mode="insecure", mode="local")
+    with pytest.raises(ValueError):
+        cyberchannel.verify_transport_mode(transport_mode="invalid_mode", mode="mode1")
+
+
+def test_cyberchannel_insecure():
+    """Test cyberchannel insecure."""
+    ch = cyberchannel.create_insecure_channel(host="localhost", port=12345)
+    assert ch is not None
+    assert ch._channel.target().decode() == "dns:///localhost:12345"
+    assert not ch.close()
+
+
+def test_cyberchannel_wnua():
+    """Test cyberchannel wnua."""
+    ch = cyberchannel.create_wnua_channel(host="localhost", port=12345)
+    assert ch is not None
+    assert ch._channel.target().decode() == "dns:///localhost:12345"
+    assert not ch.close()
+
+
+def test_cyberchannel_uds():
+    """Test cyberchannel uds."""
+    uds_file = Path(tempfile.gettempdir()) / "test_uds.sock"
+    with uds_file.open("w"):
+        pass
+    ch = cyberchannel.create_uds_channel(uds_fullpath=uds_file)
+    assert ch is not None
+    assert ch._channel.target().decode() == f"unix:{uds_file}"
+    assert not ch.close()
+
+    ch = cyberchannel.create_uds_channel("service_name")
+    assert ch is not None
+    assert ch._channel.target().decode() == f"unix:{cyberchannel.determine_uds_folder() / 'service_name.sock'}"
+    assert not ch.close()