postgresql_subscription: new module (#63661)

* postgresql_subscription: setup master-standby cluster into one container * postgresql_subscription: setup master-standby cluster into one container, fix * postgresql_subscription: setup master-standby cluster into one container, set up publication * postgresql_subscription: setup master-standby cluster into one container, add module template * postgresql_subscription: setup master-standby cluster into one container, fix tests * postgresql_subscription: setup master-standby cluster into one container, create subscr via shell * postgresql_subscription: setup replication, state stat * postgresql_subscription: add basic present mode * postgresql_subscription: add assertions * postgresql_subscription: add samples * postgresql_subscription: state absent, cascade * postgresql_subscription: add owner param * postgresql_subscription: update * postgresql_subscription: refresh * postgresql_subscription: doc, warns * postgresql_subscription: relinfo * postgresql_subscription: fixes * postgresql_subscription: fix CI tests
5 years ago · 22fe622589
parent da25c2b07b
commit 22fe622589
14 changed files with 1761 additions and 0 deletions
--- a/lib/ansible/modules/database/postgresql/postgresql_subscription.py
+++ b/lib/ansible/modules/database/postgresql/postgresql_subscription.py
@ -0,0 +1,731 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+ANSIBLE_METADATA = {'metadata_version': '1.1',
+                    'status': ['preview'],
+                    'supported_by': 'community'}
+
+
+DOCUMENTATION = r'''
+---
+module: postgresql_subscription
+short_description: Add, update, or remove PostgreSQL subscription
+description:
+- Add, update, or remove PostgreSQL subscription.
+version_added: '2.10'
+
+options:
+  name:
+    description:
+    - Name of the subscription to add, update, or remove.
+    type: str
+    required: yes
+  db:
+    description:
+    - Name of the database to connect to and where
+      the subscription state will be changed.
+    aliases: [ login_db ]
+    type: str
+    required: yes
+  state:
+    description:
+    - The subscription state.
+    - C(present) implies that if I(name) subscription doesn't exist, it will be created.
+    - C(absent) implies that if I(name) subscription exists, it will be removed.
+    - C(stat) implies that if I(name) subscription exists, returns current configuration.
+    - C(refresh) implies that if I(name) subscription exists, it will be refreshed.
+      Fetch missing table information from publisher. Always returns ``changed`` is ``True``.
+      This will start replication of tables that were added to the subscribed-to publications
+      since the last invocation of REFRESH PUBLICATION or since CREATE SUBSCRIPTION.
+      The existing data in the publications that are being subscribed to
+      should be copied once the replication starts.
+    - For more information about C(refresh) see U(https://www.postgresql.org/docs/current/sql-altersubscription.html).
+    type: str
+    choices: [ absent, present, refresh, stat ]
+    default: present
+  relinfo:
+    description:
+    - Get information of the state for each replicated relation in the subscription.
+    type: bool
+    default: false
+  owner:
+    description:
+    - Subscription owner.
+    - If I(owner) is not defined, the owner will be set as I(login_user) or I(session_role).
+    - Ignored when I(state) is not C(present).
+    type: str
+  publications:
+    description:
+    - The publication names on the publisher to use for the subscription.
+    - Ignored when I(state) is not C(present).
+    type: list
+  connparams:
+    description:
+    - The connection dict param-value to connect to the publisher.
+    - For more information see U(https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING).
+    - Ignored when I(state) is not C(present).
+    type: dict
+  cascade:
+    description:
+    - Drop subscription dependencies. Has effect with I(state=absent) only.
+    - Ignored when I(state) is not C(absent).
+    type: bool
+    default: false
+  subsparams:
+    description:
+    - Dictionary of optional parameters for a subscription, e.g. copy_data, enabled, create_slot, etc.
+    - For update the subscription allowed keys are C(enabled), C(slot_name), C(synchronous_commit), C(publication_name).
+    - See available parameters to create a new subscription
+      on U(https://www.postgresql.org/docs/current/sql-createsubscription.html).
+    - Ignored when I(state) is not C(present).
+    type: dict
+
+notes:
+- PostgreSQL version must be 10 or greater.
+
+seealso:
+- module: postgresql_publication
+- name: CREATE SUBSCRIPTION reference
+  description: Complete reference of the CREATE SUBSCRIPTION command documentation.
+  link: https://www.postgresql.org/docs/current/sql-createsubscription.html
+- name: ALTER SUBSCRIPTION reference
+  description: Complete reference of the ALTER SUBSCRIPTION command documentation.
+  link: https://www.postgresql.org/docs/current/sql-altersubscription.html
+- name: DROP SUBSCRIPTION reference
+  description: Complete reference of the DROP SUBSCRIPTION command documentation.
+  link: https://www.postgresql.org/docs/current/sql-dropsubscription.html
+
+author:
+- Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+
+extends_documentation_fragment:
+- postgres
+'''
+
+EXAMPLES = r'''
+- name: >
+    Create acme subscription in mydb database using acme_publication and
+    the following connection parameters to connect to the publisher.
+    Set the subscription owner as alice.
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    state: present
+    publications: acme_publication
+    owner: alice
+    connparams:
+      host: 127.0.0.1
+      port: 5432
+      user: repl
+      password: replpass
+      dbname: mydb
+
+- name: Assuming that acme subscription exists, try to change conn parameters
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    connparams:
+      host: 127.0.0.1
+      port: 5432
+      user: repl
+      password: replpass
+      connect_timeout: 100
+
+- name: Refresh acme publication
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    state: refresh
+
+- name: >
+    Return the configuration of subscription acme if exists in mydb database.
+    Also return state of replicated relations.
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    state: stat
+    relinfo: yes
+
+- name: Drop acme subscription from mydb with dependencies (cascade=yes)
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    state: absent
+    cascade: yes
+
+- name: Assuming that acme subscription exists and enabled, disable the subscription
+  postgresql_subscription:
+    db: mydb
+    name: acme
+    state: present
+    subsparams:
+      enabled: no
+'''
+
+RETURN = r'''
+name:
+  description:
+  - Name of the subscription.
+  returned: always
+  type: str
+  sample: acme
+exists:
+  description:
+  - Flag indicates the subscription exists or not at the end of runtime.
+  returned: always
+  type: bool
+  sample: true
+queries:
+  description: List of executed queries.
+  returned: always
+  type: str
+  sample: [ 'DROP SUBSCRIPTION "mysubscription"' ]
+initial_state:
+  description: Subscription configuration at the beginning of runtime.
+  returned: always
+  type: dict
+  sample: {"conninfo": {}, "enabled": true, "owner": "postgres", "slotname": "test", "synccommit": true}
+final_state:
+  description: Subscription configuration at the end of runtime.
+  returned: always
+  type: dict
+  sample: {"conninfo": {}, "enabled": true, "owner": "postgres", "slotname": "test", "synccommit": true}
+'''
+
+from copy import deepcopy
+
+try:
+    from psycopg2.extras import DictCursor
+except ImportError:
+    # psycopg2 is checked by connect_to_db()
+    # from ansible.module_utils.postgres
+    pass
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.postgres import (
+    connect_to_db,
+    exec_sql,
+    get_conn_params,
+    postgres_common_argument_spec,
+)
+from ansible.module_utils.six import iteritems
+
+SUPPORTED_PG_VERSION = 10000
+
+SUBSPARAMS_KEYS_FOR_UPDATE = ('enabled', 'synchronous_commit', 'slot_name')
+
+
+################################
+# Module functions and classes #
+################################
+
+def convert_conn_params(conn_dict):
+    """Converts the passed connection dictionary to string.
+
+    Args:
+        conn_dict (list): Dictionary which needs to be converted.
+
+    Returns:
+        Connection string.
+    """
+    conn_list = []
+    for (param, val) in iteritems(conn_dict):
+        conn_list.append('%s=%s' % (param, val))
+
+    return ' '.join(conn_list)
+
+
+def convert_subscr_params(params_dict):
+    """Converts the passed params dictionary to string.
+
+    Args:
+        params_dict (list): Dictionary which needs to be converted.
+
+    Returns:
+        Parameters string.
+    """
+    params_list = []
+    for (param, val) in iteritems(params_dict):
+        if val is False:
+            val = 'false'
+        elif val is True:
+            val = 'true'
+
+        params_list.append('%s = %s' % (param, val))
+
+    return ', '.join(params_list)
+
+
+class PgSubscription():
+    """Class to work with PostgreSQL subscription.
+
+    Args:
+        module (AnsibleModule): Object of AnsibleModule class.
+        cursor (cursor): Cursor object of psycopg2 library to work with PostgreSQL.
+        name (str): The name of the subscription.
+        db (str): The database name the subscription will be associated with.
+
+    Kwargs:
+        relinfo (bool): Flag indicates the relation information is needed.
+
+    Attributes:
+        module (AnsibleModule): Object of AnsibleModule class.
+        cursor (cursor): Cursor object of psycopg2 library to work with PostgreSQL.
+        name (str): Name of subscription.
+        executed_queries (list): List of executed queries.
+        attrs (dict): Dict with subscription attributes.
+        exists (bool): Flag indicates the subscription exists or not.
+    """
+
+    def __init__(self, module, cursor, name, db, relinfo):
+        self.module = module
+        self.cursor = cursor
+        self.name = name
+        self.db = db
+        self.relinfo = relinfo
+        self.executed_queries = []
+        self.attrs = {
+            'owner': None,
+            'enabled': None,
+            'synccommit': None,
+            'conninfo': {},
+            'slotname': None,
+            'publications': [],
+            'relinfo': None,
+        }
+        self.empty_attrs = deepcopy(self.attrs)
+        self.exists = self.check_subscr()
+
+    def get_info(self):
+        """Refresh the subscription information.
+
+        Returns:
+            ``self.attrs``.
+        """
+        self.exists = self.check_subscr()
+        return self.attrs
+
+    def check_subscr(self):
+        """Check the subscription and refresh ``self.attrs`` subscription attribute.
+
+        Returns:
+            True if the subscription with ``self.name`` exists, False otherwise.
+        """
+
+        subscr_info = self.__get_general_subscr_info()
+
+        if not subscr_info:
+            # The subscription does not exist:
+            self.attrs = deepcopy(self.empty_attrs)
+            return False
+
+        self.attrs['owner'] = subscr_info.get('rolname')
+        self.attrs['enabled'] = subscr_info.get('subenabled')
+        self.attrs['synccommit'] = subscr_info.get('subenabled')
+        self.attrs['slotname'] = subscr_info.get('subslotname')
+        self.attrs['publications'] = subscr_info.get('subpublications')
+        if subscr_info.get('subconninfo'):
+            for param in subscr_info['subconninfo'].split(' '):
+                tmp = param.split('=')
+                try:
+                    self.attrs['conninfo'][tmp[0]] = int(tmp[1])
+                except ValueError:
+                    self.attrs['conninfo'][tmp[0]] = tmp[1]
+
+        if self.relinfo:
+            self.attrs['relinfo'] = self.__get_rel_info()
+
+        return True
+
+    def create(self, connparams, publications, subsparams, check_mode=True):
+        """Create the subscription.
+
+        Args:
+            connparams (str): Connection string in libpq style.
+            publications (list): Publications on the master to use.
+            subsparams (str): Parameters string in WITH () clause style.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            changed (bool): True if the subscription has been created, otherwise False.
+        """
+        query_fragments = []
+        query_fragments.append("CREATE SUBSCRIPTION %s CONNECTION '%s' "
+                               "PUBLICATION %s" % (self.name, connparams, ', '.join(publications)))
+
+        if subsparams:
+            query_fragments.append("WITH (%s)" % subsparams)
+
+        changed = self.__exec_sql(' '.join(query_fragments), check_mode=check_mode)
+
+        return changed
+
+    def update(self, connparams, publications, subsparams, check_mode=True):
+        """Update the subscription.
+
+        Args:
+            connparams (str): Connection string in libpq style.
+            publications (list): Publications on the master to use.
+            subsparams (dict): Dictionary of optional parameters.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            changed (bool): True if subscription has been updated, otherwise False.
+        """
+        changed = False
+
+        if connparams:
+            if connparams != self.attrs['conninfo']:
+                changed = self.__set_conn_params(convert_conn_params(connparams),
+                                                 check_mode=check_mode)
+
+        if publications:
+            if sorted(self.attrs['publications']) != sorted(publications):
+                changed = self.__set_publications(publications, check_mode=check_mode)
+
+        if subsparams:
+            params_to_update = []
+
+            for (param, value) in iteritems(subsparams):
+                if param == 'enabled':
+                    if self.attrs['enabled'] and value is False:
+                        changed = self.enable(enabled=False, check_mode=check_mode)
+                    elif not self.attrs['enabled'] and value is True:
+                        changed = self.enable(enabled=True, check_mode=check_mode)
+
+                elif param == 'synchronous_commit':
+                    if self.attrs['synccommit'] is True and value is False:
+                        params_to_update.append("%s = false" % param)
+                    elif self.attrs['synccommit'] is False and value is True:
+                        params_to_update.append("%s = true" % param)
+
+                elif param == 'slot_name':
+                    if self.attrs['slotname'] and self.attrs['slotname'] != value:
+                        params_to_update.append("%s = %s" % (param, value))
+
+                else:
+                    self.module.warn("Parameter '%s' is not in params supported "
+                                     "for update '%s', ignored..." % (param, SUBSPARAMS_KEYS_FOR_UPDATE))
+
+            if params_to_update:
+                changed = self.__set_params(params_to_update, check_mode=check_mode)
+
+        return changed
+
+    def drop(self, cascade=False, check_mode=True):
+        """Drop the subscription.
+
+        Kwargs:
+            cascade (bool): Flag indicates that the subscription needs to be deleted
+                with its dependencies.
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            changed (bool): True if the subscription has been removed, otherwise False.
+        """
+        if self.exists:
+            query_fragments = ["DROP SUBSCRIPTION %s" % self.name]
+            if cascade:
+                query_fragments.append("CASCADE")
+
+            return self.__exec_sql(' '.join(query_fragments), check_mode=check_mode)
+
+    def set_owner(self, role, check_mode=True):
+        """Set a subscription owner.
+
+        Args:
+            role (str): Role (user) name that needs to be set as a subscription owner.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        query = 'ALTER SUBSCRIPTION %s OWNER TO "%s"' % (self.name, role)
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def refresh(self, check_mode=True):
+        """Refresh publication.
+
+        Fetches missing table info from publisher.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        query = 'ALTER SUBSCRIPTION %s REFRESH PUBLICATION' % self.name
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def __set_params(self, params_to_update, check_mode=True):
+        """Update optional subscription parameters.
+
+        Args:
+            params_to_update (list): Parameters with values to update.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        query = 'ALTER SUBSCRIPTION %s SET (%s)' % (self.name, ', '.join(params_to_update))
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def __set_conn_params(self, connparams, check_mode=True):
+        """Update connection parameters.
+
+        Args:
+            connparams (str): Connection string in libpq style.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        query = "ALTER SUBSCRIPTION %s CONNECTION '%s'" % (self.name, connparams)
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def __set_publications(self, publications, check_mode=True):
+        """Update publications.
+
+        Args:
+            publications (list): Publications on the master to use.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        query = 'ALTER SUBSCRIPTION %s SET PUBLICATION %s' % (self.name, ', '.join(publications))
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def enable(self, enabled=True, check_mode=True):
+        """Enable or disable the subscription.
+
+        Kwargs:
+            enable (bool): Flag indicates that the subscription needs
+                to be enabled or disabled.
+            check_mode (bool): If True, don't actually change anything,
+                just make SQL, add it to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        if enabled:
+            query = 'ALTER SUBSCRIPTION %s ENABLE' % self.name
+        else:
+            query = 'ALTER SUBSCRIPTION %s DISABLE' % self.name
+
+        return self.__exec_sql(query, check_mode=check_mode)
+
+    def __get_general_subscr_info(self):
+        """Get and return general subscription information.
+
+        Returns:
+            Dict with subscription information if successful, False otherwise.
+        """
+        query = ("SELECT d.datname, r.rolname, s.subenabled, "
+                 "s.subconninfo, s.subslotname, s.subsynccommit, "
+                 "s.subpublications FROM pg_catalog.pg_subscription s "
+                 "JOIN pg_catalog.pg_database d "
+                 "ON s.subdbid = d.oid "
+                 "JOIN pg_catalog.pg_roles AS r "
+                 "ON s.subowner = r.oid "
+                 "WHERE s.subname = '%s' AND d.datname = '%s'" % (self.name, self.db))
+
+        result = exec_sql(self, query, add_to_executed=False)
+        if result:
+            return result[0]
+        else:
+            return False
+
+    def __get_rel_info(self):
+        """Get and return state of relations replicated by the subscription.
+
+        Returns:
+            List of dicts containing relations state if successful, False otherwise.
+        """
+        query = ("SELECT c.relname, r.srsubstate, r.srsublsn "
+                 "FROM pg_catalog.pg_subscription_rel r "
+                 "JOIN pg_catalog.pg_subscription s ON s.oid = r.srsubid "
+                 "JOIN pg_catalog.pg_class c ON c.oid = r.srrelid "
+                 "WHERE s.subname = '%s'" % self.name)
+
+        result = exec_sql(self, query, add_to_executed=False)
+        if result:
+            return [dict(row) for row in result]
+        else:
+            return False
+
+    def __exec_sql(self, query, check_mode=False):
+        """Execute SQL query.
+
+        Note: If we need just to get information from the database,
+            we use ``exec_sql`` function directly.
+
+        Args:
+            query (str): Query that needs to be executed.
+
+        Kwargs:
+            check_mode (bool): If True, don't actually change anything,
+                just add ``query`` to ``self.executed_queries`` and return True.
+
+        Returns:
+            True if successful, False otherwise.
+        """
+        if check_mode:
+            self.executed_queries.append(query)
+            return True
+        else:
+            return exec_sql(self, query, ddl=True)
+
+
+# ===========================================
+# Module execution.
+#
+
+
+def main():
+    argument_spec = postgres_common_argument_spec()
+    argument_spec.update(
+        name=dict(required=True),
+        db=dict(type='str', aliases=['login_db']),
+        state=dict(type='str', default='present', choices=['absent', 'present', 'refresh', 'stat']),
+        publications=dict(type='list'),
+        connparams=dict(type='dict'),
+        cascade=dict(type='bool', default=False),
+        owner=dict(type='str'),
+        subsparams=dict(type='dict'),
+        relinfo=dict(type='bool', default=False),
+    )
+    module = AnsibleModule(
+        argument_spec=argument_spec,
+        supports_check_mode=True,
+    )
+
+    # Parameters handling:
+    db = module.params['db']
+    name = module.params['name']
+    state = module.params['state']
+    publications = module.params['publications']
+    cascade = module.params['cascade']
+    owner = module.params['owner']
+    subsparams = module.params['subsparams']
+    connparams = module.params['connparams']
+    relinfo = module.params['relinfo']
+
+    if state == 'present' and cascade:
+        module.warm('parameter "cascade" is ignored when state is not absent')
+
+    if state != 'present':
+        if owner:
+            module.warm("parameter 'owner' is ignored when state is not 'present'")
+        if publications:
+            module.warm("parameter 'publications' is ignored when state is not 'present'")
+        if connparams:
+            module.warm("parameter 'connparams' is ignored when state is not 'present'")
+        if subsparams:
+            module.warm("parameter 'subsparams' is ignored when state is not 'present'")
+
+    # Connect to DB and make cursor object:
+    pg_conn_params = get_conn_params(module, module.params)
+    # We check subscription state without DML queries execution, so set autocommit:
+    db_connection = connect_to_db(module, pg_conn_params, autocommit=True)
+    cursor = db_connection.cursor(cursor_factory=DictCursor)
+
+    # Check version:
+    if cursor.connection.server_version < SUPPORTED_PG_VERSION:
+        module.fail_json(msg="PostgreSQL server version should be 10.0 or greater")
+
+    # Set defaults:
+    changed = False
+    initial_state = {}
+    final_state = {}
+
+    ###################################
+    # Create object and do rock'n'roll:
+    subscription = PgSubscription(module, cursor, name, db, relinfo)
+
+    if subscription.exists:
+        initial_state = deepcopy(subscription.attrs)
+        final_state = deepcopy(initial_state)
+
+    # If module.check_mode=True, nothing will be changed:
+    if state == 'stat':
+        # Information has been collected already, so nothing is needed:
+        pass
+
+    if state == 'present':
+        if not subscription.exists:
+            if subsparams:
+                subsparams = convert_subscr_params(subsparams)
+
+            if connparams:
+                connparams = convert_conn_params(connparams)
+
+            changed = subscription.create(connparams,
+                                          publications,
+                                          subsparams,
+                                          check_mode=module.check_mode)
+
+        else:
+            changed = subscription.update(connparams,
+                                          publications,
+                                          subsparams,
+                                          check_mode=module.check_mode)
+
+        if owner and subscription.attrs['owner'] != owner:
+            changed = subscription.set_owner(owner, check_mode=module.check_mode)
+
+    elif state == 'absent':
+        changed = subscription.drop(cascade, check_mode=module.check_mode)
+
+    elif state == 'refresh':
+        if not subscription.exists:
+            module.fail_json(msg="Refresh failed: subscription '%s' does not exist" % name)
+
+        # Always returns True:
+        changed = subscription.refresh(check_mode=module.check_mode)
+
+    # Get final subscription info if needed:
+    final_state = subscription.get_info()
+
+    # Connection is not needed any more:
+    cursor.close()
+    db_connection.close()
+
+    # Return ret values and exit:
+    module.exit_json(changed=changed,
+                     name=name,
+                     exists=subscription.exists,
+                     queries=subscription.executed_queries,
+                     initial_state=initial_state,
+                     final_state=final_state)
+
+
+if __name__ == '__main__':
+    main()
--- a/test/integration/targets/postgresql_subscription/aliases
+++ b/test/integration/targets/postgresql_subscription/aliases
@ -0,0 +1,8 @@
+destructive
+shippable/posix/group1
+skip/osx
+skip/centos
+skip/freebsd
+skip/rhel
+skip/opensuse
+skip/fedora
--- a/test/integration/targets/postgresql_subscription/defaults/main.yml
+++ b/test/integration/targets/postgresql_subscription/defaults/main.yml
@ -0,0 +1,15 @@
+pg_user: postgres
+db_default: postgres
+master_port: 5433
+replica_port: 5434
+
+test_table1: acme1
+test_pub: first_publication
+test_pub2: second_publication
+replication_role: logical_replication
+replication_pass: alsdjfKJKDf1#
+test_db: acme_db
+test_subscription: test
+test_role1: alice
+test_role2: bob
+conn_timeout: 100
--- a/test/integration/targets/postgresql_subscription/meta/main.yml
+++ b/test/integration/targets/postgresql_subscription/meta/main.yml
@ -0,0 +1,2 @@
+dependencies:
+- setup_postgresql_replication
--- a/test/integration/targets/postgresql_subscription/tasks/main.yml
+++ b/test/integration/targets/postgresql_subscription/tasks/main.yml
@ -0,0 +1,7 @@
+# Initial tests of postgresql_subscription module:
+
+- import_tasks: setup_publication.yml
+  when: ansible_distribution == 'Ubuntu' and ansible_distribution_major_version >= '18'
+
+- import_tasks: postgresql_subscription_initial.yml
+  when: ansible_distribution == 'Ubuntu' and ansible_distribution_major_version >= '18'
--- a/test/integration/targets/postgresql_subscription/tasks/postgresql_subscription_initial.yml
+++ b/test/integration/targets/postgresql_subscription/tasks/postgresql_subscription_initial.yml
@ -0,0 +1,697 @@
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+- vars:
+    task_parameters: &task_parameters
+      become_user: '{{ pg_user }}'
+      become: yes
+      register: result
+    pg_parameters: &pg_parameters
+      login_user: '{{ pg_user }}'
+      login_db: '{{ test_db }}'
+
+  block:
+
+  - name: Create roles to test owner parameter
+    <<: *task_parameters
+    postgresql_user:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ item }}'
+      role_attr_flags: SUPERUSER,LOGIN
+    loop:
+    - '{{ test_role1 }}'
+    - '{{ test_role2 }}'
+
+  ####################
+  # Test mode: present
+  ####################
+  - name: Create subscription
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["CREATE SUBSCRIPTION test CONNECTION 'host=127.0.0.1 port={{ master_port }} user={{ replication_role }} password={{ replication_pass }} dbname={{ test_db }}' PUBLICATION {{ test_pub }}"]
+      - result.exists == true
+      - result.initial_state == {}
+      - result.final_state.owner == '{{ pg_user }}'
+      - result.final_state.enabled == true
+      - result.final_state.publications == ["{{ test_pub }}"]
+      - result.final_state.synccommit == true
+      - result.final_state.slotname == '{{ test_subscription }}'
+      - result.final_state.conninfo.dbname == '{{ test_db }}'
+      - result.final_state.conninfo.host == '127.0.0.1'
+      - result.final_state.conninfo.port == {{ master_port }}
+      - result.final_state.conninfo.user == '{{ replication_role }}'
+      - result.final_state.conninfo.password == '{{ replication_pass }}'
+
+  #################
+  # Test mode: stat
+  #################
+
+  - name: Stat mode in check mode
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+      - result.final_state == result.initial_state
+      - result.final_state.owner == '{{ pg_user }}'
+      - result.final_state.enabled == true
+      - result.final_state.publications == ["{{ test_pub }}"]
+      - result.final_state.synccommit == true
+      - result.final_state.slotname == '{{ test_subscription }}'
+      - result.final_state.conninfo.dbname == '{{ test_db }}'
+      - result.final_state.conninfo.host == '127.0.0.1'
+      - result.final_state.conninfo.port == {{ master_port }}
+      - result.final_state.conninfo.user == '{{ replication_role }}'
+      - result.final_state.conninfo.password == '{{ replication_pass }}'
+
+  - name: Stat mode
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+      - result.final_state == result.initial_state
+      - result.final_state.owner == '{{ pg_user }}'
+      - result.final_state.enabled == true
+      - result.final_state.publications == ["{{ test_pub }}"]
+      - result.final_state.synccommit == true
+      - result.final_state.slotname == '{{ test_subscription }}'
+      - result.final_state.conninfo.dbname == '{{ test_db }}'
+      - result.final_state.conninfo.host == '127.0.0.1'
+      - result.final_state.conninfo.port == {{ master_port }}
+      - result.final_state.conninfo.user == '{{ replication_role }}'
+      - result.final_state.conninfo.password == '{{ replication_pass }}'
+
+  ###################
+  # Test mode: absent
+  ###################
+
+  - name: Drop subscription in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: absent
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.queries == ["DROP SUBSCRIPTION {{ test_subscription }}"]
+      - result.final_state == result.initial_state
+
+  - name: Check the subscription exists
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+
+  - name: Drop subscription
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: absent
+
+  - assert:
+      that:
+      - result is changed
+      - result.queries == ["DROP SUBSCRIPTION {{ test_subscription }}"]
+      - result.final_state != result.initial_state
+
+  - name: Check the subscription doesn't exist
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == false
+
+  ##################
+  # Test owner param
+  ##################
+
+  - name: Create with owner
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      owner: '{{ test_role1 }}'
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+
+  - assert:
+      that:
+      - result.final_state.owner == '{{ test_role1 }}'
+      - result.queries[1] == 'ALTER SUBSCRIPTION {{ test_subscription }} OWNER TO "{{ test_role1 }}"'
+
+  - name: Try to set this owner again
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      owner: '{{ test_role1 }}'
+
+  - assert:
+      that:
+      - result is not changed
+      - result.initial_state == result.final_state
+      - result.final_state.owner == '{{ test_role1 }}'
+
+  - name: Check owner
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+      - result.initial_state.owner == '{{ test_role1 }}'
+
+  - name: Set another owner in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      owner: '{{ test_role2 }}'
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.initial_state == result.final_state
+      - result.final_state.owner == '{{ test_role1 }}'
+      - result.queries == ['ALTER SUBSCRIPTION {{ test_subscription }} OWNER TO "{{ test_role2 }}"']
+
+  - name: Check owner
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+      - result.initial_state.owner == '{{ test_role1 }}'
+
+  - name: Set another owner
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      owner: '{{ test_role2 }}'
+
+  - assert:
+      that:
+      - result is changed
+      - result.initial_state != result.final_state
+      - result.final_state.owner == '{{ test_role2 }}'
+      - result.queries == ['ALTER SUBSCRIPTION {{ test_subscription }} OWNER TO "{{ test_role2 }}"']
+
+  - name: Check owner
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+      - result.initial_state.owner == '{{ test_role2 }}'
+
+  ##############################
+  # Test cascade and owner param
+  ##############################
+
+  - name: Drop subscription cascade in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: absent
+      cascade: yes
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.queries == ["DROP SUBSCRIPTION {{ test_subscription }} CASCADE"]
+      - result.final_state == result.initial_state
+
+  - name: Check the subscription exists
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == true
+
+  - name: Drop subscription cascade
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: absent
+      cascade: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.queries == ["DROP SUBSCRIPTION {{ test_subscription }} CASCADE"]
+      - result.final_state != result.initial_state
+
+  - name: Check the subscription doesn't exist
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.exists == false
+
+  ###########################
+  # Test subsparams parameter
+  ###########################
+
+  - name: Create subscription with subsparams
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications: '{{ test_pub }}'
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+      subsparams:
+        enabled: no
+        synchronous_commit: no
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["CREATE SUBSCRIPTION test CONNECTION 'host=127.0.0.1 port={{ master_port }} user={{ replication_role }} password={{ replication_pass }} dbname={{ test_db }}' PUBLICATION {{ test_pub }} WITH (enabled = false, synchronous_commit = false)"]
+      - result.exists == true
+      - result.final_state.enabled == false
+      - result.final_state.synccommit == false
+
+  - name: Stat mode
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.enabled == false
+      - result.final_state.synccommit == false
+
+  - name: Enable changed params
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      publications: '{{ test_pub }}'
+      subsparams:
+        enabled: yes
+        synchronous_commit: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["ALTER SUBSCRIPTION {{ test_subscription }} ENABLE", "ALTER SUBSCRIPTION {{ test_subscription }} SET (synchronous_commit = true)"]
+      - result.exists == true
+      - result.final_state.enabled == true
+      - result.final_state.synccommit == true
+
+  - name: Stat mode
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.enabled == true
+      - result.final_state.synccommit == true
+
+  - name: Enable the same params again
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      publications: '{{ test_pub }}'
+      subsparams:
+        enabled: yes
+        synchronous_commit: yes
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == []
+      - result.exists == true
+      - result.final_state == result.initial_state
+      - result.final_state.enabled == true
+      - result.final_state.synccommit == true
+
+  ##########################
+  # Test change publications
+  ##########################
+
+  - name: Change publications in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications:
+      - '{{ test_pub }}'
+      - '{{ test_pub2 }}'
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.publications == result.initial_state.publications
+      - result.final_state.publications == ['{{ test_pub }}']
+      - result.queries == ['ALTER SUBSCRIPTION {{ test_subscription }} SET PUBLICATION {{ test_pub }}, {{ test_pub2 }}']
+
+  - name: Check publications
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.publications == ['{{ test_pub }}']
+
+  - name: Change publications
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications:
+      - '{{ test_pub }}'
+      - '{{ test_pub2 }}'
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.publications != result.initial_state.publications
+      - result.final_state.publications == ['{{ test_pub }}', '{{ test_pub2 }}']
+      - result.queries == ['ALTER SUBSCRIPTION {{ test_subscription }} SET PUBLICATION {{ test_pub }}, {{ test_pub2 }}']
+
+  - name: Check publications
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.publications == ['{{ test_pub }}', '{{ test_pub2 }}']
+
+  - name: Change publications with the same values again
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      publications:
+      - '{{ test_pub }}'
+      - '{{ test_pub2 }}'
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.publications == result.initial_state.publications
+      - result.final_state.publications == ['{{ test_pub }}', '{{ test_pub2 }}']
+      - result.queries == []
+
+  ######################
+  # Test update conninfo
+  ######################
+
+  - name: Change conninfo in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+        connect_timeout: '{{ conn_timeout }}'
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["ALTER SUBSCRIPTION {{ test_subscription }} CONNECTION 'host=127.0.0.1 port={{ master_port }} user={{ replication_role }} password={{ replication_pass }} dbname={{ test_db }} connect_timeout={{ conn_timeout }}'"]
+      - result.initial_state.conninfo == result.final_state.conninfo
+
+  - name: Change conninfo
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      relinfo: yes
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+        connect_timeout: '{{ conn_timeout }}'
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["ALTER SUBSCRIPTION {{ test_subscription }} CONNECTION 'host=127.0.0.1 port={{ master_port }} user={{ replication_role }} password={{ replication_pass }} dbname={{ test_db }} connect_timeout={{ conn_timeout }}'"]
+      - result.initial_state.conninfo != result.final_state.conninfo
+
+  - name: Check publications
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.conninfo.connect_timeout == {{ conn_timeout }}
+
+  - name: Try to change conninfo again with the same values
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: present
+      connparams:
+        host: 127.0.0.1
+        port: '{{ master_port }}'
+        user: '{{ replication_role }}'
+        password: '{{ replication_pass }}'
+        dbname: '{{ test_db }}'
+        connect_timeout: '{{ conn_timeout }}'
+
+  - assert:
+      that:
+      - result is not changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == []
+      - result.initial_state.conninfo == result.final_state.conninfo
+      - result.final_state.conninfo.connect_timeout == {{ conn_timeout }}
+
+  ####################
+  # Test state refresh
+  ####################
+
+  - name: Refresh in check mode
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: refresh
+    check_mode: yes
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["ALTER SUBSCRIPTION {{ test_subscription }} REFRESH PUBLICATION"]
+
+  - name: Refresh
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: refresh
+
+  - assert:
+      that:
+      - result is changed
+      - result.name == '{{ test_subscription }}'
+      - result.queries == ["ALTER SUBSCRIPTION {{ test_subscription }} REFRESH PUBLICATION"]
+
+  ####################
+  # Test relinfo param
+  ####################
+
+  - name: Get relinfo
+    <<: *task_parameters
+    postgresql_subscription: 
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: stat
+      relinfo: yes
+
+  - assert:
+      that:
+      - result.name == '{{ test_subscription }}'
+      - result.final_state.relinfo[0].relname == '{{ test_table1 }}'
+      - result.final_state == result.initial_state
+
+  ##########
+  # Clean up
+  ##########
+  - name: Drop subscription
+    <<: *task_parameters
+    postgresql_subscription:
+      <<: *pg_parameters
+      login_port: '{{ replica_port }}'
+      name: '{{ test_subscription }}'
+      state: absent
--- a/test/integration/targets/postgresql_subscription/tasks/setup_publication.yml
+++ b/test/integration/targets/postgresql_subscription/tasks/setup_publication.yml
@ -0,0 +1,84 @@
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+# Preparation for further tests of postgresql_subscription module.
+
+- vars:
+    task_parameters: &task_parameters
+      become_user: '{{ pg_user }}'
+      become: yes
+      register: result
+    pg_parameters: &pg_parameters
+      login_user: '{{ pg_user }}'
+      login_db: '{{ test_db }}'
+
+  block:
+  - name: postgresql_publication - create test db
+    <<: *task_parameters
+    postgresql_db:
+      login_user: '{{ pg_user }}'
+      login_port: '{{ master_port }}'
+      maintenance_db: '{{ db_default }}'
+      name: '{{ test_db }}'
+
+  - name: postgresql_publication - create test role
+    <<: *task_parameters
+    postgresql_user:
+      <<: *pg_parameters
+      login_port: '{{ master_port }}'
+      name: '{{ replication_role }}'
+      password: '{{ replication_pass }}'
+      role_attr_flags: LOGIN,REPLICATION
+
+  - name: postgresql_publication - create test table
+    <<: *task_parameters
+    postgresql_table:
+      <<: *pg_parameters
+      login_port: '{{ master_port }}'
+      name: '{{ test_table1 }}'
+      columns:
+      - id int
+
+  - name: Master - dump schema
+    <<: *task_parameters
+    shell: pg_dumpall -p '{{ master_port }}' -s > /tmp/schema.sql
+
+  - name: Replicat restore schema
+    <<: *task_parameters
+    shell: psql -p '{{ replica_port }}' -f /tmp/schema.sql
+
+  - name: postgresql_publication - create publication
+    <<: *task_parameters
+    postgresql_publication: 
+      <<: *pg_parameters
+      login_port: '{{ master_port }}'
+      name: '{{ test_pub }}'
+
+  - assert:
+      that:
+      - result is changed
+      - result.exists == true
+      - result.queries == ["CREATE PUBLICATION \"{{ test_pub }}\" FOR ALL TABLES"]
+      - result.owner == '{{ pg_user }}'
+      - result.alltables == true
+      - result.tables == []
+      - result.parameters.publish != {}
+
+  - name: postgresql_publication - create one more publication
+    <<: *task_parameters
+    postgresql_publication: 
+      <<: *pg_parameters
+      login_port: '{{ master_port }}'
+      name: '{{ test_pub2 }}'
+
+  - name: postgresql_publication - check the publication was created
+    <<: *task_parameters
+    postgresql_query:
+      <<: *pg_parameters
+      login_port: '{{ master_port }}'
+      query: >
+        SELECT * FROM pg_publication WHERE pubname = '{{ test_pub }}'
+        AND pubowner = '10' AND puballtables = 't'
+
+  - assert:
+      that:
+      - result.rowcount == 1
--- a/test/integration/targets/setup_postgresql_replication/defaults/main.yml
+++ b/test/integration/targets/setup_postgresql_replication/defaults/main.yml
@ -0,0 +1,30 @@
+# General:
+pg_user: postgres
+db_default: postgres
+
+pg_package_list:
+- apt-utils
+- postgresql
+- postgresql-contrib
+- python3-psycopg2
+
+packages_to_remove:
+- postgresql
+- postgresql-contrib
+- postgresql-server
+- postgresql-libs
+- python3-psycopg2
+
+# Master specific defaults:
+master_root_dir: '/var/lib/pgsql/master'
+master_data_dir: '{{ master_root_dir }}/data'
+master_postgresql_conf: '{{ master_data_dir }}/postgresql.conf'
+master_pg_hba_conf: '{{ master_data_dir }}/pg_hba.conf'
+master_port: 5433
+
+# Replica specific defaults:
+replica_root_dir: '/var/lib/pgsql/replica'
+replica_data_dir: '{{ replica_root_dir }}/data'
+replica_postgresql_conf: '{{ replica_data_dir }}/postgresql.conf'
+replica_pg_hba_conf: '{{ replica_data_dir }}/pg_hba.conf'
+replica_port: 5434
--- a/test/integration/targets/setup_postgresql_replication/handlers/main.yml
+++ b/test/integration/targets/setup_postgresql_replication/handlers/main.yml
@ -0,0 +1,23 @@
+- name: Stop services
+  become: yes
+  become_user: '{{ pg_user }}'
+  shell: '{{ pg_ctl }} -D {{ item.datadir }} -o "-p {{ item.port }}" -m immediate stop'
+  loop:
+  - { datadir: '{{ master_data_dir }}', port: '{{ master_port }}' }
+  - { datadir: '{{ replica_data_dir }}', port: '{{ replica_port }}' }
+  listen: stop postgresql
+
+- name: Remove packages
+  apt:
+    name: '{{ packages_to_remove }}'
+    state: absent
+  listen: cleanup postgresql
+
+- name: Remove FS objects
+  file:
+    state: absent
+    path: "{{ item }}"
+  loop:
+  - "{{ master_root_dir }}"
+  - "{{ replica_root_dir }}"
+  listen: cleanup postgresql
--- a/test/integration/targets/setup_postgresql_replication/tasks/main.yml
+++ b/test/integration/targets/setup_postgresql_replication/tasks/main.yml
@ -0,0 +1,8 @@
+# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) <aaklychkov@mail.ru>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# Setup PostgreSQL master-standby replication into one container:
+- import_tasks: setup_postgresql_cluster.yml
+  when:
+  - ansible_distribution == 'Ubuntu'
+  - ansible_distribution_major_version >= '18'
--- a/test/integration/targets/setup_postgresql_replication/tasks/setup_postgresql_cluster.yml
+++ b/test/integration/targets/setup_postgresql_replication/tasks/setup_postgresql_cluster.yml
@ -0,0 +1,93 @@
+# We run two servers listening different ports
+# to be able to check replication (one server for master, another for standby).
+
+- name: Install packages
+  apt:
+    name: '{{ pg_package_list }}'
+  notify: cleanup postgresql
+
+- name: Create root dirs
+  file:
+    state: directory
+    path: "{{ item }}"
+    owner: postgres
+    group: postgres
+    mode: 0700
+  loop:
+  - "{{ master_root_dir }}"
+  - "{{ master_data_dir }}"
+  - "{{ replica_root_dir }}"
+  - "{{ replica_data_dir }}"
+  notify: cleanup postgresql
+
+- name: Find initdb
+  shell: find /usr/lib -type f -name "initdb"
+  register: result
+
+- name: Set path to initdb
+  set_fact:
+    initdb: '{{ result.stdout }}'
+
+- name: Initialize databases
+  become: yes
+  become_user: '{{ pg_user }}'
+  shell: '{{ initdb }} --pgdata {{ item }}'
+  loop:
+  - "{{ master_data_dir }}"
+  - "{{ replica_data_dir }}"
+
+- name: Copy config templates
+  template:
+    src: '{{ item.conf_templ }}'
+    dest: '{{ item.conf_dest }}'
+    owner: postgres
+    group: postgres
+    force: yes
+  loop:
+  - { conf_templ: master_postgresql.conf.j2, conf_dest: '{{ master_postgresql_conf }}' }
+  - { conf_templ: replica_postgresql.conf.j2, conf_dest: '{{ replica_postgresql_conf }}' }
+  - { conf_templ: pg_hba.conf.j2, conf_dest: '{{ master_pg_hba_conf }}' }
+  - { conf_templ: pg_hba.conf.j2, conf_dest: '{{ replica_pg_hba_conf }}' }
+
+- name: Find pg_ctl
+  shell: find /usr/lib -type f -name "pg_ctl"
+  register: result
+
+- name: Set path to initdb
+  set_fact:
+    pg_ctl: '{{ result.stdout }}'
+
+- name: Start servers
+  become: yes
+  become_user: '{{ pg_user }}'
+  shell: '{{ pg_ctl }} -D {{ item.datadir }} -o "-p {{ item.port }}" start'
+  loop:
+  - { datadir: '{{ master_data_dir }}', port: '{{ master_port }}' }
+  - { datadir: '{{ replica_data_dir }}', port: '{{ replica_port }}' }
+  notify: stop postgresql
+
+- name: Check connectivity to the master and get PostgreSQL version
+  become: yes
+  become_user: '{{ pg_user }}'
+  postgresql_ping:
+    db: '{{ db_default }}'
+    login_user: '{{ pg_user }}'
+    login_port: '{{ master_port }}'
+  register: result
+
+- name: Check connectivity to the replica and get PostgreSQL version
+  become: yes
+  become_user: '{{ pg_user }}'
+  postgresql_ping:
+    db: '{{ db_default }}'
+    login_user: '{{ pg_user }}'
+    login_port: '{{ replica_port }}'
+
+- name: Define server version
+  set_fact:
+    pg_major_version: '{{ result.server_version.major }}'
+    pg_minor_version: '{{ result.server_version.minor }}'
+
+- name: Print PostgreSQL version
+  debug:
+    msg: 'PostgreSQL version is {{ pg_major_version }}.{{ pg_minor_version }}'
--- a/test/integration/targets/setup_postgresql_replication/templates/master_postgresql.conf.j2
+++ b/test/integration/targets/setup_postgresql_replication/templates/master_postgresql.conf.j2
@ -0,0 +1,28 @@
+# Important parameters:
+listen_addresses='*'
+port = {{ master_port }}
+wal_level = logical
+max_wal_senders = 8
+track_commit_timestamp = on
+max_replication_slots = 10
+
+# Unimportant parameters:
+max_connections=10
+shared_buffers=8MB
+dynamic_shared_memory_type=posix
+log_destination='stderr'
+logging_collector=on
+log_directory='log'
+log_filename='postgresql-%a.log'
+log_truncate_on_rotation=on
+log_rotation_age=1d
+log_rotation_size=0 
+log_line_prefix='%m[%p]'
+log_timezone='W-SU'
+datestyle='iso,mdy'
+timezone='W-SU'
+lc_messages='en_US.UTF-8'           
+lc_monetary='en_US.UTF-8'           
+lc_numeric='en_US.UTF-8'            
+lc_time='en_US.UTF-8'               
+default_text_search_config='pg_catalog.english'
--- a/test/integration/targets/setup_postgresql_replication/templates/pg_hba.conf.j2
+++ b/test/integration/targets/setup_postgresql_replication/templates/pg_hba.conf.j2
@ -0,0 +1,7 @@
+local  all          all                                trust
+local  replication  logical_replication                trust
+host   replication  logical_replication  127.0.0.1/32  trust
+host   replication  logical_replication  0.0.0.0/0     trust
+local  all          logical_replication                trust
+host   all          logical_replication  127.0.0.1/32  trust
+host   all          logical_replication  0.0.0.0/0     trust
--- a/test/integration/targets/setup_postgresql_replication/templates/replica_postgresql.conf.j2
+++ b/test/integration/targets/setup_postgresql_replication/templates/replica_postgresql.conf.j2
@ -0,0 +1,28 @@
+# Important parameters:
+listen_addresses='*'
+port = {{ replica_port }}
+wal_level = logical
+max_wal_senders = 8
+track_commit_timestamp = on
+max_replication_slots = 10
+
+# Unimportant parameters:
+max_connections=10
+shared_buffers=8MB
+dynamic_shared_memory_type=posix
+log_destination='stderr'
+logging_collector=on
+log_directory='log'
+log_filename='postgresql-%a.log'
+log_truncate_on_rotation=on
+log_rotation_age=1d
+log_rotation_size=0 
+log_line_prefix='%m[%p]'
+log_timezone='W-SU'
+datestyle='iso,mdy'
+timezone='W-SU'
+lc_messages='en_US.UTF-8'           
+lc_monetary='en_US.UTF-8'           
+lc_numeric='en_US.UTF-8'            
+lc_time='en_US.UTF-8'               
+default_text_search_config='pg_catalog.english'