atlas-iac/testing/tests/test_cluster_settle_check.py

"""Regression checks for interpreting quiet-window metadata without false health."""

from copy import deepcopy

from scripts.ops.cluster_settle_check import summarize


def baseline():
    """Return a minimal healthy snapshot without source or credential data."""
    return {
        "observed_at": "2026-10-04T09:00:00+00:00",
        "nodes": {"worker": {"conditions": {"Ready": "True"}}},
        "workloads": {}, "pods": {}, "volumes": {}, "warnings": {},
        "flux": {}, "helm": {},
    }


def test_requested_detached_storage_is_not_reported_healthy():
    """A stuck requested attachment remains visible even before it is attached."""
    current = baseline()
    current["volumes"]["storage/in-use"] = {
        "state": "detached", "robustness": "unknown", "requested_node": "worker",
    }
    current["volumes"]["storage/retained"] = {
        "state": "detached", "robustness": "unknown", "requested_node": "",
    }
    assert list(summarize(current)["unavailable_requested_volumes"]) == ["storage/in-use"]


def test_event_lifetime_count_is_not_a_measured_window_delta():
    """New event UIDs have unknown deltas; matching UIDs yield counter differences."""
    old = baseline()
    old["warnings"]["existing"] = {"reason": "BackOff", "count": 40}
    current = deepcopy(old)
    current["observed_at"] = "2026-10-04T09:10:00+00:00"
    current["warnings"]["existing"]["count"] = 42
    current["warnings"]["new-uid"] = {"reason": "BackOff", "count": 900}
    result = summarize(current, old)
    assert result["window_seconds"] == 600
    assert [w["increase"] for w in result["warning_changes"]] == [2, None]


def test_replaced_stateful_pod_is_distinct_from_a_container_restart():
    """The same pod name with a new UID cannot conceal controller churn."""
    old = baseline()
    old["pods"]["app/stateful-0"] = {
        "uid": "before", "node": "worker", "ready": "True", "restarts": {"app": 7},
    }
    current = deepcopy(old)
    current["pods"]["app/stateful-0"].update(uid="after", restarts={"app": 0})
    result = summarize(current, old)
    assert result["replaced_service_pods"] == ["app/stateful-0"]
    assert not result["restart_increases"]


def test_offline_pods_are_separate_from_new_reachable_node_failures():
    """Known offline-node failures stay explicit without hiding fresh failures."""
    current = baseline()
    current["nodes"]["offline"] = {"conditions": {"Ready": "Unknown"}}
    current["pods"] = {
        "app/old": {"node": "offline", "ready": "False"},
        "app/new": {"node": "worker", "ready": "False"},
    }
    result = summarize(current)
    assert result["offline_nodes"] == ["offline"]
    assert result["unready_pods_on_offline_nodes"] == 1
    assert result["unready_pods_on_reachable_nodes"] == ["app/new"]


def test_serving_old_replica_does_not_hide_failed_rollout():
    """Desired availability can coexist with a broken new release."""
    current = baseline()
    current["workloads"]["deployments/app/service"] = {
        "desired": 1, "ready": 1, "generation": 2, "observed_generation": 2,
        "rollout_failed": True,
    }
    assert list(summarize(current)["unready_workloads"]) == ["deployments/app/service"]


def test_snapshot_omits_content_and_short_lived_jobs(monkeypatch):
    """Only operational metadata leaves raw API objects; Jobs are not service churn."""
    import json
    from scripts.ops import cluster_settle_check as checker

    hidden = 'SYNTHETIC_CONTENT_MUST_NOT_APPEAR'
    pod = {
        'metadata': {'name': 'service', 'namespace': 'app', 'uid': 'u1'},
        'spec': {'nodeName': 'worker', 'containers': [{'env': [{'value': hidden}]}]},
        'status': {'phase': 'Running', 'conditions': [{'type': 'Ready', 'status': 'True'}],
                   'containerStatuses': [{'name': 'app', 'restartCount': 0, 'state': {}}]},
    }
    job_pod = deepcopy(pod)
    job_pod['metadata'].update(name='job', ownerReferences=[{'kind': 'Job', 'name': 'timer'}])
    done_pod = deepcopy(pod)
    done_pod['metadata']['name'] = 'finished'
    done_pod['status']['phase'] = 'Succeeded'
    data = {name: [] for name in checker.QUERIES}
    data['pods'] = [pod, job_pod, done_pod]
    data['warnings'] = [{
        'metadata': {'uid': 'event'}, 'involvedObject': {'namespace': 'app', 'name': 'service'},
        'reason': 'Unhealthy', 'count': 1, 'message': hidden,
    }]
    monkeypatch.setattr(checker, 'fetch', lambda item: (item[0], data[item[0]]))
    current = checker.snapshot()
    assert list(current['pods']) == ['app/service']
    assert hidden not in json.dumps(current)
    assert current['warnings']['event']['reason'] == 'Unhealthy'


def test_failed_api_read_cannot_become_a_partial_health_report(monkeypatch):
    """A failed required query raises without echoing an API response body."""
    from types import SimpleNamespace
    import pytest
    from scripts.ops import cluster_settle_check as checker

    hidden = 'SYNTHETIC_PROVIDER_RESPONSE_CONTENT'
    seen = []

    def fail_read(command, **kwargs):
        """Record the safe read command and simulate a failing API process."""
        seen.append(command)
        return SimpleNamespace(returncode=1, stdout=hidden, stderr=hidden)

    monkeypatch.setattr(checker.subprocess, 'run', fail_read)
    with pytest.raises(RuntimeError) as failure:
        checker.fetch(('nodes', ['nodes']))
    assert 'Cannot read nodes' in str(failure.value)
    assert hidden not in str(failure.value)
    assert seen[0][:4] == ['kubectl', '--request-timeout=20s', 'get', 'nodes']


def test_restart_increase_requires_same_pod_identity():
    """Resets cannot cancel another container's observed restarts."""
    old = baseline()
    old['pods']['app/service'] = {
        'uid': 'same', 'node': 'worker', 'ready': 'True',
        'restarts': {'app': 7, 'helper': 2},
    }
    current = deepcopy(old)
    current['pods']['app/service']['restarts'] = {'app': 0, 'helper': 5}
    assert summarize(current, old)['restart_increases'] == {'app/service': 3}


def test_successful_api_read_projects_only_items(monkeypatch):
    """The list envelope is not part of the data passed to health analysis."""
    from types import SimpleNamespace
    from scripts.ops import cluster_settle_check as checker

    monkeypatch.setattr(checker.subprocess, 'run', lambda *a, **k: SimpleNamespace(
        returncode=0, stdout='{"items": [], "metadata": {"resourceVersion": "1"}}',
    ))
    assert checker.fetch(('nodes', ['nodes'])) == ('nodes', [])


def test_cli_writes_snapshot_and_reports_comparison(tmp_path, monkeypatch, capsys):
    """The operator's two-file workflow preserves its baseline and interval."""
    import json
    from scripts.ops import cluster_settle_check as checker

    before = tmp_path / 'before.json'
    before.write_text(json.dumps(baseline()))
    current = baseline()
    current['observed_at'] = '2026-10-04T09:15:00+00:00'
    output = tmp_path / 'new' / 'after.json'
    monkeypatch.setattr(checker, 'snapshot', lambda: current)
    monkeypatch.setattr('sys.argv', ['check', '--output', str(output), '--previous', str(before)])
    checker.main()
    assert json.loads(capsys.readouterr().out)['window_seconds'] == 900
    assert json.loads(output.read_text()) == current
    assert json.loads(before.read_text()) == baseline()