From 7c376590e72f866d4bfbb61f608411ca0b7c9ccb Mon Sep 17 00:00:00 2001 From: Sebastian Pfahl Date: Mon, 14 Jul 2025 21:30:15 +0200 Subject: [PATCH] fix more tests --- .../module_utils/test_mariadb_replication.py | 8 +++----- .../module_utils/test_mysql_replication.py | 20 +++++-------------- 2 files changed, 8 insertions(+), 20 deletions(-) diff --git a/tests/unit/plugins/module_utils/test_mariadb_replication.py b/tests/unit/plugins/module_utils/test_mariadb_replication.py index 7442e30..7aaa667 100644 --- a/tests/unit/plugins/module_utils/test_mariadb_replication.py +++ b/tests/unit/plugins/module_utils/test_mariadb_replication.py @@ -41,10 +41,10 @@ def test_uses_replica_terminology(f_output, c_output, c_ret_type): @pytest.mark.parametrize( 'user,password,expected_query', [ - (None, None, "START GROUP_REPLICATION"), + (None, None, "START GROUP_REPLICATION "), ("repl_user", None, "START GROUP_REPLICATION USER='repl_user'"), - (None, "repl_pass", "START GROUP_REPLICATION"), - ("repl_user", "repl_pass", "START GROUP_REPLICATION USER='repl_user' PASSWORD='repl_pass'"), + (None, "repl_pass", "START GROUP_REPLICATION PASSWORD='repl_pass'"), + ("repl_user", "repl_pass", "START GROUP_REPLICATION USER='repl_user', PASSWORD='repl_pass'"), ] ) def test_start_group_replication(user, password, expected_query): @@ -66,7 +66,6 @@ def test_start_group_replication(user, password, expected_query): assert result is True assert cursor.executed_queries[0] == expected_query - assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" def test_stop_group_replication(): @@ -82,7 +81,6 @@ def test_stop_group_replication(): assert result is True assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION" - assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" def test_start_group_replication_fail(): diff --git a/tests/unit/plugins/module_utils/test_mysql_replication.py b/tests/unit/plugins/module_utils/test_mysql_replication.py index 19ba12b..8768b1e 100644 --- a/tests/unit/plugins/module_utils/test_mysql_replication.py +++ b/tests/unit/plugins/module_utils/test_mysql_replication.py @@ -45,10 +45,10 @@ def test_uses_replica_terminology(f_output, c_output, c_ret_type): @pytest.mark.parametrize( 'user,password,expected_query', [ - (None, None, "START GROUP_REPLICATION"), + (None, None, "START GROUP_REPLICATION "), ("repl_user", None, "START GROUP_REPLICATION USER='repl_user'"), - (None, "repl_pass", "START GROUP_REPLICATION"), - ("repl_user", "repl_pass", "START GROUP_REPLICATION USER='repl_user' PASSWORD='repl_pass'"), + (None, "repl_pass", "START GROUP_REPLICATION PASSWORD='repl_pass'"), + ("repl_user", "repl_pass", "START GROUP_REPLICATION USER='repl_user', PASSWORD='repl_pass'"), ] ) def test_start_group_replication(user, password, expected_query): @@ -62,15 +62,14 @@ def test_start_group_replication(user, password, expected_query): chm = [] if user: - chm.append(" USER='%s'" % user) + chm.append("USER='%s'" % user) if password: - chm.append(" PASSWORD='%s'" % password) + chm.append("PASSWORD='%s'" % password) result = startgroupreplication(module, cursor, chm, False) assert result is True assert cursor.executed_queries[0] == expected_query - assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" def test_stop_group_replication(): @@ -86,22 +85,17 @@ def test_stop_group_replication(): assert result is True assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION" - assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" def test_start_group_replication_fail(): """Test startgroupreplication function with failure.""" from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication - import pymysql cursor = MockCursor(status="ERROR") module = type('obj', (object,), { 'fail_json': lambda msg: None, }) - # Mock the Warning exception - pymysql.Warning = Exception - result = startgroupreplication(module, cursor, [], True) assert result is False @@ -110,16 +104,12 @@ def test_start_group_replication_fail(): def test_stop_group_replication_fail(): """Test stopgroupreplication function with failure.""" from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication - import pymysql cursor = MockCursor(status="ERROR") module = type('obj', (object,), { 'fail_json': lambda msg: None, }) - # Mock the Warning exception - pymysql.Warning = Exception - result = stopgroupreplication(module, cursor, True) assert result is False