diff --git a/defaults/main.yml b/defaults/main.yml index f575734c..0392eac2 100644 --- a/defaults/main.yml +++ b/defaults/main.yml @@ -146,6 +146,8 @@ __logging_no_subsets_facts: __logging_required_facts_subsets: "{{ ['!all', '!min'] + __logging_required_facts | difference(__logging_no_subsets_facts) }}" +__logging_write_log_file: false + # BEGIN - DO NOT EDIT THIS BLOCK - rh distros variables # Ansible distribution identifiers that the role treats like RHEL __logging_rh_distros: diff --git a/library/sr_fingerprint.py b/library/sr_fingerprint.py index 593dd39f..8e4f473f 100644 --- a/library/sr_fingerprint.py +++ b/library/sr_fingerprint.py @@ -7,29 +7,148 @@ DOCUMENTATION = """ --- module: sr_fingerprint -short_description: Write a message string to syslog using Ansible C(module.log) function. +short_description: Write role fingerprint data to syslog and optionally to a JSONL log file description: - - Writes the given string to the system log using Ansible C(module.log) function. + - Collects role fingerprint data into a canonical record and writes it to + syslog using Ansible C(module.log) as C(key=value) pairs. + - Optionally appends the same record as a JSON line to a log file + (one JSON object per line, JSONL format), by default + C(/var/log/sysroles.jsonl). + - Playbook variables are not available inside modules automatically. Roles + pass C(role_name), C(role_path), C(ansible_play_hosts_all), + C(distribution), and C(distribution_version) from the task. + - C(ansible_check_mode) is collected from the module execution context. - Intended for role-internal or diagnostic use. author: Rich Megginson (@richm) options: - sr_message: - description: Text to record in syslog. + status: + description: Role execution status. type: str required: true + choices: + - begin + - success + write_log_file: + description: >- + If C(true), append fingerprint data to the JSONL log file. + Defaults to C(false). + type: bool + default: false + log_file: + description: >- + Path to the JSONL log file. A lock sidecar (C(.lock)) + is created next to the log file for cross-process safety. + type: path + default: /var/log/sysroles.jsonl + max_log_size: + description: >- + Maximum log file size in bytes. When appending a new record + would exceed this limit, the oldest records are removed first. + Set to C(0) to disable trimming. + type: int + default: 2000000 + role_name: + description: Name of the role, typically C({{ role_name }}). + type: str + required: true + role_path: + description: Path to the role, typically C({{ role_path }}). + type: path + required: true + ansible_play_hosts_all: + description: >- + All hosts in the play, typically C({{ ansible_play_hosts_all }}). + Used to derive C(play_hosts_number). + type: list + elements: str + required: true + distribution: + description: >- + OS distribution name, typically + C({{ ansible_facts["distribution"] }}). + type: str + default: "" + distribution_version: + description: >- + OS distribution version, typically + C({{ ansible_facts["distribution_version"] }}). + type: str + default: "" """ EXAMPLES = """ -- name: Record a fingerprint message in syslog +- name: Record role begin fingerprint to syslog only (not log file) + sr_fingerprint: + status: begin + role_name: bootloader + role_path: "{{ role_path }}" + ansible_play_hosts_all: "{{ ansible_play_hosts_all }}" + distribution: "{{ ansible_facts['distribution'] }}" + distribution_version: "{{ ansible_facts['distribution_version'] }}" + write_log_file: false + +- name: Record role success fingerprint sr_fingerprint: - sr_message: "system_role:ROLENAME" + status: success + role_name: bootloader + role_path: "{{ role_path }}" + ansible_play_hosts_all: "{{ ansible_play_hosts_all }}" + distribution: "{{ ansible_facts['distribution'] }}" + distribution_version: "{{ ansible_facts['distribution_version'] }}" + write_log_file: true """ -RETURN = r""" # """ +RETURN = r""" +fingerprint: + description: The fingerprint record written to syslog and optionally to the log file. + returned: always + type: dict + sample: + date: "2026-08-03T10:15:00+02:00" + role_name: network + role_path: /usr/share/ansible/roles/linux-system-roles.network + status: success + ansible_version: "2.16.3" + managed_node_distro: RedHat-9.4 + play_hosts_number: 3 + ansible_check_mode: false +message: + description: Informational message shown in check mode. + returned: check mode + type: str + sample: "Check mode: message not logged - [date=... role_name=...]" +jsonl_row: + description: The JSON line that would be appended to the log file. + returned: check mode and O(write_log_file=true) + type: str +log_file: + description: Path to the log file that would be written. + returned: check mode and O(write_log_file=true) + type: str +""" from ansible.module_utils.basic import AnsibleModule import datetime +import errno +import fcntl +import json +import os +import stat +import tempfile + +FINGERPRINT_FIELDS = ( + "date", + "role_name", + "role_path", + "status", + "ansible_version", + "managed_node_distro", + "play_hosts_number", + "ansible_check_mode", +) + +FINGERPRINT_SYSLOG_SEPARATOR = " " def _local_iso8601_no_microseconds(): @@ -51,33 +170,192 @@ def _local_iso8601_no_microseconds(): return datetime.datetime.now(utc).astimezone().replace(microsecond=0).isoformat() -def run_module(): - module_args = dict( - sr_message=dict(type="str", required=True), - ) +def _ensure_parent_dir(path): + parent = os.path.dirname(path) + if not parent: + return + if os.path.isdir(parent): + return + try: + os.makedirs(parent) + except OSError as exc: + # another process may have created the directory + if exc.errno != errno.EEXIST or not os.path.isdir(parent): + raise - module = AnsibleModule( - argument_spec=module_args, - supports_check_mode=True, - ) - log_message = "%s %s" % ( - module.params["sr_message"], - _local_iso8601_no_microseconds(), - ) +def _format_fingerprint_jsonl(record): + """Format the canonical fingerprint record as a single JSON line.""" + return json.dumps(record, separators=(",", ":"), sort_keys=False) + + +def _trim_log_file(log_file, size_needed): + """Remove oldest records until the file can accommodate size_needed bytes.""" + with open(log_file, "r") as log_fd: + lines = log_fd.readlines() + size_removed = 0 + while lines and size_removed < size_needed: + size_removed += len(lines.pop(0)) + orig_stat = os.stat(log_file) + dir_name = os.path.dirname(log_file) or "." + fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp") + try: + os.fchmod(fd, stat.S_IMODE(orig_stat.st_mode)) + try: + os.fchown(fd, orig_stat.st_uid, orig_stat.st_gid) + except OSError: + # not running as root; keep default ownership + pass + with os.fdopen(fd, "w") as tmp_fd: + tmp_fd.writelines(lines) + tmp_fd.flush() + os.fsync(tmp_fd.fileno()) + os.rename(tmp_path, log_file) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + # already removed or never created + pass + raise + + +def _write_jsonl_log(log_file, record, max_size=0): + _ensure_parent_dir(log_file) + new_line = _format_fingerprint_jsonl(record) + "\n" + lock_path = log_file + ".lock" + lock_fd = open(lock_path, "w") + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX) + try: + cur_size = os.path.getsize(log_file) + except OSError: + # file does not exist yet + cur_size = 0 + if max_size > 0 and cur_size + len(new_line) > max_size and cur_size > 0: + _trim_log_file(log_file, len(new_line)) + with open(log_file, "a") as log_fd: + log_fd.write(new_line) + finally: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + lock_fd.close() + + +def _get_managed_node_distro(distribution, distribution_version): + if distribution and distribution_version: + return "%s-%s" % (distribution, distribution_version) + return "unknown" + + +def _get_play_hosts_number(play_hosts_all): + return len(play_hosts_all) + + +def _get_ansible_version(module): + version = getattr(module, "ansible_version", None) + if version: + return version + return "unknown" + + +def _get_check_mode(module): + return bool(getattr(module, "check_mode", False)) + + +def _collect_fingerprint_record(module, status): + """Build the canonical fingerprint record used by all output formatters.""" + return { + "date": _local_iso8601_no_microseconds(), + "role_name": module.params["role_name"], + "role_path": module.params["role_path"], + "status": status, + "ansible_version": _get_ansible_version(module), + "managed_node_distro": _get_managed_node_distro( + module.params["distribution"], module.params["distribution_version"] + ), + "play_hosts_number": _get_play_hosts_number( + module.params["ansible_play_hosts_all"] + ), + "ansible_check_mode": _get_check_mode(module), + } + + +def _fingerprint_record_items(record): + return [(field, record[field]) for field in FINGERPRINT_FIELDS] + + +def _format_fingerprint_key_value(field, value): + text = "" if value is None else str(value) + if any(char in text for char in ' "='): + return '%s="%s"' % (field, text.replace('"', '""')) + return "%s=%s" % (field, text) + + +def _format_fingerprint_syslog(record): + """Format the canonical fingerprint record as key=value syslog text.""" + pairs = [ + _format_fingerprint_key_value(field, value) + for field, value in _fingerprint_record_items(record) + ] + return FINGERPRINT_SYSLOG_SEPARATOR.join(pairs) + + +def _handle_fingerprint(module): + max_log_size = module.params["max_log_size"] + if max_log_size < 0: + module.fail_json( + msg="max_log_size must be 0 or a positive integer, got %d" % max_log_size + ) + + fingerprint_record = _collect_fingerprint_record(module, module.params["status"]) + log_message = _format_fingerprint_syslog(fingerprint_record) if module.check_mode: - module.exit_json( + result = dict( changed=False, message="Check mode: message not logged - [%s]" % log_message, + fingerprint=fingerprint_record, ) + if module.params["write_log_file"]: + result["jsonl_row"] = _format_fingerprint_jsonl(fingerprint_record) + result["log_file"] = module.params["log_file"] + module.exit_json(**result) module.log(log_message) - # we don't actually change anything, so we're not changed - writing a log message - # is not considered a change - # also, we don't want to report changed every time the role runs - module.exit_json(changed=False) + if module.params["write_log_file"]: + log_file = module.params["log_file"] + try: + _write_jsonl_log( + log_file, fingerprint_record, module.params["max_log_size"] + ) + except (IOError, OSError) as exc: + module.fail_json( + msg="Failed to write fingerprint log file %s: %s" % (log_file, exc) + ) + + module.exit_json(changed=False, fingerprint=fingerprint_record) + + +def run_module(): + module_args = dict( + status=dict(type="str", required=True, choices=["begin", "success"]), + write_log_file=dict(type="bool", default=False), + log_file=dict(type="path", default="/var/log/sysroles.jsonl"), + max_log_size=dict(type="int", default=2000000), + role_name=dict(type="str", required=True), + role_path=dict(type="path", required=True), + ansible_play_hosts_all=dict(type="list", elements="str", required=True), + distribution=dict(type="str", default=""), + distribution_version=dict(type="str", default=""), + ) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + ) + + _handle_fingerprint(module) def main(): diff --git a/tasks/main.yml b/tasks/main.yml index 0a0e02fd..ed36fe4b 100644 --- a/tasks/main.yml +++ b/tasks/main.yml @@ -140,6 +140,10 @@ - name: Record role success fingerprint sr_fingerprint: - sr_message: >- - success system_role:logging ansible_version={{ ansible_version.full }} - {{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }} + status: success + role_name: logging + role_path: "{{ role_path }}" + ansible_play_hosts_all: "{{ ansible_play_hosts_all }}" + distribution: "{{ ansible_facts['distribution'] }}" + distribution_version: "{{ ansible_facts['distribution_version'] }}" + write_log_file: "{{ __logging_write_log_file }}" diff --git a/tasks/set_vars.yml b/tasks/set_vars.yml index 6aff4537..aaa1c0be 100644 --- a/tasks/set_vars.yml +++ b/tasks/set_vars.yml @@ -8,9 +8,13 @@ - name: Record role begin fingerprint sr_fingerprint: - sr_message: >- - begin system_role:logging ansible_version={{ ansible_version.full }} - {{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }} + status: begin + role_name: logging + role_path: "{{ role_path }}" + ansible_play_hosts_all: "{{ ansible_play_hosts_all }}" + distribution: "{{ ansible_facts['distribution'] }}" + distribution_version: "{{ ansible_facts['distribution_version'] }}" + write_log_file: "{{ __logging_write_log_file }}" - name: Determine if system is booted with systemd when: __logging_is_booted is not defined diff --git a/tests/tests_default.yml b/tests/tests_default.yml index 14aa2956..d9927a10 100644 --- a/tests/tests_default.yml +++ b/tests/tests_default.yml @@ -17,18 +17,47 @@ - name: Default run (NOOP) include_tasks: tasks/run_role_with_clear_facts.yml + vars: + __logging_write_log_file: true - # look for the exact module invocation, not some other message that might contain the string - - name: Check system journal contains role fingerprints + - name: Get fingerprint entries from journal ansible.builtin.shell: executable: /bin/bash cmd: >- set -eo pipefail; journalctl --since "{{ __journal_start_time }}" --no-pager | - grep -v " Invoked with" | grep "sr_fingerprint.*begin system_role:logging" || - { echo ERROR: BEGIN fingerprint not found; exit 1; }; - journalctl --since "{{ __journal_start_time }}" --no-pager | - grep -v " Invoked with" | grep "sr_fingerprint.*success system_role:logging" || - { echo ERROR: SUCCESS fingerprint not found; exit 1; } + grep -v " Invoked with" | + grep "sr_fingerprint.*role_name=logging" + register: __register_journal_fingerprints changed_when: false when: __register_dev_log.stat.exists + + - name: Check that the log file was written + ansible.builtin.slurp: + path: /var/log/sysroles.jsonl + register: __register_log_file + + - name: Verify log file and journal fingerprints + when: __register_dev_log.stat.exists + vars: + __journal_lines: "{{ __register_journal_fingerprints.stdout_lines }}" + __journal_begin: "{{ __journal_lines | select('search', 'status=begin') | list }}" + __journal_success: "{{ __journal_lines | select('search', 'status=success') | list }}" + __begin_date: "{{ (__journal_begin[0] | regex_search('date=([^ ]+)', '\\1'))[0] }}" + __success_date: "{{ (__journal_success[0] | regex_search('date=([^ ]+)', '\\1'))[0] }}" + __file_content: "{{ __register_log_file.content | b64decode }}" + block: + - name: Print contents of logs + debug: + var: item + loop: + - "{{ __file_content }}" + - "{{ __journal_lines }}" + + - name: Assert content is correct + ansible.builtin.assert: + that: + - __journal_begin | length > 0 + - __journal_success | length > 0 + - __begin_date in __file_content + - __success_date in __file_content diff --git a/tests/unit/test_sr_fingerprint.py b/tests/unit/test_sr_fingerprint.py new file mode 100644 index 00000000..6ae839b5 --- /dev/null +++ b/tests/unit/test_sr_fingerprint.py @@ -0,0 +1,414 @@ +# -*- coding: utf-8 -*- + +# Copyright: (c) 2026, Red Hat, Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for sr_fingerprint module helpers.""" + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import json +import os +import re +import tempfile +import unittest + +import sr_fingerprint + + +class _ExitJsonException(Exception): + def __init__(self, kwargs): + self.kwargs = kwargs + + +class _FailJsonException(Exception): + def __init__(self, kwargs): + self.kwargs = kwargs + + +class _FakeModule(object): + ansible_version = "2.16.3" + + def __init__(self, params=None, check_mode=False): + self.params = params or {} + self.check_mode = check_mode + self.logged = [] + + def log(self, msg): + self.logged.append(msg) + + def exit_json(self, **kwargs): + raise _ExitJsonException(kwargs) + + def fail_json(self, **kwargs): + raise _FailJsonException(kwargs) + + +def _cleanup_log(log_file): + for path in (log_file, log_file + ".lock"): + try: + os.unlink(path) + except OSError: + # file may not exist + pass + + +def _sample_fingerprint_record(): + return { + "date": "2026-06-10T12:00:00+00:00", + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "status": "begin", + "ansible_version": "2.16.3", + "managed_node_distro": "RedHat-9.4", + "play_hosts_number": 3, + "ansible_check_mode": False, + } + + +class TestSrFingerprint(unittest.TestCase): + def test_fingerprint_fields_match_record_keys(self): + record = _sample_fingerprint_record() + self.assertEqual(set(sr_fingerprint.FINGERPRINT_FIELDS), set(record.keys())) + + def test_format_fingerprint_syslog(self): + record = _sample_fingerprint_record() + message = sr_fingerprint._format_fingerprint_syslog(record) + self.assertEqual( + message, + "date=2026-06-10T12:00:00+00:00 role_name=systemd " + "role_path=/usr/share/ansible/roles/linux-system-roles.systemd status=begin " + "ansible_version=2.16.3 managed_node_distro=RedHat-9.4 " + "play_hosts_number=3 ansible_check_mode=False", + ) + for field in sr_fingerprint.FINGERPRINT_FIELDS: + self.assertIn("%s=" % field, message) + + def test_format_fingerprint_jsonl(self): + record = _sample_fingerprint_record() + line = sr_fingerprint._format_fingerprint_jsonl(record) + parsed = json.loads(line) + self.assertEqual(parsed, record) + + def test_collect_fingerprint_record_from_passed_inputs(self): + module = _FakeModule( + { + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "ansible_play_hosts_all": ["host1", "host2", "host3"], + "distribution": "RedHat", + "distribution_version": "9.4", + }, + check_mode=True, + ) + record = sr_fingerprint._collect_fingerprint_record(module, "begin") + self.assertEqual(record["role_name"], "systemd") + self.assertEqual( + record["role_path"], "/usr/share/ansible/roles/linux-system-roles.systemd" + ) + self.assertEqual(record["managed_node_distro"], "RedHat-9.4") + self.assertEqual(record["play_hosts_number"], 3) + self.assertTrue(record["ansible_check_mode"]) + self.assertEqual( + set(record.keys()), + set(sr_fingerprint.FINGERPRINT_FIELDS), + ) + + def test_get_managed_node_distro_from_params(self): + distro = sr_fingerprint._get_managed_node_distro("Fedora", "42") + self.assertEqual(distro, "Fedora-42") + + def test_get_managed_node_distro_missing(self): + self.assertEqual(sr_fingerprint._get_managed_node_distro("", ""), "unknown") + + def test_get_play_hosts_number(self): + self.assertEqual( + sr_fingerprint._get_play_hosts_number(["a", "b"]), + 2, + ) + self.assertEqual(sr_fingerprint._get_play_hosts_number([]), 0) + + def test_format_fingerprint_syslog_quotes_values_with_spaces(self): + record = _sample_fingerprint_record() + record["role_path"] = ( + "/usr/share/ansible/roles/linux-system-roles.systemd extra" + ) + message = sr_fingerprint._format_fingerprint_syslog(record) + self.assertIn( + 'role_path="/usr/share/ansible/roles/linux-system-roles.systemd extra"', + message, + ) + + def test_write_jsonl_log_appends_valid_json_lines(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + sr_fingerprint._write_jsonl_log(log_file, record) + sr_fingerprint._write_jsonl_log(log_file, record) + + with open(log_file, "r") as log_fd: + lines = log_fd.read().splitlines() + + self.assertEqual(len(lines), 2) + for line in lines: + parsed = json.loads(line) + self.assertEqual(parsed, record) + finally: + _cleanup_log(log_file) + + def test_write_jsonl_log_creates_parent_dir(self): + tmpdir = tempfile.mkdtemp() + log_file = os.path.join(tmpdir, "subdir", "fingerprint.jsonl") + + try: + record = _sample_fingerprint_record() + sr_fingerprint._write_jsonl_log(log_file, record) + + with open(log_file, "r") as log_fd: + parsed = json.loads(log_fd.readline()) + self.assertEqual(parsed["role_name"], "systemd") + finally: + subdir = os.path.dirname(log_file) + for name in os.listdir(subdir): + os.unlink(os.path.join(subdir, name)) + os.rmdir(subdir) + os.rmdir(tmpdir) + + def test_write_jsonl_log_preserves_types(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + sr_fingerprint._write_jsonl_log(log_file, record) + + with open(log_file, "r") as log_fd: + parsed = json.loads(log_fd.readline()) + + self.assertIsInstance(parsed["play_hosts_number"], int) + self.assertIsInstance(parsed["ansible_check_mode"], bool) + finally: + _cleanup_log(log_file) + + def test_trim_removes_oldest_lines(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + sample = dict(record, role_name="role_0") + line_size = len(sr_fingerprint._format_fingerprint_jsonl(sample) + "\n") + max_size = line_size * 5 + for _i in range(10): + record_copy = dict(record, role_name="role_%d" % _i) + sr_fingerprint._write_jsonl_log( + log_file, record_copy, max_size=max_size + ) + + with open(log_file, "r") as log_fd: + lines = log_fd.read().splitlines() + + self.assertEqual(len(lines), 5) + first = json.loads(lines[0]) + last = json.loads(lines[-1]) + self.assertEqual(first["role_name"], "role_5") + self.assertEqual(last["role_name"], "role_9") + finally: + _cleanup_log(log_file) + + def test_trim_multiple_lines(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + sample = dict(record, role_name="role_0") + line_size = len(sr_fingerprint._format_fingerprint_jsonl(sample) + "\n") + # Log contains more than 3 lines. + n_initial = 4 + max_size = line_size * n_initial + for _i in range(n_initial): + sr_fingerprint._write_jsonl_log( + log_file, + dict(record, role_name="role_%d" % _i), + max_size=max_size, + ) + + with open(log_file, "r") as log_fd: + initial_lines = log_fd.read().splitlines() + self.assertEqual(len(initial_lines), n_initial) + + # New record larger than one existing line (up to two) via a very + # long role_path, so trim removes exactly two oldest records. + base_record = dict(record, role_name="role_long", role_path="") + base_size = len( + sr_fingerprint._format_fingerprint_jsonl(base_record) + "\n" + ) + path_len = 2 * line_size - base_size + self.assertGreater(path_len, 0) + long_path = "x" * path_len + long_record = dict(record, role_name="role_long", role_path=long_path) + new_line = sr_fingerprint._format_fingerprint_jsonl(long_record) + "\n" + self.assertGreater(len(new_line), line_size) + self.assertLessEqual(len(new_line), 2 * line_size) + + sr_fingerprint._write_jsonl_log(log_file, long_record, max_size=max_size) + + with open(log_file, "r") as log_fd: + lines = log_fd.read().splitlines() + + # Exactly two oldest records removed; new record appended. + self.assertEqual(len(lines), n_initial - 2 + 1) + parsed = [json.loads(line) for line in lines] + role_names = [entry["role_name"] for entry in parsed] + self.assertEqual(role_names, ["role_2", "role_3", "role_long"]) + self.assertEqual(parsed[-1], long_record) + self.assertEqual(parsed[-1]["role_path"], long_path) + self.assertNotIn("role_0", role_names) + self.assertNotIn("role_1", role_names) + finally: + _cleanup_log(log_file) + + def test_trim_disabled_when_zero(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + for _i in range(20): + sr_fingerprint._write_jsonl_log(log_file, record, max_size=0) + + with open(log_file, "r") as log_fd: + lines = log_fd.read().splitlines() + + self.assertEqual(len(lines), 20) + finally: + _cleanup_log(log_file) + + def test_trim_no_op_when_under_limit(self): + with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as tmp: + log_file = tmp.name + + try: + record = _sample_fingerprint_record() + for _i in range(3): + sr_fingerprint._write_jsonl_log(log_file, record, max_size=2000000) + + with open(log_file, "r") as log_fd: + lines = log_fd.read().splitlines() + + self.assertEqual(len(lines), 3) + finally: + _cleanup_log(log_file) + + def test_handle_fingerprint_check_mode_without_log_file(self): + module = _FakeModule( + { + "status": "begin", + "write_log_file": False, + "max_log_size": 2000000, + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "ansible_play_hosts_all": ["host1"], + "distribution": "RedHat", + "distribution_version": "9.4", + }, + check_mode=True, + ) + with self.assertRaises(_ExitJsonException) as ctx: + sr_fingerprint._handle_fingerprint(module) + result = ctx.exception.kwargs + self.assertFalse(result["changed"]) + self.assertIn("Check mode", result["message"]) + self.assertIn("fingerprint", result) + self.assertNotIn("jsonl_row", result) + + def test_handle_fingerprint_check_mode_with_log_file(self): + log_path = os.path.join(tempfile.gettempdir(), "test_sr_fingerprint.jsonl") + module = _FakeModule( + { + "status": "success", + "write_log_file": True, + "log_file": log_path, + "max_log_size": 2000000, + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "ansible_play_hosts_all": ["host1"], + "distribution": "RedHat", + "distribution_version": "9.4", + }, + check_mode=True, + ) + with self.assertRaises(_ExitJsonException) as ctx: + sr_fingerprint._handle_fingerprint(module) + result = ctx.exception.kwargs + self.assertIn("jsonl_row", result) + self.assertEqual(result["log_file"], log_path) + parsed = json.loads(result["jsonl_row"]) + self.assertEqual(parsed["role_name"], "systemd") + + def test_handle_fingerprint_write_failure_calls_fail_json(self): + log_path = os.path.join(tempfile.gettempdir(), "test_write_fail.jsonl") + module = _FakeModule( + { + "status": "success", + "write_log_file": True, + "log_file": log_path, + "max_log_size": 2000000, + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "ansible_play_hosts_all": ["host1"], + "distribution": "RedHat", + "distribution_version": "9.4", + }, + check_mode=False, + ) + original = sr_fingerprint._write_jsonl_log + + def _raise_ioerror(*args, **kwargs): + raise IOError("disk full") + + sr_fingerprint._write_jsonl_log = _raise_ioerror + try: + with self.assertRaises(_FailJsonException) as ctx: + sr_fingerprint._handle_fingerprint(module) + self.assertIn( + "Failed to write fingerprint log file", ctx.exception.kwargs["msg"] + ) + finally: + sr_fingerprint._write_jsonl_log = original + + def test_handle_fingerprint_rejects_negative_max_log_size(self): + module = _FakeModule( + { + "status": "begin", + "write_log_file": False, + "max_log_size": -1, + "role_name": "systemd", + "role_path": "/usr/share/ansible/roles/linux-system-roles.systemd", + "ansible_play_hosts_all": ["host1"], + "distribution": "RedHat", + "distribution_version": "9.4", + }, + check_mode=False, + ) + with self.assertRaises(_FailJsonException) as ctx: + sr_fingerprint._handle_fingerprint(module) + self.assertIn( + "max_log_size must be 0 or a positive integer", + ctx.exception.kwargs["msg"], + ) + + def test_local_iso8601_no_microseconds_has_no_fraction(self): + timestamp = sr_fingerprint._local_iso8601_no_microseconds() + match = re.match( + r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:?\d{2}$", timestamp + ) + self.assertIsNotNone(match) + + +if __name__ == "__main__": + unittest.main()