From 22d1f6a2b62bd0d2bdf60b0d41d8bb31ce330e2f Mon Sep 17 00:00:00 2001 From: Markus Frei Date: Mon, 25 May 2026 10:12:50 +0200 Subject: [PATCH] fix(plugins/modules/sqlite_query): fail the task on a failed query main() ignored the success flag returned by select(), so a failed query exited with changed=false and the error message smuggled into query_result - reporting success for a broken query. Check the flag and call fail_json with the error instead. Add a reusable Ansible module test harness (tests/ansible_harness.py: set_module_args + exit_json/fail_json patching, profile-aware so it works on ansible-core 2.15 through 2.21) and main()-level tests covering both the success and the failure path. --- CHANGELOG.md | 1 + plugins/modules/sqlite_query.py | 4 ++ tests/ansible_harness.py | 60 +++++++++++++++++++ tests/conftest.py | 3 + .../unit/plugins/modules/test_sqlite_query.py | 33 ++++++++++ 5 files changed, 101 insertions(+) create mode 100644 tests/ansible_harness.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d8f1fad3..b990216f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +* **plugin:sqlite_query**: A failed query now fails the task instead of reporting success with the error text in `query_result`. Playbooks that relied on the previous silent success will now correctly fail. * **plugin:sqlite_query**: A `REGEXP` query against a column that contains NULL values no longer fails; a NULL value simply does not match. * **plugin:uptimerobot_\***: The modules no longer crash when the UptimeRobot API returns a non-list response for a list endpoint; the response is passed through instead. * **plugin:nextcloud_occ_app_config, plugin:nextcloud_occ_system_config, plugin:uptimerobot_monitor, plugin:uptimerobot_psp**: Fixed their documentation so `ansible-doc` renders them again. A unit-test guard now catches this class of error for every in-house plugin. diff --git a/plugins/modules/sqlite_query.py b/plugins/modules/sqlite_query.py index c354b177..4131bd6d 100644 --- a/plugins/modules/sqlite_query.py +++ b/plugins/modules/sqlite_query.py @@ -228,6 +228,10 @@ def main(): if query_type == 'select': success, query_result = select(conn, query, named_args, fetchone=fetch_one, as_dict=as_dict) changed = False + if not success: + close(conn) + # query_result holds the error message when the query failed + module.fail_json(msg=query_result) close(conn) # in the event of a successful module execution, you will want to diff --git a/tests/ansible_harness.py b/tests/ansible_harness.py new file mode 100644 index 00000000..1f946f4e --- /dev/null +++ b/tests/ansible_harness.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8; py-indent-offset: 4 -*- +# +# Author: Linuxfabrik GmbH, Zurich, Switzerland +# Contact: info (at) linuxfabrik (dot) ch +# https://www.linuxfabrik.ch/ +# License: The Unlicense, see LICENSE file. + +"""Helpers to unit-test an Ansible module's main() function. + +Standard ansible-test pattern: feed arguments via set_module_args(), then +patch AnsibleModule.exit_json / fail_json so they raise instead of calling +sys.exit(), so a test can assert on the outcome. tests/conftest.py puts the +tests/ directory on sys.path so test modules can `import ansible_harness`. +""" + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import json +import unittest.mock + +from ansible.module_utils import basic +from ansible.module_utils.common.text.converters import to_bytes + + +class AnsibleExitJson(Exception): + """Raised in place of AnsibleModule.exit_json().""" + + +class AnsibleFailJson(Exception): + """Raised in place of AnsibleModule.fail_json().""" + + +def set_module_args(args): + """Prepare module arguments as if passed in by Ansible.""" + basic._ANSIBLE_ARGS = to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': args})) + # ansible-core 2.19+ also requires a serialization profile to decode the args. + if hasattr(basic, '_ANSIBLE_PROFILE'): + basic._ANSIBLE_PROFILE = 'legacy' + + +def _exit_json(self, **kwargs): + kwargs.setdefault('changed', False) + raise AnsibleExitJson(kwargs) + + +def _fail_json(self, **kwargs): + kwargs.setdefault('failed', True) + raise AnsibleFailJson(kwargs) + + +def patch_module(): + """Return a mock.patch.multiple context manager for exit_json / fail_json.""" + return unittest.mock.patch.multiple( + basic.AnsibleModule, + exit_json=_exit_json, + fail_json=_fail_json, + ) diff --git a/tests/conftest.py b/tests/conftest.py index 1e4b9f4c..117ffb1c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -39,4 +39,7 @@ def _make_collection_importable(): os.environ.setdefault('ANSIBLE_COLLECTIONS_PATH', str(root)) +# make the tests/ directory importable so test modules can `import ansible_harness` +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) + _make_collection_importable() diff --git a/tests/unit/plugins/modules/test_sqlite_query.py b/tests/unit/plugins/modules/test_sqlite_query.py index a162c3df..684892ae 100644 --- a/tests/unit/plugins/modules/test_sqlite_query.py +++ b/tests/unit/plugins/modules/test_sqlite_query.py @@ -22,6 +22,8 @@ import tempfile import unittest +import ansible_harness + from ansible_collections.linuxfabrik.lfops.plugins.modules import sqlite_query as mod @@ -91,5 +93,36 @@ def test_connect_to_unwritable_path(self): self.assertIn('failed', result.lower()) +class TestMain(unittest.TestCase): + """main() must fail the task on a failed query instead of reporting success.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp(prefix='lfops_sqlite_main_test_') + ok, conn = mod.connect(path=self.tmpdir, filename='main.db') + conn.execute('CREATE TABLE t (id INTEGER)') + conn.execute('INSERT INTO t VALUES (1)') + conn.commit() + mod.close(conn) + + def test_successful_query_exits_with_result(self): + ansible_harness.set_module_args({ + 'path': self.tmpdir, 'db': 'main.db', 'query': 'SELECT id FROM t', + }) + with ansible_harness.patch_module(): + with self.assertRaises(ansible_harness.AnsibleExitJson) as cm: + mod.main() + self.assertEqual(cm.exception.args[0]['query_result'], [{'id': 1}]) + self.assertFalse(cm.exception.args[0]['changed']) + + def test_failed_query_fails_the_task(self): + ansible_harness.set_module_args({ + 'path': self.tmpdir, 'db': 'main.db', 'query': 'SELECT * FROM does_not_exist', + }) + with ansible_harness.patch_module(): + with self.assertRaises(ansible_harness.AnsibleFailJson) as cm: + mod.main() + self.assertIn('Query failed', cm.exception.args[0]['msg']) + + if __name__ == '__main__': unittest.main()