| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475 |
- # -*- coding:utf-8 -*-
- """dw_base.io.db.postgresql 连接工厂 + 查询封装单测（不连真 PG）。"""
- import sys
- from unittest.mock import MagicMock
- import pytest
- from dw_base.io.db import postgresql as pgdb
- def test_parse_jdbc_url_with_port():
- host, port, db = pgdb.parse_jdbc_url('jdbc:postgresql://10.0.0.1:5433/hobby_stocks')
- assert host == '10.0.0.1'
- assert port == 5433
- assert db == 'hobby_stocks'
- def test_parse_jdbc_url_default_port():
- host, port, db = pgdb.parse_jdbc_url('jdbc:postgresql://pg.example.com/mydb')
- assert host == 'pg.example.com'
- assert port == 5432
- assert db == 'mydb'
- def test_parse_jdbc_url_invalid():
- with pytest.raises(ValueError, match='无法解析'):
- pgdb.parse_jdbc_url('mysql://10.0.0.1:3306/foo')
- def test_resolve_returns_conn_params(monkeypatch):
- fake_ds = MagicMock()
- fake_ds.parse.return_value = {
- pgdb.DS_POSTGRE_SQL_JDBC_URL: 'jdbc:postgresql://10.0.0.1:5432/mydb',
- 'username': 'u',
- 'password': 'p',
- }
- monkeypatch.setattr(pgdb, '_resolve_datasource', lambda ref: fake_ds)
- assert pgdb.resolve('postgresql/prd-hobby') == {
- 'host': '10.0.0.1', 'port': 5432, 'database': 'mydb',
- 'username': 'u', 'password': 'p',
- }
- def test_connect_passes_resolved_params(monkeypatch):
- monkeypatch.setattr(pgdb, 'resolve', lambda ref: {
- 'host': 'h', 'port': 5432, 'database': 'mydb',
- 'username': 'u', 'password': 'p',
- })
- fake_pg8000 = MagicMock()
- monkeypatch.setitem(sys.modules, 'pg8000', fake_pg8000)
- monkeypatch.setitem(sys.modules, 'pg8000.dbapi', fake_pg8000.dbapi)
- pgdb.connect('postgresql/prd-hobby')
- fake_pg8000.dbapi.connect.assert_called_once_with(
- host='h', port=5432, database='mydb', user='u', password='p',
- )
- def test_query_with_params():
- conn = MagicMock()
- cur = conn.cursor.return_value
- cur.fetchall.return_value = [(1, 'a'), (2, 'b')]
- rows = pgdb.query(conn, 'SELECT 1 WHERE x = %s', ('v',))
- assert rows == [(1, 'a'), (2, 'b')]
- cur.execute.assert_called_once_with('SELECT 1 WHERE x = %s', ('v',))
- cur.close.assert_called_once()
- def test_query_without_params():
- conn = MagicMock()
- cur = conn.cursor.return_value
- cur.fetchall.return_value = []
- pgdb.query(conn, 'SELECT 1')
- cur.execute.assert_called_once_with('SELECT 1')
- cur.close.assert_called_once()
|
