guke
88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
"""scripts/ensure_pg.py 纯函数单测(不需要 Docker/PG)。"""
|
|
from __future__ import annotations
|
|
|
|
import socket
|
|
|
|
from scripts.ensure_pg import (
|
|
_docker_desktop_cmd,
|
|
_is_sqlite,
|
|
_parse_host_port,
|
|
_port_open,
|
|
ensure,
|
|
)
|
|
|
|
|
|
def test_is_sqlite():
|
|
assert _is_sqlite("sqlite:///./data/app.db")
|
|
assert _is_sqlite(" SQLite:///x ")
|
|
assert not _is_sqlite("postgresql+psycopg://u:p@localhost:5432/db")
|
|
|
|
|
|
def test_parse_host_port_full():
|
|
assert _parse_host_port(
|
|
"postgresql+psycopg://u:p@localhost:5432/shaguabijia"
|
|
) == ("localhost", 5432)
|
|
|
|
|
|
def test_parse_host_port_defaults():
|
|
# 缺端口 → 5432
|
|
assert _parse_host_port("postgresql+psycopg://u:p@db.example/x")[1] == 5432
|
|
# 缺 host → localhost
|
|
assert _parse_host_port("postgresql+psycopg:///x") == ("localhost", 5432)
|
|
|
|
|
|
def test_parse_host_port_testdb():
|
|
assert _parse_host_port(
|
|
"postgresql+psycopg://u:p@localhost:5432/shaguabijia_test"
|
|
) == ("localhost", 5432)
|
|
|
|
|
|
def test_port_open_true():
|
|
srv = socket.socket()
|
|
srv.bind(("127.0.0.1", 0))
|
|
srv.listen(1)
|
|
port = srv.getsockname()[1]
|
|
try:
|
|
assert _port_open("127.0.0.1", port, timeout=1.0)
|
|
finally:
|
|
srv.close()
|
|
|
|
|
|
def test_port_open_false():
|
|
s = socket.socket()
|
|
s.bind(("127.0.0.1", 0))
|
|
port = s.getsockname()[1]
|
|
s.close() # 释放端口,无人监听 → 连接应失败
|
|
assert not _port_open("127.0.0.1", port, timeout=0.3)
|
|
|
|
|
|
def test_docker_desktop_cmd_windows():
|
|
cmd = _docker_desktop_cmd("win32", r"C:\Program Files")
|
|
assert cmd is not None
|
|
assert cmd[0].endswith("Docker Desktop.exe")
|
|
assert "Docker" in cmd[0]
|
|
|
|
|
|
def test_docker_desktop_cmd_darwin():
|
|
assert _docker_desktop_cmd("darwin", "") == ["open", "-a", "Docker"]
|
|
|
|
|
|
def test_docker_desktop_cmd_linux():
|
|
assert _docker_desktop_cmd("linux", "") is None
|
|
|
|
|
|
def test_ensure_rejects_sqlite():
|
|
# dev 守卫:sqlite 直接 False(不碰 Docker)
|
|
assert ensure("sqlite:///./data/app.db") is False
|
|
|
|
|
|
def test_ensure_shortcircuits_when_pg_up(monkeypatch):
|
|
# 端口通 → 直接 True,绝不触碰 docker
|
|
monkeypatch.setattr("scripts.ensure_pg._port_open", lambda *a, **k: True)
|
|
|
|
def _boom():
|
|
raise AssertionError("端口通时不应调用 docker")
|
|
|
|
monkeypatch.setattr("scripts.ensure_pg._docker_cli_ok", _boom)
|
|
assert ensure("postgresql+psycopg://u:p@localhost:5432/shaguabijia") is True
|