From 12a95c444685382e3ed0b3ef9458cf54bba9a2dd Mon Sep 17 00:00:00 2001 From: Sebastian Pfahl Date: Tue, 15 Jul 2025 12:59:42 +0200 Subject: [PATCH] fix warning baseexception class error --- .../module_utils/test_mariadb_replication.py | 6 ++++-- .../module_utils/test_mysql_replication.py | 6 ++++-- tests/unit/plugins/utils.py | 18 +++++++++--------- 3 files changed, 17 insertions(+), 13 deletions(-) diff --git a/tests/unit/plugins/module_utils/test_mariadb_replication.py b/tests/unit/plugins/module_utils/test_mariadb_replication.py index a984d90..f5f3792 100644 --- a/tests/unit/plugins/module_utils/test_mariadb_replication.py +++ b/tests/unit/plugins/module_utils/test_mariadb_replication.py @@ -72,10 +72,11 @@ def test_stop_group_replication(): def test_start_group_replication_fail(): """Test startgroupreplication function with failure.""" from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication + from ..utils import MockWarning # Create a mock mysql_driver with a Warning attribute class MockDriver: - Warning = MockCursor.Warning + Warning = MockWarning # Save the original mysql_driver from ansible_collections.community.mysql.plugins.modules import mysql_replication @@ -101,10 +102,11 @@ 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 + from ..utils import MockWarning # Create a mock mysql_driver with a Warning attribute class MockDriver: - Warning = MockCursor.Warning + Warning = MockWarning # Save the original mysql_driver from ansible_collections.community.mysql.plugins.modules import mysql_replication diff --git a/tests/unit/plugins/module_utils/test_mysql_replication.py b/tests/unit/plugins/module_utils/test_mysql_replication.py index 6dfea29..ead1e87 100644 --- a/tests/unit/plugins/module_utils/test_mysql_replication.py +++ b/tests/unit/plugins/module_utils/test_mysql_replication.py @@ -76,10 +76,11 @@ def test_stop_group_replication(): def test_start_group_replication_fail(): """Test startgroupreplication function with failure.""" from ansible_collections.community.mysql.plugins.modules.mysql_replication import startgroupreplication + from ..utils import MockWarning # Create a mock mysql_driver with a Warning attribute class MockDriver: - Warning = MockCursor.Warning + Warning = MockWarning # Save the original mysql_driver from ansible_collections.community.mysql.plugins.modules import mysql_replication @@ -105,10 +106,11 @@ 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 + from ..utils import MockWarning # Create a mock mysql_driver with a Warning attribute class MockDriver: - Warning = MockCursor.Warning + Warning = MockWarning # Save the original mysql_driver from ansible_collections.community.mysql.plugins.modules import mysql_replication diff --git a/tests/unit/plugins/utils.py b/tests/unit/plugins/utils.py index fea9ad4..bea8514 100644 --- a/tests/unit/plugins/utils.py +++ b/tests/unit/plugins/utils.py @@ -19,22 +19,22 @@ class dummy_cursor_class(): return [self.output] -class MockCursor: - Warning = None +# Define MockWarning at module level to ensure it's a proper exception class +class MockWarning(Exception): + pass - def __init__(self, status="ONLINE", warning=None): + +class MockCursor: + # Set the Warning class at the class level + Warning = MockWarning + + def __init__(self, status="ONLINE"): self.status = status self.executed_queries = [] - self.Warning = warning def execute(self, query): self.executed_queries.append(query) if self.status == "ERROR": - # Create a custom exception that mimics mysql_driver.Warning - class MockWarning(Exception): - pass - # Make it available as a class attribute for tests to use - MockCursor.Warning = MockWarning raise MockWarning("Mocked execution error") def fetchone(self):