# -*- 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()