Support Oracle Easy Connect protocol and parameters

sqlit/domains/connections/providers/oracle/adapter.py

@@ -83,7 +83,18 @@ def connect(self, config: ConnectionConfig) -> Any:
             sid = config.get_option("oracle_sid") or endpoint.database
             dsn = oracledb.makedsn(endpoint.host, port, sid=sid)
         else:
-            dsn = f"{endpoint.host}:{port}/{endpoint.database}"
+            protocol = str(config.get_option("oracle_protocol", "default")).strip().lower()
+            if protocol not in {"", "default", "tcp", "tcps"}:
+                raise ValueError("Oracle protocol must be Default, TCP, or TCPS")
+            protocol_prefix = f"{protocol}://" if protocol in {"tcp", "tcps"} else ""
+            dsn = f"{protocol_prefix}{endpoint.host}:{port}/{endpoint.database}"
+
+            parameters = str(
+                config.get_option("oracle_easy_connect_parameters", "") or ""
+            ).strip()
+            parameters = parameters.lstrip("?")
+            if parameters:
+                dsn = f"{dsn}?{parameters}"
 
         # Determine connection mode based on oracle_role
         oracle_role = config.get_option("oracle_role", "normal")

sqlit/domains/connections/providers/oracle/schema.py

@@ -27,6 +27,14 @@ def _get_oracle_connection_type_options() -> tuple[SelectOption, ...]:
     )
 
 
+def _get_oracle_protocol_options() -> tuple[SelectOption, ...]:
+    return (
+        SelectOption("default", "Default"),
+        SelectOption("tcp", "TCP"),
+        SelectOption("tcps", "TCPS"),
+    )
+
+
 def _oracle_connection_type_is_service_name(values: dict) -> bool:
     return values.get("oracle_connection_type", "service_name") != "sid"
 
@@ -61,6 +69,21 @@ def _oracle_connection_type_is_sid(values: dict) -> bool:
             required=True,
             visible_when=_oracle_connection_type_is_service_name,
         ),
+        SchemaField(
+            name="oracle_protocol",
+            label="Protocol",
+            field_type=FieldType.DROPDOWN,
+            options=_get_oracle_protocol_options(),
+            default="default",
+            visible_when=_oracle_connection_type_is_service_name,
+        ),
+        SchemaField(
+            name="oracle_easy_connect_parameters",
+            label="Easy Connect Parameters",
+            placeholder="ssl_server_dn_match=no&retry_count=3",
+            description="Oracle Easy Connect parameters, without the leading question mark.",
+            visible_when=_oracle_connection_type_is_service_name,
+        ),
         SchemaField(
             name="oracle_sid",
             label="SID",

sqlit/domains/connections/providers/oracle_legacy/schema.py

@@ -27,6 +27,14 @@ def _get_oracle_connection_type_options() -> tuple[SelectOption, ...]:
     )
 
 
+def _get_oracle_protocol_options() -> tuple[SelectOption, ...]:
+    return (
+        SelectOption("default", "Default"),
+        SelectOption("tcp", "TCP"),
+        SelectOption("tcps", "TCPS"),
+    )
+
+
 def _oracle_connection_type_is_service_name(values: dict) -> bool:
     return values.get("oracle_connection_type", "service_name") != "sid"
 
@@ -72,6 +80,21 @@ def _oracle_thick_mode_enabled(values: dict) -> bool:
             required=True,
             visible_when=_oracle_connection_type_is_service_name,
         ),
+        SchemaField(
+            name="oracle_protocol",
+            label="Protocol",
+            field_type=FieldType.DROPDOWN,
+            options=_get_oracle_protocol_options(),
+            default="default",
+            visible_when=_oracle_connection_type_is_service_name,
+        ),
+        SchemaField(
+            name="oracle_easy_connect_parameters",
+            label="Easy Connect Parameters",
+            placeholder="ssl_server_dn_match=no&retry_count=3",
+            description="Oracle Easy Connect parameters, without the leading question mark.",
+            visible_when=_oracle_connection_type_is_service_name,
+        ),
         SchemaField(
             name="oracle_sid",
             label="SID",

tests/integration/test_oracle_easy_connect_cli_config.py

@@ -0,0 +1,42 @@
+"""Regression coverage for Oracle Easy Connect CLI configuration."""
+
+from __future__ import annotations
+
+import argparse
+
+import pytest
+
+from sqlit.domains.connections.cli.helpers import (
+    add_schema_arguments,
+    build_connection_config_from_args,
+)
+from sqlit.domains.connections.providers.catalog import get_provider_schema
+
+
+@pytest.mark.parametrize("db_type", ["oracle", "oracle_legacy"])
+def test_cli_builds_oracle_easy_connect_options_into_config(db_type: str) -> None:
+    schema = get_provider_schema(db_type)
+    parser = argparse.ArgumentParser()
+    add_schema_arguments(parser, schema, include_name=True, name_required=True)
+
+    args = parser.parse_args(
+        [
+            "--name",
+            "secure-oracle",
+            "--server",
+            "localhost",
+            "--database",
+            "service-name.com",
+            "--username",
+            "testuser",
+            "--oracle-protocol",
+            "tcps",
+            "--oracle-easy-connect-parameters",
+            "ssl_server_dn_match=no&retry_count=3",
+        ]
+    )
+
+    config = build_connection_config_from_args(schema, args, name=args.name)
+
+    assert config.get_option("oracle_protocol") == "tcps"
+    assert config.get_option("oracle_easy_connect_parameters") == ("ssl_server_dn_match=no&retry_count=3")

tests/unit/test_oracle_adapter.py

@@ -161,6 +161,132 @@ def test_connect_service_name_format(self):
             # Service name uses slash separator: host:port/service_name
             assert call_kwargs["dsn"] == "localhost:1521/XEPDB1"
 
+    def test_connect_tcps_with_easy_connect_parameters(self):
+        """Issue #261: protocol and parameters must be included in the DSN."""
+        mock_oracledb = MagicMock()
+        mock_oracledb.AUTH_MODE_SYSDBA = 2
+        mock_oracledb.AUTH_MODE_SYSOPER = 4
+
+        with patch.dict("sys.modules", {"oracledb": mock_oracledb}):
+            from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter
+
+            adapter = OracleAdapter()
+            config = ConnectionConfig(
+                name="test",
+                db_type="oracle",
+                server="localhost",
+                port="1521",
+                database="service-name.com",
+                username="testuser",
+                password="testpass",
+                options={
+                    "oracle_connection_type": "service_name",
+                    "oracle_protocol": "tcps",
+                    "oracle_easy_connect_parameters": (
+                        "ssl_server_dn_match=no&retry_count=3"
+                    ),
+                },
+            )
+
+            adapter.connect(config)
+
+            call_kwargs = mock_oracledb.connect.call_args.kwargs
+            assert call_kwargs["dsn"] == (
+                "tcps://localhost:1521/service-name.com"
+                "?ssl_server_dn_match=no&retry_count=3"
+            )
+
+    def test_connect_easy_connect_parameters_accept_leading_question_mark(self):
+        """Users may paste Easy Connect parameters with their separator."""
+        mock_oracledb = MagicMock()
+
+        with patch.dict("sys.modules", {"oracledb": mock_oracledb}):
+            from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter
+
+            config = ConnectionConfig(
+                name="test",
+                db_type="oracle",
+                server="localhost",
+                port="1521",
+                database="XEPDB1",
+                username="testuser",
+                password="testpass",
+                options={"oracle_easy_connect_parameters": "?expire_time=2"},
+            )
+
+            OracleAdapter().connect(config)
+
+            assert mock_oracledb.connect.call_args.kwargs["dsn"] == (
+                "localhost:1521/XEPDB1?expire_time=2"
+            )
+
+    def test_connect_rejects_unknown_oracle_protocol(self):
+        """Configs outside the schema must not silently discard bad protocols."""
+        mock_oracledb = MagicMock()
+
+        with patch.dict("sys.modules", {"oracledb": mock_oracledb}):
+            from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter
+
+            config = ConnectionConfig(
+                name="test",
+                db_type="oracle",
+                server="localhost",
+                port="1521",
+                database="XEPDB1",
+                username="testuser",
+                password="testpass",
+                options={"oracle_protocol": "udp"},
+            )
+
+            with pytest.raises(ValueError, match="Default, TCP, or TCPS"):
+                OracleAdapter().connect(config)
+
+            mock_oracledb.connect.assert_not_called()
+
+    def test_connect_sid_ignores_easy_connect_options(self):
+        """Easy Connect protocol and parameters do not apply to SID descriptors."""
+        mock_oracledb = MagicMock()
+
+        with patch.dict("sys.modules", {"oracledb": mock_oracledb}):
+            from sqlit.domains.connections.providers.oracle.adapter import OracleAdapter
+
+            config = ConnectionConfig(
+                name="test",
+                db_type="oracle",
+                server="localhost",
+                port="1521",
+                username="testuser",
+                password="testpass",
+                options={
+                    "oracle_connection_type": "sid",
+                    "oracle_sid": "ORCL",
+                    "oracle_protocol": "tcps",
+                    "oracle_easy_connect_parameters": "ssl_server_dn_match=no",
+                },
+            )
+
+            OracleAdapter().connect(config)
+
+            mock_oracledb.makedsn.assert_called_once_with("localhost", 1521, sid="ORCL")
+            assert mock_oracledb.connect.call_args.kwargs["dsn"] is (
+                mock_oracledb.makedsn.return_value
+            )
+
+    def test_tcps_easy_connect_dsn_parses_in_real_driver(self):
+        """The issue #261 DSN must be accepted by python-oracledb itself."""
+        oracledb = pytest.importorskip("oracledb")
+        params = oracledb.ConnectParams()
+
+        params.parse_connect_string(
+            "tcps://localhost:1521/service-name.com?ssl_server_dn_match=no"
+        )
+
+        assert params.protocol == "tcps"
+        assert params.host == "localhost"
+        assert params.port == 1521
+        assert params.service_name == "service-name.com"
+        assert params.ssl_server_dn_match is False
+
     def test_connect_sid_format(self):
         """SID connection type must go through oracledb.makedsn — see issue #106.