From bfc039c6520ebbc7e908c7354810b4ed04be28fa Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Mon, 3 Aug 2026 10:50:45 +0200 Subject: [PATCH] Add integration tests for the Ansible facts freshness check check_ansible_facts() scans Redis for ansible_facts* keys and reads ansible_date_time.epoch to decide whether cached facts are stale. Its unit tests drive it through a MagicMock client, so the cursor-based SCAN loop, the bytes decoding of keys and values, and JSON parsing of data returned by a real server are only simulated. The new module covers those paths against the live Redis the python-osism-integration-tests job provides: an empty cache, fresh facts, a stale host alongside a fresh one, facts without an epoch field, a malformed JSON value, and an empty value. Seeded keys use itest- hostnames and are removed after each test. Every assertion is scoped to a seeded hostname or to a message unique to its scenario, so unrelated keys in a shared Redis cannot fail the suite. The empty-cache test skips instead of deleting when the database already holds ansible_facts keys. Assisted-by: Claude:claude-opus-5 Signed-off-by: Christian Berendt --- tests/integration/test_facts.py | 156 ++++++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 tests/integration/test_facts.py diff --git a/tests/integration/test_facts.py b/tests/integration/test_facts.py new file mode 100644 index 000000000..d71e59ab6 --- /dev/null +++ b/tests/integration/test_facts.py @@ -0,0 +1,156 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Integration tests for ``osism.utils.check_ansible_facts()``. + +The function scans Redis for ``ansible_facts*`` keys with a cursor-based +``SCAN`` loop and reads ``ansible_date_time.epoch`` from each JSON value to +decide whether the cached facts are stale, reporting the outcome through +loguru. The unit tests cover the same scenarios against a ``MagicMock`` +client; running them against a live Redis exercises what the mocks stand in +for: the cursor loop, keys and values arriving as ``bytes``, and JSON that +round-tripped through a Redis server. The suite is skipped automatically when +Redis is not reachable (see ``conftest.py``). +""" + +import json +import time +import uuid + +import pytest + +from osism import utils + +pytestmark = pytest.mark.integration + +# Passed explicitly to every call: the ``settings.FACTS_MAX_AGE`` default is 12 +# hours, which would tie the stale case to the environment. +MAX_AGE = 300 + + +@pytest.fixture +def seed_facts(): + """Seed ``ansible_facts`` keys and remove them after the test.""" + keys = [] + + def _seed(host, value): + key = f"ansible_facts{host}" + utils.redis.set(key, value) + keys.append(key) + + yield _seed + + for key in keys: + utils.redis.delete(key) + + +def _host(): + """Return a hostname unique to one test. + + The ``itest-`` prefix keeps it clear of ``LOCAL_FACT_HOSTS``, whose facts + ``check_ansible_facts()`` skips outright. + """ + return f"itest-{uuid.uuid4()}" + + +def _facts(epoch): + """Return a facts blob carrying ``epoch``. + + Ansible stores the epoch as a string, and ``check_ansible_facts()`` casts + it with ``float()``, so the string form is what the test seeds. + """ + return json.dumps({"ansible_date_time": {"epoch": str(int(epoch))}}) + + +def _messages(records, level, needle): + """Return the captured messages at ``level`` that contain ``needle``.""" + return [ + record["message"] + for record in records + if record["level"] == level and needle in record["message"] + ] + + +def test_no_facts_warns_about_empty_cache(loguru_logs): + """An empty facts cache is reported as such. + + ``check_ansible_facts()`` scans the whole database, so a Redis holding + real facts cannot produce this case. Skip there rather than deleting data + that does not belong to the suite; the CI container starts empty. + """ + leftover = sorted( + key.decode() for key in utils.redis.scan_iter(match="ansible_facts*") + ) + if leftover: + pytest.skip(f"Redis already holds ansible_facts keys: {leftover}") + + utils.check_ansible_facts(max_age=MAX_AGE) + + assert _messages(loguru_logs, "WARNING", "No Ansible facts found in Redis cache") + + +def test_fresh_facts_produce_no_warning(seed_facts, loguru_logs): + """A host whose facts are current is not reported.""" + host = _host() + seed_facts(host, _facts(time.time())) + + utils.check_ansible_facts(max_age=MAX_AGE) + + assert not _messages(loguru_logs, "WARNING", host) + + +def test_stale_facts_report_the_stale_host_only(seed_facts, loguru_logs): + """Facts older than ``max_age`` are reported, fresh ones are not.""" + stale_host = _host() + fresh_host = _host() + seed_facts(stale_host, _facts(time.time() - 9999)) + seed_facts(fresh_host, _facts(time.time())) + + utils.check_ansible_facts(max_age=MAX_AGE) + + stale_messages = _messages( + loguru_logs, "WARNING", f"Host '{stale_host}': facts are" + ) + assert stale_messages + assert all("seconds old" in message for message in stale_messages) + assert _messages(loguru_logs, "WARNING", "Run 'osism sync facts' to update facts.") + assert not _messages(loguru_logs, "WARNING", fresh_host) + + +def test_facts_without_epoch_are_skipped(seed_facts, loguru_logs): + """Facts lacking ``ansible_date_time.epoch`` are skipped, not reported.""" + host = _host() + seed_facts(host, json.dumps({"ansible_hostname": "node-1"})) + + utils.check_ansible_facts(max_age=MAX_AGE) + + assert _messages( + loguru_logs, "DEBUG", f"Host '{host}': facts missing ansible_date_time.epoch" + ) + assert not _messages(loguru_logs, "WARNING", host) + + +def test_malformed_json_is_skipped(seed_facts, loguru_logs): + """A value that is not JSON is skipped without failing the check. + + The debug call uses printf-style placeholders that loguru's ``str.format`` + leaves in place, so the recorded message is the literal template. Match on + its prefix rather than on interpolated values. + """ + host = _host() + seed_facts(host, "{ not valid json") + + utils.check_ansible_facts(max_age=MAX_AGE) + + assert _messages(loguru_logs, "DEBUG", "Skipping malformed ansible_facts entry") + assert not _messages(loguru_logs, "WARNING", host) + + +def test_empty_value_is_skipped(seed_facts, loguru_logs): + """An empty value is skipped silently.""" + host = _host() + seed_facts(host, "") + + utils.check_ansible_facts(max_age=MAX_AGE) + + assert not _messages(loguru_logs, "WARNING", host) + assert not _messages(loguru_logs, "DEBUG", host)