@@ -813,6 +813,21 @@ def fake_connect_sandbox(self, conn):
813813 assert executor .connection_id is None
814814
815815
816+ def test_connect_reraises_non_sandbox_operational_error (monkeypatch ) -> None :
817+ executor = make_executor_for_connect_tests ()
818+ executor .ssl = None
819+
820+ def fake_connect (** _kwargs ):
821+ raise pymysql .OperationalError (1045 , 'access denied' )
822+
823+ monkeypatch .setattr (sqlexecute .pymysql , 'connect' , fake_connect )
824+
825+ with pytest .raises (pymysql .OperationalError ) as exc_info :
826+ executor .connect ()
827+
828+ assert exc_info .value .args == (1045 , 'access denied' )
829+
830+
816831def test_connect_uses_ssh_tunnel_when_ssh_host_is_set (monkeypatch ) -> None :
817832 executor = make_executor_for_connect_tests ()
818833 executor .ssl = None
@@ -911,6 +926,29 @@ def start(self) -> None:
911926 executor .connect (ssh_host = 'bastion.internal' )
912927
913928
929+ def test_connect_sandbox_temporarily_disables_set_character_set () -> None :
930+ original_calls = []
931+ connect_observed_stub = []
932+
933+ class FakeSandboxConnection :
934+ def set_character_set (self , * args , ** kwargs ) -> None :
935+ original_calls .append ((args , kwargs ))
936+
937+ def connect (self ) -> None :
938+ self .set_character_set ('utf8mb4' )
939+ connect_observed_stub .append (original_calls == [])
940+
941+ conn = FakeSandboxConnection ()
942+ original_set_character_set = conn .set_character_set
943+
944+ SQLExecute ._connect_sandbox (conn )
945+
946+ assert connect_observed_stub == [True ]
947+ assert conn .set_character_set == original_set_character_set
948+ conn .set_character_set ('latin1' )
949+ assert original_calls == [(('latin1' ,), {})]
950+
951+
914952def test_run_returns_empty_result_for_blank_statement (monkeypatch ) -> None :
915953 split_inputs : list [str ] = []
916954
0 commit comments