Skip to content

Commit 01640cc

Browse files
eryx12o45eryx12o45
committed
fix 'fail' tests
1 parent f89e238 commit 01640cc

File tree

3 files changed

+107
-58
lines changed

3 files changed

+107
-58
lines changed

‎tests/unit/plugins/module_utils/test_mariadb_replication.py‎

Lines changed: 43 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -7,23 +7,7 @@
77
import pytest
88

99
from ansible_collections.community.mysql.plugins.module_utils.implementations.mariadb.replication import uses_replica_terminology
10-
from ..utils import dummy_cursor_class
11-
12-
13-
class MockCursor:
14-
def __init__(self, status="ONLINE"):
15-
self.status = status
16-
self.executed_queries = []
17-
18-
def execute(self, query):
19-
if self.status == "ERROR":
20-
raise Exception("Mocked execution error")
21-
self.executed_queries.append(query)
22-
23-
def fetchone(self):
24-
if "group_replication_status" in self.executed_queries[-1]:
25-
return ["group_replication_status", self.status]
26-
return None
10+
from ..utils import dummy_cursor_class, MockCursor
2711

2812

2913
@pytest.mark.parametrize(
@@ -89,25 +73,55 @@ def test_start_group_replication_fail():
8973
"""Test startgroupreplication function with failure."""
9074
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
9175

92-
cursor = MockCursor(status="ERROR")
93-
module = type('obj', (object,), {
94-
'fail_json': lambda msg: None,
95-
})
76+
# Create a mock mysql_driver with a Warning attribute
77+
class MockDriver:
78+
Warning = MockCursor.Warning
79+
80+
# Save the original mysql_driver
81+
from ansible_collections.community.mysql.plugins.modules import mysql_replication
82+
original_driver = mysql_replication.mysql_driver
9683

97-
result = startgroupreplication(module, cursor, [], True)
84+
try:
85+
# Replace with our mock driver
86+
mysql_replication.mysql_driver = MockDriver
9887

99-
assert result is False
88+
cursor = MockCursor(status="ERROR")
89+
module = type('obj', (object,), {
90+
'fail_json': lambda msg: None,
91+
})
92+
93+
result = startgroupreplication(module, cursor, [], True)
94+
95+
assert result is False
96+
finally:
97+
# Restore the original driver
98+
mysql_replication.mysql_driver = original_driver
10099

101100

102101
def test_stop_group_replication_fail():
103102
"""Test stopgroupreplication function with failure."""
104103
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
105104

106-
cursor = MockCursor(status="ERROR")
107-
module = type('obj', (object,), {
108-
'fail_json': lambda msg: None,
109-
})
105+
# Create a mock mysql_driver with a Warning attribute
106+
class MockDriver:
107+
Warning = MockCursor.Warning
108+
109+
# Save the original mysql_driver
110+
from ansible_collections.community.mysql.plugins.modules import mysql_replication
111+
original_driver = mysql_replication.mysql_driver
112+
113+
try:
114+
# Replace with our mock driver
115+
mysql_replication.mysql_driver = MockDriver
116+
117+
cursor = MockCursor(status="ERROR")
118+
module = type('obj', (object,), {
119+
'fail_json': lambda msg: None,
120+
})
110121

111-
result = stopgroupreplication(module, cursor, True)
122+
result = stopgroupreplication(module, cursor, True)
112123

113-
assert result is False
124+
assert result is False
125+
finally:
126+
# Restore the original driver
127+
mysql_replication.mysql_driver = original_driver

‎tests/unit/plugins/module_utils/test_mysql_replication.py‎

Lines changed: 43 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -7,23 +7,7 @@
77
import pytest
88

99
from ansible_collections.community.mysql.plugins.module_utils.implementations.mysql.replication import uses_replica_terminology
10-
from ..utils import dummy_cursor_class
11-
12-
13-
class MockCursor:
14-
def __init__(self, status="ONLINE"):
15-
self.status = status
16-
self.executed_queries = []
17-
18-
def execute(self, query):
19-
if self.status == "ERROR":
20-
raise Exception("Mocked execution error")
21-
self.executed_queries.append(query)
22-
23-
def fetchone(self):
24-
if "group_replication_status" in self.executed_queries[-1]:
25-
return ["group_replication_status", self.status]
26-
return None
10+
from ..utils import dummy_cursor_class, MockCursor
2711

2812

2913
@pytest.mark.parametrize(
@@ -93,25 +77,55 @@ def test_start_group_replication_fail():
9377
"""Test startgroupreplication function with failure."""
9478
from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication
9579

96-
cursor = MockCursor(status="ERROR")
97-
module = type('obj', (object,), {
98-
'fail_json': lambda msg: None,
99-
})
80+
# Create a mock mysql_driver with a Warning attribute
81+
class MockDriver:
82+
Warning = MockCursor.Warning
83+
84+
# Save the original mysql_driver
85+
from ansible_collections.community.mysql.plugins.modules import mysql_replication
86+
original_driver = mysql_replication.mysql_driver
10087

101-
result = startgroupreplication(module, cursor, [], True)
88+
try:
89+
# Replace with our mock driver
90+
mysql_replication.mysql_driver = MockDriver
10291

103-
assert result is False
92+
cursor = MockCursor(status="ERROR")
93+
module = type('obj', (object,), {
94+
'fail_json': lambda msg: None,
95+
})
96+
97+
result = startgroupreplication(module, cursor, [], True)
98+
99+
assert result is False
100+
finally:
101+
# Restore the original driver
102+
mysql_replication.mysql_driver = original_driver
104103

105104

106105
def test_stop_group_replication_fail():
107106
"""Test stopgroupreplication function with failure."""
108107
from ansible_collections.community.mysql.plugins.modules.mysql_replication import stopgroupreplication
109108

110-
cursor = MockCursor(status="ERROR")
111-
module = type('obj', (object,), {
112-
'fail_json': lambda msg: None,
113-
})
109+
# Create a mock mysql_driver with a Warning attribute
110+
class MockDriver:
111+
Warning = MockCursor.Warning
112+
113+
# Save the original mysql_driver
114+
from ansible_collections.community.mysql.plugins.modules import mysql_replication
115+
original_driver = mysql_replication.mysql_driver
116+
117+
try:
118+
# Replace with our mock driver
119+
mysql_replication.mysql_driver = MockDriver
120+
121+
cursor = MockCursor(status="ERROR")
122+
module = type('obj', (object,), {
123+
'fail_json': lambda msg: None,
124+
})
114125

115-
result = stopgroupreplication(module, cursor, True)
126+
result = stopgroupreplication(module, cursor, True)
116127

117-
assert result is False
128+
assert result is False
129+
finally:
130+
# Restore the original driver
131+
mysql_replication.mysql_driver = original_driver

‎tests/unit/plugins/utils.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,3 +17,24 @@ def fetchone(self):
1717

1818
elif self.ret_val_type == 'list':
1919
return [self.output]
20+
21+
22+
class MockCursor:
23+
def __init__(self, status="ONLINE"):
24+
self.status = status
25+
self.executed_queries = []
26+
27+
def execute(self, query):
28+
self.executed_queries.append(query)
29+
if self.status == "ERROR":
30+
# Create a custom exception that mimics mysql_driver.Warning
31+
class MockWarning(Exception):
32+
pass
33+
# Make it available as a class attribute for tests to use
34+
MockCursor.Warning = MockWarning
35+
raise MockWarning("Mocked execution error")
36+
37+
def fetchone(self):
38+
if len(self.executed_queries) > 0 and "group_replication_status" in self.executed_queries[-1]:
39+
return ["group_replication_status", self.status]
40+
return None

0 commit comments

Comments
 (0)