diff --git a/sqlit/domains/connections/providers/oracle/adapter.py b/sqlit/domains/connections/providers/oracle/adapter.py index 76edc895..a79f5b27 100644 --- a/sqlit/domains/connections/providers/oracle/adapter.py +++ b/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") diff --git a/sqlit/domains/connections/providers/oracle/schema.py b/sqlit/domains/connections/providers/oracle/schema.py index f004a6d9..7aee63c9 100644 --- a/sqlit/domains/connections/providers/oracle/schema.py +++ b/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", diff --git a/sqlit/domains/connections/providers/oracle_legacy/schema.py b/sqlit/domains/connections/providers/oracle_legacy/schema.py index 4252f721..1cac2353 100644 --- a/sqlit/domains/connections/providers/oracle_legacy/schema.py +++ b/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", diff --git a/tests/integration/test_oracle_easy_connect_cli_config.py b/tests/integration/test_oracle_easy_connect_cli_config.py new file mode 100644 index 00000000..c9e74faf --- /dev/null +++ b/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") diff --git a/tests/unit/test_oracle_adapter.py b/tests/unit/test_oracle_adapter.py index 7693b76e..5c02a36e 100644 --- a/tests/unit/test_oracle_adapter.py +++ b/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.