psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-23 01:16:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Added test to verify bug #482

Browse Source
This commit is contained in:
Daniele Varrazzo 2016-12-25 17:46:11 +01:00
parent e27579292a
commit b73115ac41
2 changed files with 25 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tests/test_replication.py Normal file → Executable file
View File

@ -23,7 +23,6 @@
# License for more details. # License for more details.
import psycopg2 import psycopg2
import psycopg2.extensions
from psycopg2.extras import ( from psycopg2.extras import (
PhysicalReplicationConnection, LogicalReplicationConnection, StopReplication) PhysicalReplicationConnection, LogicalReplicationConnection, StopReplication)
@ -89,6 +88,20 @@ class ReplicationTest(ReplicationTestCase):
cur.execute("IDENTIFY_SYSTEM") cur.execute("IDENTIFY_SYSTEM")
cur.fetchall() cur.fetchall()
@skip_before_postgres(9, 0)
def test_datestyle(self):
if testconfig.repl_dsn is None:
return self.skipTest("replication tests disabled by default")
conn = self.repl_connect(
dsn=testconfig.repl_dsn, options='-cdatestyle=german',
connection_factory=PhysicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
cur.execute("IDENTIFY_SYSTEM")
cur.fetchall()
@skip_before_postgres(9, 4) @skip_before_postgres(9, 4)
def test_logical_replication_connection(self): def test_logical_replication_connection(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection) conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
@ -168,7 +181,7 @@ class AsyncReplicationTest(ReplicationTestCase):
connection_factory=LogicalReplicationConnection, async=1) connection_factory=LogicalReplicationConnection, async=1)
if conn is None: if conn is None:
return return
self.wait(conn)
cur = conn.cursor() cur = conn.cursor()
self.create_replication_slot(cur, output_plugin='test_decoding') self.create_replication_slot(cur, output_plugin='test_decoding')

11
tests/testutils.py
View File

@ -130,8 +130,17 @@ class ConnectingTestCase(unittest.TestCase):
import psycopg2 import psycopg2
try: try:
conn = self.connect(**kwargs) conn = self.connect(**kwargs)
if conn.async == 1:
self.wait(conn)
except psycopg2.OperationalError, e: except psycopg2.OperationalError, e:
return self.skipTest("replication db not configured: %s" % e) # If pgcode is not set it is a genuine connection error
# Otherwise we tried to run some bad operation in the connection
# (e.g. bug #482) and we'd rather know that.
if e.pgcode is None:
return self.skipTest("replication db not configured: %s" % e)
else:
raise
return conn return conn
def _get_conn(self): def _get_conn(self):