Raise an error with no params, accept keywords only.
@classmethod
def connect(
- cls, conninfo: str, connection_factory: Any = None, **kwargs: Any
+ cls, conninfo: Optional[str] = None, **kwargs: Any,
) -> "Connection":
- if connection_factory is not None:
- raise NotImplementedError()
- conninfo = make_conninfo(conninfo, **kwargs)
+ if conninfo is None and not kwargs:
+ raise TypeError("missing conninfo and not parameters specified")
+ conninfo = make_conninfo(conninfo or "", **kwargs)
gen = generators.connect(conninfo)
pgconn = cls.wait(gen)
return cls(pgconn)
self.cursor_factory = cursor.AsyncCursor
@classmethod
- async def connect(cls, conninfo: str = "", **kwargs: Any) -> "AsyncConnection":
- conninfo = make_conninfo(conninfo, **kwargs)
+ async def connect(
+ cls, conninfo: Optional[str] = None, **kwargs: Any
+ ) -> "AsyncConnection":
+ if conninfo is None and not kwargs:
+ raise TypeError("missing conninfo and not parameters specified")
+ conninfo = make_conninfo(conninfo or "", **kwargs)
gen = generators.connect(conninfo)
pgconn = await cls.wait(gen)
return cls(pgconn)
import psycopg3
from psycopg3 import AsyncConnection
+from psycopg3.conninfo import conninfo_to_dict
def test_connect(dsn, loop):
def test_set_encoding_bad(aconn, loop):
with pytest.raises(psycopg3.DatabaseError):
loop.run_until_complete(aconn.set_client_encoding("WAT"))
+
+
+@pytest.mark.parametrize(
+ "testdsn, kwargs, want",
+ [
+ ("", {}, ""),
+ ("host=foo user=bar", {}, "host=foo user=bar"),
+ ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ (
+ "host=foo port=5432",
+ {"host": "qux", "user": "joe"},
+ "host=qux user=joe port=5432",
+ ),
+ ("host=foo", {"user": None}, "host=foo"),
+ ],
+)
+def test_connect_args(monkeypatch, pgconn, loop, testdsn, kwargs, want):
+ the_conninfo = None
+
+ def fake_connect(conninfo):
+ nonlocal the_conninfo
+ the_conninfo = conninfo
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ loop.run_until_complete(
+ psycopg3.AsyncConnection.connect(testdsn, **kwargs)
+ )
+ assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
+
+
+@pytest.mark.parametrize(
+ "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})],
+)
+def test_connect_badargs(monkeypatch, pgconn, loop, args, kwargs):
+ def fake_connect(conninfo):
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ with pytest.raises((TypeError, psycopg3.ProgrammingError)):
+ loop.run_until_complete(
+ psycopg3.AsyncConnection.connect(*args, **kwargs)
+ )
import psycopg3
from psycopg3 import Connection
+from psycopg3.conninfo import conninfo_to_dict
def test_connect(dsn):
def test_set_encoding_bad(conn):
with pytest.raises(psycopg3.DatabaseError):
conn.set_client_encoding("WAT")
+
+
+@pytest.mark.parametrize(
+ "testdsn, kwargs, want",
+ [
+ ("", {}, ""),
+ ("host=foo user=bar", {}, "host=foo user=bar"),
+ ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ (
+ "host=foo port=5432",
+ {"host": "qux", "user": "joe"},
+ "host=qux user=joe port=5432",
+ ),
+ ("host=foo", {"user": None}, "host=foo"),
+ ],
+)
+def test_connect_args(monkeypatch, pgconn, testdsn, kwargs, want):
+ the_conninfo = None
+
+ def fake_connect(conninfo):
+ nonlocal the_conninfo
+ the_conninfo = conninfo
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ psycopg3.Connection.connect(testdsn, **kwargs)
+ assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
+
+
+@pytest.mark.parametrize(
+ "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})],
+)
+def test_connect_badargs(monkeypatch, pgconn, args, kwargs):
+ def fake_connect(conninfo):
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ with pytest.raises((TypeError, psycopg3.ProgrammingError)):
+ psycopg3.Connection.connect(*args, **kwargs)
import datetime as dt
import psycopg3
+from psycopg3.conninfo import conninfo_to_dict
from . import dbapi20
s = psycopg3.TimeFromTicks(ticks)
want = dt.datetime.strptime(want, "%H:%M:%S.%f").time()
assert s.replace(hour=0) == want
+
+
+@pytest.mark.parametrize(
+ "testdsn, kwargs, want",
+ [
+ ("", {}, ""),
+ ("host=foo user=bar", {}, "host=foo user=bar"),
+ ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ (
+ "host=foo port=5432",
+ {"host": "qux", "user": "joe"},
+ "host=qux user=joe port=5432",
+ ),
+ ("host=foo", {"user": None}, "host=foo"),
+ ],
+)
+def test_connect_args(monkeypatch, pgconn, testdsn, kwargs, want):
+ the_conninfo = None
+
+ def fake_connect(conninfo):
+ nonlocal the_conninfo
+ the_conninfo = conninfo
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ psycopg3.connect(testdsn, **kwargs)
+ assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
+
+
+@pytest.mark.parametrize(
+ "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})],
+)
+def test_connect_badargs(monkeypatch, pgconn, args, kwargs):
+ def fake_connect(conninfo):
+ return pgconn
+ yield
+
+ monkeypatch.setattr(psycopg3.generators, "connect", fake_connect)
+ with pytest.raises((TypeError, psycopg3.ProgrammingError)):
+ psycopg3.connect(*args, **kwargs)