From 386cef18ed05b45c043618827825071724f0b16e Mon Sep 17 00:00:00 2001 From: Andrey Klychkov Date: Sat, 18 May 2019 09:33:45 +0300 Subject: [PATCH] module_utils.postgres: added unittests (#56381) --- lib/ansible/module_utils/postgres.py | 23 +- .../module_utils/postgresql/test_postgres.py | 292 ++++++++++++++++++ 2 files changed, 311 insertions(+), 4 deletions(-) create mode 100644 test/units/module_utils/postgresql/test_postgres.py diff --git a/lib/ansible/module_utils/postgres.py b/lib/ansible/module_utils/postgres.py index 6000f4e8a5a..20c70519778 100644 --- a/lib/ansible/module_utils/postgres.py +++ b/lib/ansible/module_utils/postgres.py @@ -27,9 +27,9 @@ # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE # USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +psycopg2 = None # This line needs for unit tests try: import psycopg2 - from psycopg2.extras import DictCursor HAS_PSYCOPG2 = True except ImportError: HAS_PSYCOPG2 = False @@ -55,6 +55,11 @@ def ensure_libs(sslrootcert=None): def postgres_common_argument_spec(): + """ + Return a dictionary with connection options. + + The options are commonly used by most of PostgreSQL modules. + """ return dict( login_user=dict(default='postgres'), login_password=dict(default='', no_log=True), @@ -67,6 +72,7 @@ def postgres_common_argument_spec(): def ensure_required_libs(module): + """Check required libraries.""" if not HAS_PSYCOPG2: module.fail_json(msg=missing_required_lib('psycopg2')) @@ -75,7 +81,14 @@ def ensure_required_libs(module): def connect_to_db(module, autocommit=False, fail_on_conn=True, warn_db_default=True): + """Return psycopg2 connection object. + Keyword arguments: + module -- object of ansible.module_utils.basic.AnsibleModule class + autocommit -- commit automatically (default False) + fail_on_conn -- fail if connection failed or just warn and return None (default True) + warn_db_default -- warn that the default DB is used (default True) + """ ensure_required_libs(module) # To use defaults values, keyword arguments must be absent, so @@ -110,22 +123,24 @@ def connect_to_db(module, autocommit=False, fail_on_conn=True, warn_db_default=T if is_localhost and module.params["login_unix_socket"] != "": kw["host"] = module.params["login_unix_socket"] + db_connection = None try: db_connection = psycopg2.connect(**kw) if autocommit: - if psycopg2.__version__ >= '2.4.2': + if LooseVersion(psycopg2.__version__) >= LooseVersion('2.4.2'): db_connection.set_session(autocommit=True) else: db_connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT) # Switch role, if specified: - cursor = db_connection.cursor(cursor_factory=DictCursor) if module.params.get('session_role'): + cursor = db_connection.cursor(cursor_factory=psycopg2.extras.DictCursor) try: cursor.execute('SET ROLE %s' % module.params['session_role']) except Exception as e: module.fail_json(msg="Could not switch role: %s" % to_native(e)) - cursor.close() + finally: + cursor.close() except TypeError as e: if 'sslrootcert' in e.args[0]: diff --git a/test/units/module_utils/postgresql/test_postgres.py b/test/units/module_utils/postgresql/test_postgres.py new file mode 100644 index 00000000000..0f5d4260c2d --- /dev/null +++ b/test/units/module_utils/postgresql/test_postgres.py @@ -0,0 +1,292 @@ +# Copyright: (c) 2019, Andrew Klychkov (@Andersson007) +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +import pytest + +import ansible.module_utils.postgres as pg + + +class TestPostgresCommonArgSpec(): + + """ + Namespace for testing postgresql_common_arg_spec() function. + """ + + def test_postgres_common_argument_spec(self): + """ + Test for postgresql_common_arg_spec() function. + + The tested function just returns a dictionary with the default + parameters and their values for PostgreSQL modules. + The return and expected dictionaries must be compared. + """ + expected_dict = dict( + login_user=dict(default='postgres'), + login_password=dict(default='', no_log=True), + login_host=dict(default=''), + login_unix_socket=dict(default=''), + port=dict(type='int', default=5432, aliases=['login_port']), + ssl_mode=dict( + default='prefer', + choices=['allow', 'disable', 'prefer', 'require', 'verify-ca', 'verify-full'] + ), + ca_cert=dict(aliases=['ssl_rootcert']), + ) + assert pg.postgres_common_argument_spec() == expected_dict + + +@pytest.fixture +def m_psycopg2(): + """Return mock object for psycopg2 emulation.""" + global Cursor + Cursor = None + + class Cursor(): + def __init__(self): + self.passed_query = None + + def execute(self, query): + self.passed_query = query + + def close(self): + pass + + global DbConnection + DbConnection = None + + class DbConnection(): + def __init__(self): + pass + + def cursor(self, cursor_factory=None): + return Cursor() + + def set_session(self, autocommit=None): + pass + + def set_isolation_level(self, isolevel): + pass + + class Extras(): + def __init__(self): + self.DictCursor = True + + class Extensions(): + def __init__(self): + self.ISOLATION_LEVEL_AUTOCOMMIT = True + + class DummyPsycopg2(): + def __init__(self): + self.__version__ = '2.4.3' + self.extras = Extras() + self.extensions = Extensions() + + def connect(self, host=None, port=None, user=None, + password=None, sslmode=None, sslrootcert=None): + if user == 'Exception': + raise Exception() + + return DbConnection() + + return DummyPsycopg2() + + +class TestEnsureReqLibs(): + + """ + Namespace for testing ensure_required_libs() function. + + If there is something wrong with libs, the function invokes fail_json() + method of AnsibleModule object passed as an argument called 'module'. + Therefore we must check: + 1. value of err_msg attribute of m_ansible_module mock object. + """ + + @pytest.fixture + def m_ansible_module(self, scope='class'): + """Return an object of dummy AnsibleModule class.""" + class Dummym_ansible_module(): + def __init__(self): + self.params = {'ca_cert': False} + self.err_msg = '' + + def fail_json(self, msg): + self.err_msg = msg + + return Dummym_ansible_module() + + def test_ensure_req_libs_has_not_psycopg2(self, m_ansible_module): + """Test ensure_required_libs() with psycopg2 is None.""" + # HAS_PSYCOPG2 is False by default + pg.ensure_required_libs(m_ansible_module) + assert 'Failed to import the required Python library (psycopg2)' in m_ansible_module.err_msg + + def test_ensure_req_libs_has_psycopg2(self, m_ansible_module, monkeypatch): + """Test ensure_required_libs() with psycopg2 is not None.""" + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + + pg.ensure_required_libs(m_ansible_module) + assert m_ansible_module.err_msg == '' + + def test_ensure_req_libs_ca_cert(self, m_ansible_module, m_psycopg2, monkeypatch): + """ + Test with module.params['ca_cert'], psycopg2 version is suitable. + """ + m_ansible_module.params['ca_cert'] = True + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + pg.ensure_required_libs(m_ansible_module) + assert m_ansible_module.err_msg == '' + + def test_ensure_req_libs_ca_cert_low_psycopg2_ver(self, m_ansible_module, m_psycopg2, monkeypatch): + """ + Test with module.params['ca_cert'], psycopg2 version is wrong. + """ + m_ansible_module.params['ca_cert'] = True + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + # Set wrong psycopg2 version number: + psycopg2 = m_psycopg2 + psycopg2.__version__ = '2.4.2' + monkeypatch.setattr(pg, 'psycopg2', psycopg2) + + pg.ensure_required_libs(m_ansible_module) + assert 'psycopg2 must be at least 2.4.3' in m_ansible_module.err_msg + + +class TestConnectToDb(): + + """ + Namespace for testing connect_to_db() function. + + When some connection errors occure connect_to_db() caught any of them + and invoke fail_json() or warn() methods of AnsibleModule object + depending on the passed parameters. + connect_to_db may return db_connection object or None if errors occured. + Therefore we must check: + 1. Values of err_msg and warn_msg attributes of m_ansible_module mock object. + 2. Types of return objects (db_connection and cursor). + """ + + @pytest.fixture + def m_ansible_module(self, scope='class'): + """Return an object of dummy AnsibleModule class.""" + class DummyAnsibleModule(): + def __init__(self): + self.params = pg.postgres_common_argument_spec() + self.err_msg = '' + self.warn_msg = '' + + def fail_json(self, msg): + self.err_msg = msg + + def warn(self, msg): + self.warn_msg = msg + + return DummyAnsibleModule() + + def test_connect_to_db(self, m_ansible_module, monkeypatch, m_psycopg2): + """Test connect_to_db(), common test.""" + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + db_connection = pg.connect_to_db(m_ansible_module) + cursor = db_connection.cursor() + # if errors, db_connection returned as None: + assert isinstance(db_connection, DbConnection) + assert isinstance(cursor, Cursor) + assert m_ansible_module.err_msg == '' + # The default behaviour, normal in this case: + assert 'Database name has not been passed' in m_ansible_module.warn_msg + + def test_session_role(self, m_ansible_module, monkeypatch, m_psycopg2): + """Test connect_to_db(), switch on session_role.""" + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + m_ansible_module.params['session_role'] = 'test_role' + db_connection = pg.connect_to_db(m_ansible_module) + cursor = db_connection.cursor() + # if errors, db_connection returned as None: + assert isinstance(db_connection, DbConnection) + assert isinstance(cursor, Cursor) + assert m_ansible_module.err_msg == '' + # The default behaviour, normal in this case: + assert 'Database name has not been passed' in m_ansible_module.warn_msg + + def test_warn_db_default_non_default(self, m_ansible_module, monkeypatch, m_psycopg2): + """ + Test connect_to_db(), warn_db_default arg passed as False (by default is True). + """ + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + db_connection = pg.connect_to_db(m_ansible_module, warn_db_default=False) + cursor = db_connection.cursor() + # if errors, db_connection returned as None: + assert isinstance(db_connection, DbConnection) + assert isinstance(cursor, Cursor) + assert m_ansible_module.err_msg == '' + assert m_ansible_module.warn_msg == '' + # pay attention that warn_db_defaul=True has been checked + # in the previous tests by + # assert('Database name has not been passed' in m_ansible_module.warn_msg) + # because of this is the default behavior + + def test_fail_on_conn_true(self, m_ansible_module, monkeypatch, m_psycopg2): + """ + Test connect_to_db(), fail_on_conn arg passed as True (the default behavior). + """ + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + m_ansible_module.params['login_user'] = 'Exception' # causes Exception + + db_connection = pg.connect_to_db(m_ansible_module, fail_on_conn=True) + + assert 'unable to connect to database' in m_ansible_module.err_msg + assert db_connection is None + + def test_fail_on_conn_false(self, m_ansible_module, monkeypatch, m_psycopg2): + """ + Test connect_to_db(), fail_on_conn arg passed as False. + """ + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + m_ansible_module.params['login_user'] = 'Exception' # causes Exception + + db_connection = pg.connect_to_db(m_ansible_module, fail_on_conn=False) + + assert m_ansible_module.err_msg == '' + assert 'PostgreSQL server is unavailable' in m_ansible_module.warn_msg + assert db_connection is None + + def test_autocommit_true(self, m_ansible_module, monkeypatch, m_psycopg2): + """ + Test connect_to_db(), autocommit arg passed as True (the default is False). + """ + monkeypatch.setattr(pg, 'HAS_PSYCOPG2', True) + + # case 1: psycopg2.__version >= 2.4.2 (the default in m_psycopg2) + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + db_connection = pg.connect_to_db(m_ansible_module, autocommit=True) + cursor = db_connection.cursor() + + # if errors, db_connection returned as None: + assert isinstance(db_connection, DbConnection) + assert isinstance(cursor, Cursor) + assert m_ansible_module.err_msg == '' + + # case 2: psycopg2.__version < 2.4.2 + m_psycopg2.__version__ = '2.4.1' + monkeypatch.setattr(pg, 'psycopg2', m_psycopg2) + + db_connection = pg.connect_to_db(m_ansible_module, autocommit=True) + cursor = db_connection.cursor() + + # if errors, db_connection returned as None: + assert isinstance(db_connection, DbConnection) + assert isinstance(cursor, Cursor) + assert 'psycopg2 must be at least 2.4.3' in m_ansible_module.err_msg