| 124 | |
| 125 | |
| 126 | def test_config(tmpdir: os.PathLike, mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicMock) -> None: |
| 127 | pgclirc = str(tmpdir.join("rcfile")) |
| 128 | |
| 129 | tunnel_user = "tunnel_user" |
| 130 | tunnel_passwd = "tunnel_pass" |
| 131 | tunnel_host = "tunnel.host" |
| 132 | tunnel_port = 1022 |
| 133 | tunnel_url = f"{tunnel_user}:{tunnel_passwd}@{tunnel_host}:{tunnel_port}" |
| 134 | |
| 135 | tunnel2_url = "tunnel2.host" |
| 136 | |
| 137 | config = ConfigObj() |
| 138 | config.filename = pgclirc |
| 139 | config["ssh tunnels"] = {} |
| 140 | config["ssh tunnels"][r"\.com$"] = tunnel_url |
| 141 | config["ssh tunnels"][r"^hello-"] = tunnel2_url |
| 142 | config.write() |
| 143 | |
| 144 | # Unmatched host |
| 145 | pgcli = PGCli(pgclirc_file=pgclirc) |
| 146 | pgcli.connect(host="unmatched.host") |
| 147 | mock_ssh_tunnel_forwarder.assert_not_called() |
| 148 | |
| 149 | # Host matching first tunnel |
| 150 | pgcli = PGCli(pgclirc_file=pgclirc) |
| 151 | pgcli.connect(host="matched.host.com") |
| 152 | mock_ssh_tunnel_forwarder.assert_called_once() |
| 153 | call_args, call_kwargs = mock_ssh_tunnel_forwarder.call_args |
| 154 | assert call_kwargs["ssh_address_or_host"] == (tunnel_host, tunnel_port) |
| 155 | assert call_kwargs["ssh_username"] == tunnel_user |
| 156 | assert call_kwargs["ssh_password"] == tunnel_passwd |
| 157 | mock_ssh_tunnel_forwarder.reset_mock() |
| 158 | |
| 159 | # Host matching second tunnel |
| 160 | pgcli = PGCli(pgclirc_file=pgclirc) |
| 161 | pgcli.connect(host="hello-i-am-matched") |
| 162 | mock_ssh_tunnel_forwarder.assert_called_once() |
| 163 | |
| 164 | call_args, call_kwargs = mock_ssh_tunnel_forwarder.call_args |
| 165 | assert call_kwargs["ssh_address_or_host"] == (tunnel2_url, 22) |
| 166 | mock_ssh_tunnel_forwarder.reset_mock() |
| 167 | |
| 168 | # Host matching both tunnels (will use the first one matched) |
| 169 | pgcli = PGCli(pgclirc_file=pgclirc) |
| 170 | pgcli.connect(host="hello-i-am-matched.com") |
| 171 | mock_ssh_tunnel_forwarder.assert_called_once() |
| 172 | |
| 173 | call_args, call_kwargs = mock_ssh_tunnel_forwarder.call_args |
| 174 | assert call_kwargs["ssh_address_or_host"] == (tunnel_host, tunnel_port) |
| 175 | assert call_kwargs["ssh_username"] == tunnel_user |
| 176 | assert call_kwargs["ssh_password"] == tunnel_passwd |