Aug 8, 2019 · Mar 29, 2018 · Mar 29, 2018 · May 31, 2018 · May 31, 2018 · May 31, 2018
diff --git a/docs/source/testgres.rst b/docs/source/testgres.rst
   .. automethod:: __init__

 .. autoclass:: testgres.node.ProcessProxy
   :members:
   :members:

 testgres.standby
 ----------------

 .. automodule:: testgres.standby
    :members:
    :undoc-members:
    :show-inheritance:
diff --git a/testgres/node.py b/testgres/node.py
 import subprocess
 import time

 from collections import Iterable
 from shutil import rmtree
 from six import raise_from, iteritems
 from tempfile import mkstemp, mkdtemp

 from .logger import TestgresLogger

 from .standby import First

 from .utils import \
    eprint, \
    get_bin_path, \

    def reload(self, params=[]):
        """
 Reload config files using pg_ctl.
 Asynchronously reload config files using pg_ctl.

        Args:
            params: additional arguments for pg_ctl.
        with clean_on_error(self.backup(**kwargs)) as backup:
            return backup.spawn_replica(name=name, destroy=True, slot=slot)

    def set_synchronous_standbys(self, standbys):
        """
        Set standby synchronization options. This corresponds to
        `synchronous_standby_names <https://www.postgresql.org/docs/current/static/runtime-config-replication.html#GUC-SYNCHRONOUS-STANDBY-NAMES>`_
        option. Note that :meth:`~.PostgresNode.reload` or
        :meth:`~.PostgresNode.restart` is needed for changes to take place.

        Args:
            standbys: either :class:`.First` or :class:`.Any` object specifying
                sychronization parameters. It is also possible to pass simply
                a list of replicas which would be equivalent to passing
                ``First(1, <list>)``

        Example::

            master = get_new_node().init().start()
            with master.replicate.start() as standby:
                master.append_conf("synchronous_commit = remote_apply")
                master.set_synchronous_standbys(First(1, [standby]))
                master.restart()

        """
        if isinstance(standbys, Iterable):
            standbys = First(1, standbys)

        self.append_conf("synchronous_standby_names = '{}'".format(standbys))

    def catchup(self, dbname=None, username=None):
        """
        Wait until async replica catches up with its master.
diff --git a/testgres/standby.py b/testgres/standby.py
 # coding: utf-8

 import six


 @six.python_2_unicode_compatible
 class First:
    """
 Specifies a priority-based synchronous replication and makes transaction
 commits wait until their WAL records are replicated to ``num_sync``
 synchronous standbys chosen based on their priorities.

 Args:
 sync_num (int): the number of standbys that transaction need to wait
 for replies from
 standbys (:obj:`list` of :class:`.PostgresNode`): the list of standby
 nodes
 """

    def __init__(self, sync_num, standbys):
        self.sync_num = sync_num
        self.standbys = standbys

    def __str__(self):
        return u"{} ({})".format(self.sync_num, u", ".join(
            u"\"{}\"".format(r.name) for r in self.standbys))


 @six.python_2_unicode_compatible
 class Any:
    """
 Specifies a quorum-based synchronous replication and makes transaction
 commits wait until their WAL records are replicated to at least ``num_sync``
 listed standbys. Only available for Postgres 10 and newer.

 Args:
 sync_num (int): the number of standbys that transaction need to wait
 for replies from
 standbys (:obj:`list` of :class:`.PostgresNode`): the list of standby
 nodes
 """

    def __init__(self, sync_num, standbys):
        self.sync_num = sync_num
        self.standbys = standbys

    def __str__(self):
        return u"ANY {} ({})".format(self.sync_num, u", ".join(
            u"\"{}\"".format(r.name) for r in self.standbys))
diff --git a/tests/test_simple.py b/tests/test_simple.py
 from testgres import bound_ports
 from testgres.utils import pg_version_ge
 from testgres.enums import ProcessType
 from testgres.standby import First, Any


 def util_exists(util):
                res = node.execute('select * from test')
                self.assertListEqual(res, [])

    def test_synchronous_replication(self):
        with get_new_node() as master:
            master.init(allow_streaming=True).start()
            master.append_conf('synchronous_commit = remote_apply')

            # create standby
            with master.replicate() as standby1, master.replicate() as standby2:
                standby1.start()
                standby2.start()

                # check formatting
                self.assertEqual('1 ("{}", "{}")'.format(
                    standby1.name, standby2.name),
                                 str(First(1, (standby1, standby2))))
                self.assertEqual('ANY 1 ("{}", "{}")'.format(
                    standby1.name, standby2.name),
                                 str(Any(1, (standby1, standby2))))

                # set synchronous_standby_names
                master.set_synchronous_standbys([standby1, standby2])
                master.restart()
                master.safe_psql('create table abc(a int)')

                # Create a large transaction that will take some time to apply
                # on standby to check that it applies synchronously
                # (If set synchronous_commit to 'on' or other lower level then
                # standby most likely won't catchup so fast and test will fail)
                master.safe_psql(
                    'insert into abc select generate_series(1, 1000000)')
                res = standby1.safe_psql('select count(*) from abc')
                self.assertEqual(res, b'1000000\n')

    def test_replication_slots(self):
        with get_new_node() as node:
            node.init(allow_streaming=True).start()
Original file line number	Diff line number	Diff line change
Expand Up		@@ -59,4 +59,12 @@ testgres.node
		.. automethod:: __init__

		.. autoclass:: testgres.node.ProcessProxy
		:members:
		:members:

		testgres.standby
		----------------

		.. automodule:: testgres.standby
		:members:
		:undoc-members:
		:show-inheritance:
Original file line number	Diff line number	Diff line change
Expand Up		@@ -6,6 +6,7 @@
		import subprocess
		import time

		from collections import Iterable
		from shutil import rmtree
		from six import raise_from, iteritems
		from tempfile import mkstemp, mkdtemp
Expand DownExpand Up		@@ -58,6 +59,8 @@

		from .logger import TestgresLogger

		from .standby import First

		from .utils import \
		eprint, \
		get_bin_path, \
Expand DownExpand Up		@@ -650,7 +653,7 @@ def restart(self, params=[]):

		def reload(self, params=[]):
		"""
		Reload config files using pg_ctl.
		Asynchronously reload config files using pg_ctl.

		Args:
		params: additional arguments for pg_ctl.
Expand DownExpand Up		@@ -979,6 +982,33 @@ def replicate(self, name=None, slot=None, **kwargs):
		with clean_on_error(self.backup(**kwargs)) as backup:
		return backup.spawn_replica(name=name, destroy=True, slot=slot)

		def set_synchronous_standbys(self, standbys):
		"""
		Set standby synchronization options. This corresponds to
		`synchronous_standby_names <https://www.postgresql.org/docs/current/static/runtime-config-replication.html#GUC-SYNCHRONOUS-STANDBY-NAMES>`_
		option. Note that :meth:`~.PostgresNode.reload` or
		:meth:`~.PostgresNode.restart` is needed for changes to take place.

		Args:
		standbys: either :class:`.First` or :class:`.Any` object specifying
		sychronization parameters. It is also possible to pass simply
		a list of replicas which would be equivalent to passing
		``First(1, <list>)``

		Example::

		master = get_new_node().init().start()
		with master.replicate.start() as standby:
		master.append_conf("synchronous_commit = remote_apply")
		master.set_synchronous_standbys(First(1, [standby]))
		master.restart()

		"""
		if isinstance(standbys, Iterable):
		standbys = First(1, standbys)

		self.append_conf("synchronous_standby_names = '{}'".format(standbys))

		def catchup(self, dbname=None, username=None):
		"""
		Wait until async replica catches up with its master.
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,49 @@
		# coding: utf-8

		import six


		@six.python_2_unicode_compatible
		class First:
		"""
		Specifies a priority-based synchronous replication and makes transaction
		commits wait until their WAL records are replicated to ``num_sync``
		synchronous standbys chosen based on their priorities.

		Args:
		sync_num (int): the number of standbys that transaction need to wait
		for replies from
		standbys (:obj:`list` of :class:`.PostgresNode`): the list of standby
		nodes
		"""

		def __init__(self, sync_num, standbys):
		self.sync_num = sync_num
		self.standbys = standbys

		def __str__(self):
		return u"{} ({})".format(self.sync_num, u", ".join(
Copy link Collaborator ildusJun 5, 2018 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. тут мне кажется лучше добавить ключевое слово first, чтобы было явно
		u"\"{}\"".format(r.name) for r in self.standbys))


		@six.python_2_unicode_compatible
		class Any:
		"""
		Specifies a quorum-based synchronous replication and makes transaction
		commits wait until their WAL records are replicated to at least ``num_sync``
		listed standbys. Only available for Postgres 10 and newer.

		Args:
		sync_num (int): the number of standbys that transaction need to wait
		for replies from
		standbys (:obj:`list` of :class:`.PostgresNode`): the list of standby
		nodes
		"""

		def __init__(self, sync_num, standbys):
		self.sync_num = sync_num
		self.standbys = standbys

		def __str__(self):
		return u"ANY {} ({})".format(self.sync_num, u", ".join(
		u"\"{}\"".format(r.name) for r in self.standbys))
Original file line number	Diff line number	Diff line change
Expand Up		@@ -42,6 +42,7 @@
		from testgres import bound_ports
		from testgres.utils import pg_version_ge
		from testgres.enums import ProcessType
		from testgres.standby import First, Any


		def util_exists(util):
Expand DownExpand Up		@@ -388,6 +389,38 @@ def test_replicate(self):
		res = node.execute('select * from test')
		self.assertListEqual(res, [])

		def test_synchronous_replication(self):
		with get_new_node() as master:
		master.init(allow_streaming=True).start()
		master.append_conf('synchronous_commit = remote_apply')

		# create standby
		with master.replicate() as standby1, master.replicate() as standby2:
		standby1.start()
		standby2.start()

		# check formatting
		self.assertEqual('1 ("{}", "{}")'.format(
		standby1.name, standby2.name),
		str(First(1, (standby1, standby2))))
		self.assertEqual('ANY 1 ("{}", "{}")'.format(
		standby1.name, standby2.name),
		str(Any(1, (standby1, standby2))))

		# set synchronous_standby_names
		master.set_synchronous_standbys([standby1, standby2])
		master.restart()
		master.safe_psql('create table abc(a int)')

		# Create a large transaction that will take some time to apply
		# on standby to check that it applies synchronously
		# (If set synchronous_commit to 'on' or other lower level then
		# standby most likely won't catchup so fast and test will fail)
		master.safe_psql(
		'insert into abc select generate_series(1, 1000000)')
		res = standby1.safe_psql('select count(*) from abc')
		self.assertEqual(res, b'1000000\n')

		def test_replication_slots(self):
		with get_new_node() as node:
		node.init(allow_streaming=True).start()
Expand Down