|
1 | | -import subprocess |
2 | 1 | import getpass |
3 | | -import pytest |
| 2 | +import subprocess |
| 3 | + |
4 | 4 | import psycopg2 |
| 5 | +import pytest |
5 | 6 | from psycopg2 import OperationalError |
6 | 7 |
|
7 | 8 | # Import all functions to test |
8 | 9 | from codes.tune import ( |
9 | | - _make_db_url, |
10 | | - _check_remote_reachable, |
11 | 10 | _check_postgres_running_local, |
12 | | - _start_postgres_server_local, |
| 11 | + _check_remote_reachable, |
13 | 12 | _initialize_postgres_local, |
14 | 13 | _initialize_postgres_remote, |
| 14 | + _make_db_url, |
| 15 | + _start_postgres_server_local, |
15 | 16 | initialize_optuna_database, |
16 | 17 | ) |
17 | 18 |
|
@@ -194,6 +195,36 @@ def test_initialize_postgres_remote_interactive(monkeypatch): |
194 | 195 | monkeypatch.setattr( |
195 | 196 | "codes.tune.postgres_fcts._check_remote_reachable", lambda conf: None |
196 | 197 | ) |
| 198 | + |
| 199 | + # fake psycopg2 connection/cursor to avoid real network calls |
| 200 | + def fake_connect(**kwargs): |
| 201 | + class FakeCursor: |
| 202 | + def execute(self, query, params=None): |
| 203 | + # accept any SQL (CREATE/DROP/SELECT) |
| 204 | + self._last_query = query |
| 205 | + |
| 206 | + def fetchone(self): |
| 207 | + # simulate "schema does not exist" |
| 208 | + return None |
| 209 | + |
| 210 | + def close(self): |
| 211 | + pass |
| 212 | + |
| 213 | + class FakeConn: |
| 214 | + def __init__(self): |
| 215 | + self.autocommit = False |
| 216 | + |
| 217 | + def cursor(self): |
| 218 | + return FakeCursor() |
| 219 | + |
| 220 | + def close(self): |
| 221 | + pass |
| 222 | + |
| 223 | + return FakeConn() |
| 224 | + |
| 225 | + # patch the connect used inside the module under test |
| 226 | + monkeypatch.setattr("codes.tune.postgres_fcts.psycopg2.connect", fake_connect) |
| 227 | + |
197 | 228 | url = _initialize_postgres_remote(cfg, "ignored") |
198 | 229 | assert "sslmode=require" in url |
199 | 230 |
|
|
0 commit comments