fix(sqlite_query): fail the task on a failed query

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.

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

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,
+    )

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()

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()