class_docstr: A test case providing connections for tests.
  
  A connection for the test is always available as `self.conn`. Others can be
  created with `self.connect()`. All are closed on tearDown.
  
  Subclasses needing to customize setUp and tearDown should remember to call
  the base class implementations.
class_docstr_tok: ["A","test","case","providing","connections","for","tests",".","A","connection","for","the","test","is","always","available","as","`","self.conn","`",".","Others","can","be","created","with","`","self.connect","(",")","`",".","All","are","closed","on","tearDown",".","Subclasses","needing","to","customize","setUp","and","tearDown","should","remember","to","call","the","base","class","implementations","."]
docstr: Return a connection set up for replication
  
  The connection is on "PSYCOPG2_TEST_REPL_DSN" unless overridden by
  a *dsn* kwarg.
  
  Should raise a skip test if not available, but guard for None on
  old Python versions.
docstr_tok: ["Return","a","connection","set","up","for","replication","The","connection","is","on","``","PSYCOPG2_TEST_REPL_DSN","''","unless","overridden","by","a","*","dsn","*","kwarg",".","Should","raise","a","skip","test","if","not","available",",","but","guard","for","None","on","old","Python","versions","."]
code: def repl_connect(self, **kwargs):
      """Return a connection set up for replication
  
      The connection is on "PSYCOPG2_TEST_REPL_DSN" unless overridden by
      a *dsn* kwarg.
  
      Should raise a skip test if not available, but guard for None on
      old Python versions.
      """
      if repl_dsn is None:
          return self.skipTest("replication tests disabled by default")
  
      if "dsn" not in kwargs:
          kwargs["dsn"] = repl_dsn
      try:
          conn = self.connect(**kwargs)
          if conn.async_ == 1:
              self.wait(conn)
      except psycopg2.OperationalError as 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(
                  f"replication db not configured: {e}"
              )
          else:
              raise
  
      return conn
  
imports: import re
  import os
  import sys
  import types
  import ctypes
  import select
  import operator
  import platform
  import unittest
  from functools import wraps
  from ctypes.util import find_library
  from io import StringIO
  from io import TextIOBase
  from importlib import reload
  import psycopg2
  import psycopg2.errors
  import psycopg2.extensions
  from .testconfig import green, dsn, repl_dsn
label_desc: Use image node_id 5 for calling the ConnectingTestCase obj's underlying member method code with example usage: obj.repl_connect() and returns: conn, self, self