7 Commits
Author SHA1 Message Date
marko 7497474653 Merge tag '1.4.0'
Ansible Lint / ansible_lint (map[ansible:2.16.* ansible_lint:24.* python:3.12]) (push) Skipped
Ansible Lint / ansible_lint (map[ansible:2.20.* ansible_lint:26.* python:3.13]) (push) Skipped
Check for ansible_managed variable use in comments / ansible_managed_var_comment (push) Skipped
Ansible Test / ansible_test (map[ansible:2-14 python:3.9]) (push) Skipped
Ansible Test / ansible_test (map[ansible:2-16 python:3.11]) (push) Skipped
Ansible Test / ansible_test (map[ansible:2-17 python:3.12]) (push) Skipped
Ansible Test / ansible_test (map[ansible:2-18 python:3.12]) (push) Skipped
Ansible Test / ansible_test (map[ansible:2-19 python:3.13]) (push) Skipped
Ansible Test / ansible_test (map[ansible:milestone python:3.13]) (push) Skipped
Ansible Test / ansible_test (map[ansible:2-20 python:3.13]) (push) Skipped
Tag, release, and publish role based on CHANGELOG.md push / tag_release_publish (push) Failing after 58s
Markdown Lint / markdownlint (push) Skipped
Test / scenario (map[env:container-ansible-core-2-16 image:centos-9-bootc]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-16 image:centos-9]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-17 image:centos-10-bootc]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-20 image:fedora-43]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-20 image:fedora-43-bootc]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-20 image:fedora-44-bootc]) (push) Skipped
Test / scenario (map[env:container-ansible-core-2-21 image:fedora-44]) (push) Skipped
Test / scenario (map[env:qemu-ansible-core-2-16 image:centos-9]) (push) Skipped
Test / scenario (map[env:qemu-ansible-core-2-17 image:centos-10]) (push) Skipped
Test / scenario (map[env:qemu-ansible-core-2-18 image:leap-15.6]) (push) Skipped
Test / scenario (map[env:qemu-ansible-core-2-21 image:fedora-44]) (push) Skipped
Test / scenario (map[env:qemu-ansible-core-2-20 image:fedora-43]) (push) Skipped
ShellCheck / shellcheck (push) Successful in 16s
Test converting README.md to README.html / test_converting_readme (push) Skipped
[1.4.0] - 2026-08-06
--------------------

- feat: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip] (#100)

- ci: bump actions/setup-python from 6 to 7 (#98)
- ci: ensure dependabot updates do not invoke ci tests [citest_skip] (#99)
2026-08-07 22:51:41 +02:00
Rich MegginsonandRichard Megginson 6007ac734f docs(changelog): version 1.4.0 [citest_skip]
Update changelog and .README.html for version 1.4.0

Signed-off-by: Rich Megginson <rmeggins@redhat.com>
2026-08-06 16:22:39 -06:00
83471be571 fix: Update sr_fingerprint task calls to use new structured parameters
The sr_fingerprint module was rewritten to accept structured parameters
(status, role_name, role_path, etc.) instead of a free-form sr_message.
Update the role tasks and tests to match the new module interface.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-08-06 16:11:09 -06:00
Sergei PetrosianandRichard Megginson d9ba105ef3 feat: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip]
Feature: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip]

Reason: By default logs are printed to rsyslog. This change adds a possibility to write logs to a file on the system for the downstream users.

Result: For the upstream, this makes rsyslog log message more detailed. For the downstream - also writes logs to /var/log/sysroles.jsonl
Signed-off-by: Sergei Petrosian <spetrosi@redhat.com>
2026-08-06 16:11:09 -06:00
Rich MegginsonandRichard Megginson c04dc49ded ci: ensure dependabot updates do not invoke ci tests [citest_skip]
dependabot updates are DoSing our ci tests - we have a limited number of github
action jobs and a dependabot update can eat up all of those for several hours,
causing problems for "real" PRs that we need testing for immediately.

Instead, tell these updates to skip ci testing, and we will manually
trigger specific roles/tests as needed when there is a dependabot update.

see https://docs.github.com/en/code-security/reference/supply-chain-security/dependabot-options-reference#prefix
for more information.

Signed-off-by: Rich Megginson <rmeggins@redhat.com>
2026-08-04 15:29:55 -06:00
dependabot[bot]andRichard Megginson 91631c556e ci: bump actions/setup-python from 6 to 7
Bumps [actions/setup-python](https://github.com/actions/setup-python) from 6 to 7.
- [Release notes](https://github.com/actions/setup-python/releases)
- [Commits](https://github.com/actions/setup-python/compare/v6...v7)

---
updated-dependencies:
- dependency-name: actions/setup-python
  dependency-version: '7'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-08-04 06:33:55 -06:00
Rich MegginsonandRichard Megginson 87545d0200 docs(changelog): version 1.3.2 [citest_skip]
Update changelog and .README.html for version 1.3.2

Signed-off-by: Rich Megginson <rmeggins@redhat.com>
2026-07-27 10:42:35 -06:00
10 changed files with 799 additions and 40 deletions
+1 -1
View File
@@ -6,4 +6,4 @@ updates:
schedule:
interval: monthly
commit-message:
prefix: ci
prefix: "ci: [citest_skip] "
+1 -1
View File
@@ -47,7 +47,7 @@ jobs:
pip3 install "git+https://github.com/linux-system-roles/tox-lsr@3.20.1"
- name: Set up Python
uses: actions/setup-python@v6
uses: actions/setup-python@v7
with:
python-version: ${{ matrix.versions.python }}
+1 -1
View File
@@ -50,7 +50,7 @@ jobs:
pip3 install "git+https://github.com/linux-system-roles/tox-lsr@3.20.1"
- name: Set up Python
uses: actions/setup-python@v6
uses: actions/setup-python@v7
with:
python-version: ${{ matrix.versions.python }}
+27
View File
@@ -1,6 +1,33 @@
Changelog
=========
[1.4.0] - 2026-08-06
--------------------
### New Features
- feat: Write roles fingerprints to /var/log/sysroles.jsonl [citest_skip] (#100)
### Other Changes
- ci: bump actions/setup-python from 6 to 7 (#98)
- ci: ensure dependabot updates do not invoke ci tests [citest_skip] (#99)
[1.3.2] - 2026-07-27
--------------------
### Bug Fixes
- fix: Better support for check mode (#93)
### Other Changes
- ci: use gha checkout v7, codecov v7 [citest_skip] (#90)
- ci: Use our own pr_title_lint.py instead of NPM commitlint [citest_skip] (#91)
- ci: bump actions/checkout from 6 to 7 (#92)
- ci: bump tox-lsr version to 3.20.0 to fix tox 4.58 api breakage [citest_skip] (#94)
- ci: Add support for Fedora 44 and drop Fedora 42 - use ansible-core 2.21 [citest_skip] (#96)
[1.3.1] - 2026-06-24
--------------------
+303 -25
View File
@@ -7,29 +7,148 @@ __metaclass__ = type
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(<log_file>.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:
sr_message: "system_role:ROLENAME"
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:
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,9 +170,184 @@ def _local_iso8601_no_microseconds():
return datetime.datetime.now(utc).astimezone().replace(microsecond=0).isoformat()
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
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:
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)
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(
sr_message=dict(type="str", required=True),
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(
@@ -61,23 +355,7 @@ def run_module():
supports_check_mode=True,
)
log_message = "%s %s" % (
module.params["sr_message"],
_local_iso8601_no_microseconds(),
)
if module.check_mode:
module.exit_json(
changed=False,
message="Check mode: message not logged - [%s]" % log_message,
)
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)
_handle_fingerprint(module)
def main():
+7 -3
View File
@@ -162,6 +162,10 @@
- name: Record role success fingerprint
sr_fingerprint:
sr_message: >-
success system_role:aide ansible_version={{ ansible_version.full }}
{{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }}
status: success
role_name: aide
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: "{{ __aide_write_log_file }}"
+7 -3
View File
@@ -7,9 +7,13 @@
- name: Record role begin fingerprint
sr_fingerprint:
sr_message: >-
begin system_role:aide ansible_version={{ ansible_version.full }}
{{ ansible_facts['distribution'] }}-{{ ansible_facts['distribution_version'] }}
status: begin
role_name: aide
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: "{{ __aide_write_log_file }}"
- name: Determine if system is ostree and set flag
when: not __aide_is_ostree is defined
+36 -6
View File
@@ -15,22 +15,52 @@
- name: Run the role
include_tasks: tasks/run_role_with_clear_facts.yml
vars:
__aide_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:aide" ||
{ 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:aide" ||
{ echo ERROR: SUCCESS fingerprint not found; exit 1; }
grep -v " Invoked with" |
grep "sr_fingerprint.*role_name=aide"
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
- name: Check if the file exists
ansible.builtin.stat:
path: /etc/aide.conf
+414
View File
@@ -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()
+2
View File
@@ -23,6 +23,8 @@ __aide_required_facts:
__aide_required_facts_subsets: "{{ ['!all', '!min'] +
__aide_required_facts }}"
__aide_write_log_file: false
# BEGIN - DO NOT EDIT THIS BLOCK - rh distros variables
# Ansible distribution identifiers that the role treats like RHEL
__aide_rh_distros: