from .pq_enums import (
ConnStatus,
- PostgresPollingStatus,
+ PollingStatus,
TransactionStatus,
- PGPing,
+ Ping,
)
from . import _pq_ctypes as impl
def connect_poll(self):
rv = impl.PQconnectPoll(self.pgconn_ptr)
- return PostgresPollingStatus(rv)
+ return PollingStatus(rv)
def finish(self):
self.pgconn_ptr, p = None, self.pgconn_ptr
def reset_poll(self):
rv = impl.PQresetPoll(self.pgconn_ptr)
- return PostgresPollingStatus(rv)
+ return PollingStatus(rv)
@classmethod
def ping(self, conninfo):
raise TypeError("bytes expected, got %r instead" % conninfo)
rv = impl.PQping(conninfo)
- return PGPing(rv)
+ return Ping(rv)
@property
def db(self):
import pytest
-from psycopg3.pq_enums import ConnStatus, PostgresPollingStatus, PGPing
+from psycopg3.pq_enums import ConnStatus, PollingStatus, Ping
def test_connectdb(pq, dsn):
while 1:
assert conn.status != ConnStatus.CONNECTION_BAD
rv = conn.connect_poll()
- if rv == PostgresPollingStatus.PGRES_POLLING_OK:
+ if rv == PollingStatus.PGRES_POLLING_OK:
break
- elif rv == PostgresPollingStatus.PGRES_POLLING_READING:
+ elif rv == PollingStatus.PGRES_POLLING_READING:
select([conn.socket], [], [])
- elif rv == PostgresPollingStatus.PGRES_POLLING_WRITING:
+ elif rv == PollingStatus.PGRES_POLLING_WRITING:
select([], [conn.socket], [])
else:
assert False, rv
while 1:
assert conn.status != ConnStatus.CONNECTION_BAD
rv = conn.connect_poll()
- if rv == PostgresPollingStatus.PGRES_POLLING_FAILED:
+ if rv == PollingStatus.PGRES_POLLING_FAILED:
break
- elif rv == PostgresPollingStatus.PGRES_POLLING_READING:
+ elif rv == PollingStatus.PGRES_POLLING_READING:
select([conn.socket], [], [])
- elif rv == PostgresPollingStatus.PGRES_POLLING_WRITING:
+ elif rv == PollingStatus.PGRES_POLLING_WRITING:
select([], [conn.socket], [])
else:
assert False, rv
pgconn.reset_start()
while 1:
rv = pgconn.connect_poll()
- if rv == PostgresPollingStatus.PGRES_POLLING_READING:
+ if rv == PollingStatus.PGRES_POLLING_READING:
select([pgconn.socket], [], [])
- elif rv == PostgresPollingStatus.PGRES_POLLING_WRITING:
+ elif rv == PollingStatus.PGRES_POLLING_WRITING:
select([], [pgconn.socket], [])
else:
break
- assert rv == PostgresPollingStatus.PGRES_POLLING_OK
+ assert rv == PollingStatus.PGRES_POLLING_OK
assert pgconn.status == ConnStatus.CONNECTION_OK
def test_ping(pq, dsn):
rv = pq.PGconn.ping(dsn)
- assert rv == PGPing.PQPING_OK
+ assert rv == Ping.PQPING_OK
rv = pq.PGconn.ping("port=99999")
- assert rv == PGPing.PQPING_NO_RESPONSE
+ assert rv == Ping.PQPING_NO_RESPONSE
def test_db(pgconn):