import datetime
import warnings

import pytest

import pg8000


class PG8000TestException(Exception):
    pass


def raise_exception(val):
    raise PG8000TestException("oh noes!")


def test_py_value_fail(con, mocker):
    # Ensure that if types.py_value throws an exception, the original
    # exception is raised (PG8000TestException), and the connection is
    # still usable after the error.
    mocker.patch.object(con, "py_types")
    con.py_types = {datetime.time: raise_exception}

    with con.cursor() as c, pytest.raises(PG8000TestException):
        c.execute("SELECT CAST(%s AS TIME)", (datetime.time(10, 30),))
        c.fetchall()

        # ensure that the connection is still usable for a new query
        c.execute("VALUES ('hw3'::text)")
        assert c.fetchone()[0] == "hw3"


def test_no_data_error_recovery(con):
    for i in range(1, 4):
        with con.cursor() as c, pytest.raises(pg8000.ProgrammingError) as e:
            c.execute("DROP TABLE t1")
        assert e.value.args[0]["C"] == "42P01"
        con.rollback()


def testClosedConnection(db_kwargs):
    warnings.simplefilter("ignore")
    my_db = pg8000.connect(**db_kwargs)
    cursor = my_db.cursor()
    my_db.close()
    with pytest.raises(my_db.InterfaceError, match="connection is closed"):
        cursor.execute("VALUES ('hw1'::text)")

    warnings.resetwarnings()