atlas-iac/testing/tests/test_node_pod_log_cleanup.py

"""Exercise orphan cleanup against temporary kubelet and log inventories."""

import importlib.util
import json
import os
from pathlib import Path
import time

import pytest

SOURCE = Path(__file__).resolve().parents[2] / 'services/maintenance/node-ops/scripts/node_pod_log_cleanup.py'
spec = importlib.util.spec_from_file_location('node_pod_log_cleanup', SOURCE)
cleaner = importlib.util.module_from_spec(spec)
spec.loader.exec_module(cleaner)
ACTIVE = '11111111-1111-1111-1111-111111111111'
ORPHAN = '22222222-2222-2222-2222-222222222222'


def inventory(root):
    """Create one current pod with a UID-only kubelet directory."""
    (root / 'var/lib/kubelet/pods' / ACTIVE).mkdir(parents=True)


def logs(root, uid, relative='var/log/pods', old=True):
    """Create the real namespace_name_UID directory layout with a synthetic log."""
    path = root / relative / ('namespace_name_' + uid)
    (path / 'container').mkdir(parents=True)
    (path / 'container/0.log').write_text('SYNTHETIC LOG CONTENT')
    if old:
        stamp = time.time() - 10 * 86400
        for entry in [*path.rglob('*'), path]:
            os.utime(entry, (stamp, stamp))
    return path


def test_active_uid_preserves_old_current_and_hdd_logs(tmp_path):
    """Old quiet containers remain protected in both log locations."""
    inventory(tmp_path)
    paths = [logs(tmp_path, ACTIVE, relative=r) for r in ('var/log/pods', 'var/log.hdd/pods')]
    assert cleaner.cleanup(tmp_path, 3)['removed'] == 0
    assert all((p / 'container/0.log').exists() for p in paths)


def test_only_expired_orphans_are_removed(tmp_path):
    """Dry-run reports the same eligible directory without deleting it."""
    inventory(tmp_path)
    path = logs(tmp_path, ORPHAN)
    assert cleaner.cleanup(tmp_path, 3, dry_run=True)['eligible'] == 1
    assert path.exists()
    assert cleaner.cleanup(tmp_path, 3)['removed'] == 1
    assert not path.exists()


def test_recent_log_preserves_old_parent_directory(tmp_path):
    """Writing a log does not update the pod directory's own mtime."""
    inventory(tmp_path)
    path = logs(tmp_path, ORPHAN)
    os.utime(path / 'container/0.log', None)
    assert cleaner.cleanup(tmp_path, 3)['skipped'] == 1
    assert path.exists()


@pytest.mark.parametrize('empty', [False, True])
def test_unavailable_inventory_cannot_authorize_deletion(tmp_path, empty):
    """Missing and empty inventories both preserve all candidate logs."""
    if empty:
        (tmp_path / 'var/lib/kubelet/pods').mkdir(parents=True)
    path = logs(tmp_path, ORPHAN)
    assert cleaner.cleanup(tmp_path, 3)['inventory_unavailable'] == 1
    assert path.exists()


def test_symlinks_unknown_names_and_nondirectories_are_preserved(tmp_path):
    """Only recognized pod directories enter deletion; links never escape root."""
    inventory(tmp_path)
    parent = tmp_path / 'var/log/pods'
    parent.mkdir(parents=True)
    outside = tmp_path / 'outside'
    outside.mkdir()
    (parent / ('ns_link_' + ORPHAN)).symlink_to(outside)
    (parent / 'unrecognized').mkdir()
    (parent / ('ns_file_' + ORPHAN)).write_text('metadata')
    assert cleaner.cleanup(tmp_path, 3)['removed'] == 0
    assert outside.exists()
    assert (parent / 'unrecognized').exists()


def test_scan_failure_preserves_candidate(tmp_path, monkeypatch):
    """An unreadable log subtree must not look like an old empty directory."""
    inventory(tmp_path)
    path = logs(tmp_path, ORPHAN)

    def fail_scan(*args, **kwargs):
        """Simulate the walker reporting a storage error."""
        kwargs['onerror'](OSError('synthetic scan failure'))

    monkeypatch.setattr(cleaner.os, 'walk', fail_scan)
    assert cleaner.cleanup(tmp_path, 3)['skipped'] == 1
    assert path.exists()


@pytest.mark.parametrize('new_active', [True, False])
def test_inventory_change_before_deletion_is_rechecked(tmp_path, monkeypatch, new_active):
    """A newly active UID or vanished inventory invalidates the initial decision."""
    inventory(tmp_path)
    path = logs(tmp_path, ORPHAN)
    original = cleaner.newest_mtime

    def change_inventory(candidate):
        """Alter only inventory metadata after the age scan."""
        newest = original(candidate)
        pods = tmp_path / 'var/lib/kubelet/pods'
        if new_active:
            (pods / ORPHAN).mkdir()
        else:
            (pods / ACTIVE).rmdir()
        return newest

    monkeypatch.setattr(cleaner, 'newest_mtime', change_inventory)
    assert cleaner.cleanup(tmp_path, 3)['removed'] == 0
    assert path.exists()


def test_cli_reports_counts_and_rejects_zero_retention(tmp_path, monkeypatch, capsys):
    """No content or candidate path enters the operator's cleanup summary."""
    inventory(tmp_path)
    path = logs(tmp_path, ORPHAN)
    monkeypatch.setattr('sys.argv', ['clean', '--host-root', str(tmp_path), '--dry-run'])
    cleaner.main()
    result = json.loads(capsys.readouterr().out)
    assert result['eligible'] == 1 and result['removed'] == 0
    assert path.exists()
    with pytest.raises(ValueError, match='at least one day'):
        cleaner.cleanup(tmp_path, 0)