fix liniting

This commit is contained in:
Sebastian Pfahl 2025-07-14 17:19:32 +02:00
commit a2dff5c61c
2 changed files with 30 additions and 30 deletions

View file

@ -50,20 +50,20 @@ def test_uses_replica_terminology(f_output, c_output, c_ret_type):
def test_start_group_replication(user, password, expected_query): def test_start_group_replication(user, password, expected_query):
"""Test startgroupreplication function with different parameters.""" """Test startgroupreplication function with different parameters."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
cursor = MockCursor() cursor = MockCursor()
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
chm = [] chm = []
if user: if user:
chm.append(" USER='%s'" % user) chm.append(" USER='%s'" % user)
if password: if password:
chm.append(" PASSWORD='%s'" % password) chm.append(" PASSWORD='%s'" % password)
result = startgroupreplication(module, cursor, chm, False) result = startgroupreplication(module, cursor, chm, False)
assert result is True assert result is True
assert cursor.executed_queries[0] == expected_query assert cursor.executed_queries[0] == expected_query
assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';"
@ -72,14 +72,14 @@ def test_start_group_replication(user, password, expected_query):
def test_stop_group_replication(): def test_stop_group_replication():
"""Test stopgroupreplication function.""" """Test stopgroupreplication function."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
cursor = MockCursor() cursor = MockCursor()
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
result = stopgroupreplication(module, cursor, False) result = stopgroupreplication(module, cursor, False)
assert result is True assert result is True
assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION" assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION"
assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';"
@ -89,17 +89,17 @@ def test_start_group_replication_fail():
"""Test startgroupreplication function with failure.""" """Test startgroupreplication function with failure."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
import pymysql import pymysql
cursor = MockCursor(status="ERROR") cursor = MockCursor(status="ERROR")
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
# Mock the Warning exception # Mock the Warning exception
pymysql.Warning = Exception pymysql.Warning = Exception
result = startgroupreplication(module, cursor, [], True) result = startgroupreplication(module, cursor, [], True)
assert result is False assert result is False
@ -107,15 +107,15 @@ def test_stop_group_replication_fail():
"""Test stopgroupreplication function with failure.""" """Test stopgroupreplication function with failure."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
import pymysql import pymysql
cursor = MockCursor(status="ERROR") cursor = MockCursor(status="ERROR")
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
# Mock the Warning exception # Mock the Warning exception
pymysql.Warning = Exception pymysql.Warning = Exception
result = stopgroupreplication(module, cursor, True) result = stopgroupreplication(module, cursor, True)
assert result is False assert result is False

View file

@ -54,20 +54,20 @@ def test_uses_replica_terminology(f_output, c_output, c_ret_type):
def test_start_group_replication(user, password, expected_query): def test_start_group_replication(user, password, expected_query):
"""Test startgroupreplication function with different parameters.""" """Test startgroupreplication function with different parameters."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
cursor = MockCursor() cursor = MockCursor()
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
chm = [] chm = []
if user: if user:
chm.append(" USER='%s'" % user) chm.append(" USER='%s'" % user)
if password: if password:
chm.append(" PASSWORD='%s'" % password) chm.append(" PASSWORD='%s'" % password)
result = startgroupreplication(module, cursor, chm, False) result = startgroupreplication(module, cursor, chm, False)
assert result is True assert result is True
assert cursor.executed_queries[0] == expected_query assert cursor.executed_queries[0] == expected_query
assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';"
@ -76,14 +76,14 @@ def test_start_group_replication(user, password, expected_query):
def test_stop_group_replication(): def test_stop_group_replication():
"""Test stopgroupreplication function.""" """Test stopgroupreplication function."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
cursor = MockCursor() cursor = MockCursor()
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
result = stopgroupreplication(module, cursor, False) result = stopgroupreplication(module, cursor, False)
assert result is True assert result is True
assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION" assert cursor.executed_queries[0] == "STOP GROUP_REPLICATION"
assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';" assert cursor.executed_queries[1] == "SHOW STATUS LIKE 'group_replication_status';"
@ -93,17 +93,17 @@ def test_start_group_replication_fail():
"""Test startgroupreplication function with failure.""" """Test startgroupreplication function with failure."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
import pymysql import pymysql
cursor = MockCursor(status="ERROR") cursor = MockCursor(status="ERROR")
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
# Mock the Warning exception # Mock the Warning exception
pymysql.Warning = Exception pymysql.Warning = Exception
result = startgroupreplication(module, cursor, [], True) result = startgroupreplication(module, cursor, [], True)
assert result is False assert result is False
@ -111,15 +111,15 @@ def test_stop_group_replication_fail():
"""Test stopgroupreplication function with failure.""" """Test stopgroupreplication function with failure."""
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
import pymysql import pymysql
cursor = MockCursor(status="ERROR") cursor = MockCursor(status="ERROR")
module = type('obj', (object,), { module = type('obj', (object,), {
'fail_json': lambda msg: None, 'fail_json': lambda msg: None,
}) })
# Mock the Warning exception # Mock the Warning exception
pymysql.Warning = Exception pymysql.Warning = Exception
result = stopgroupreplication(module, cursor, True) result = stopgroupreplication(module, cursor, True)
assert result is False assert result is False