"""Tests for the inode-cache watchdog.

The property that matters is restraint: this thing runs as root on a timer and drops the
whole machine's dentry cache, so it must do nothing at all until the count genuinely
approaches the ceiling, and it must never talk itself into a higher ceiling than the one
it has evidence for.  Most of what follows is about the *absence* of action.
"""

import errno
import json
import os
import threading

import pytest

import reclaim


@pytest.fixture(autouse=True)
def fresh_probe_lock(monkeypatch):
    """A probe the watchdog abandoned holds the lock until it returns, by design.  Tests
    must not inherit that from each other."""
    monkeypatch.setattr(reclaim, "_probe_lock", threading.Lock())


SLABINFO = """\
slabinfo - version: 2.1
# name            <active_objs> <num_objs> <objsize> <objperslab> <pagesperslab>
ext4_inode_cache   40312  41800    1096   29    8 : tunables    0    0    0
fuse_inode        812345 900000     832   19    4 : tunables    0    0    0
dentry             38423  45150     208   19    1 : tunables    0    0    0
"""


def write(tmp_path, name, text):
    path = tmp_path / name
    path.write_text(text)
    return str(path)


# --------------------------------------------------------------------------- the metric


def test_fuse_inodes_reads_the_slab_line(tmp_path):
    assert reclaim.fuse_inodes(write(tmp_path, "slabinfo", SLABINFO)) == 812345


def test_fuse_inodes_is_none_when_the_line_is_absent(tmp_path):
    """No fuse_inode line means no FUSE mounts, which is not the same as zero cached --
    it means this metric does not apply and the caller must fall back."""
    trimmed = "\n".join(l for l in SLABINFO.splitlines() if "fuse_inode" not in l)
    assert reclaim.fuse_inodes(write(tmp_path, "slabinfo", trimmed)) is None


def test_fuse_inodes_is_none_when_unreadable(tmp_path):
    """The usual case: /proc/slabinfo is root-only and we are not root."""
    assert reclaim.fuse_inodes(str(tmp_path / "nope")) is None


def test_fuse_inodes_survives_a_garbled_line(tmp_path):
    assert reclaim.fuse_inodes(write(tmp_path, "s", "fuse_inode notanumber\n")) is None


def test_total_inodes_reads_the_first_field(tmp_path):
    assert reclaim.total_inodes(write(tmp_path, "inode-nr", "64282\t1\n")) == 64282


def test_total_inodes_is_zero_when_malformed(tmp_path):
    assert reclaim.total_inodes(write(tmp_path, "inode-nr", "")) == 0
    assert reclaim.total_inodes(write(tmp_path, "i2", "x y\n")) == 0
    assert reclaim.total_inodes(str(tmp_path / "missing")) == 0


def test_total_inodes_reads_the_real_procfs():
    assert reclaim.total_inodes() > 0


def test_cached_inodes_prefers_the_exact_source(monkeypatch):
    monkeypatch.setattr(reclaim, "fuse_inodes", lambda *a: 500)
    monkeypatch.setattr(reclaim, "total_inodes", lambda *a: 999)
    assert reclaim.cached_inodes() == (500, "fuse_inode")


def test_cached_inodes_falls_back_when_not_root(monkeypatch):
    monkeypatch.setattr(reclaim, "fuse_inodes", lambda *a: None)
    monkeypatch.setattr(reclaim, "total_inodes", lambda *a: 999)
    assert reclaim.cached_inodes() == (999, "inode-nr")


# ---------------------------------------------------------------------------- the probe


def test_probe_succeeds_on_a_readable_file(tmp_path):
    assert reclaim.probe(write(tmp_path, "f", "x")) == 0


def test_probe_reports_the_errno_that_stopped_it(tmp_path):
    assert reclaim.probe(str(tmp_path / "absent")) == errno.ENOENT


def test_probe_treats_a_mount_that_does_not_answer_as_exhaustion(tmp_path):
    """A FIFO with no writer blocks in open() exactly as a wedged mount does.  The
    watchdog must not be the thing that hangs."""
    fifo = str(tmp_path / "fifo")
    os.mkfifo(fifo)
    try:
        assert reclaim.probe(fifo, timeout=0.2) == errno.ENFILE
    finally:
        # Let the abandoned reader finish, so it does not outlive the test.
        os.close(os.open(fifo, os.O_WRONLY | os.O_NONBLOCK))


def test_probe_does_not_pile_up_threads_on_a_wedged_mount(tmp_path):
    """One abandoned probe is the budget.  A probe still outstanding from last time is
    itself the answer, so the next poll must not start another thread."""
    assert reclaim._probe_lock.acquire(blocking=False), "lock leaked from another test"
    try:
        assert reclaim.probe(str(tmp_path)) == errno.ENFILE
    finally:
        reclaim._probe_lock.release()


# --------------------------------------------------------------------------- the config


def test_config_defaults_when_the_file_is_absent(tmp_path):
    assert reclaim.load_config(str(tmp_path / "none.conf")) == {}


def test_config_parses_values_comments_and_blanks(tmp_path):
    path = write(tmp_path, "c.conf", "# a comment\n\nceiling = 400000\n"
                                     "interval=2.5  # trailing\nprobe_path=/tmp/x\n")
    assert reclaim.load_config(path) == {
        "ceiling": 400000, "interval": 2.5, "probe_path": "/tmp/x"}


def test_config_rejects_an_unknown_setting(tmp_path):
    """Silently ignoring a misspelt key would leave the watchdog on defaults while the
    operator believed otherwise, which is the worst of both."""
    path = write(tmp_path, "c.conf", "cieling=400000\n")
    with pytest.raises(SystemExit, match="unknown setting"):
        reclaim.load_config(path)


def test_config_rejects_a_value_that_would_disarm_the_watchdog(tmp_path):
    """An interval of zero spins; a ceiling of zero drops caches forever."""
    with pytest.raises(SystemExit, match="interval"):
        reclaim.load_config(write(tmp_path, "a.conf", "interval=0\n"))
    with pytest.raises(SystemExit, match="ceiling"):
        reclaim.load_config(write(tmp_path, "b.conf", "ceiling=0\n"))


def test_config_rejects_a_line_without_a_value(tmp_path):
    with pytest.raises(SystemExit, match="expected key=value"):
        reclaim.load_config(write(tmp_path, "c.conf", "ceiling\n"))


# ---------------------------------------------------------------------------- the state


def test_corrupt_state_is_tolerated(tmp_path):
    assert reclaim.load_state(write(tmp_path, "s.json", "{not json")) == {}
    assert reclaim.load_state(write(tmp_path, "s2.json", "[1, 2]")) == {}
    assert reclaim.load_state(str(tmp_path / "absent.json")) == {}


def test_state_round_trips(tmp_path):
    path = str(tmp_path / "sub" / "state.json")
    assert reclaim.save_state({"ceiling": 123}, path) is True
    assert reclaim.load_state(path)["ceiling"] == 123


# ------------------------------------------------------------------------- the watchdog


@pytest.fixture
def watchdog(tmp_path, monkeypatch):
    """A watchdog with a known ceiling, whose lever and probe are recorded not fired."""
    made = reclaim.Watchdog({"ceiling": 100_000, "cooldown": 60.0},
                            state_path=str(tmp_path / "state.json"))
    made.dropped = []
    monkeypatch.setattr(reclaim, "drop_caches",
                        lambda: made.dropped.append(1) or True)
    monkeypatch.setattr(reclaim, "probe", lambda *a, **k: 0)
    return made


def at(monkeypatch, count):
    monkeypatch.setattr(reclaim, "cached_inodes", lambda: (count, "test"))


def test_high_water_defaults_to_a_fraction_of_the_ceiling(watchdog):
    assert watchdog.high == 75_000


def test_an_explicit_high_water_overrides_the_fraction(tmp_path):
    made = reclaim.Watchdog({"ceiling": 100_000, "high": 90_000},
                            state_path=str(tmp_path / "s.json"))
    assert made.high == 90_000


def test_it_does_nothing_while_the_cache_is_comfortable(watchdog, monkeypatch):
    """The whole point.  Reclaiming early throws away cache that costs SMB round trips
    to rebuild, so below the mark the watchdog must be inert."""
    at(monkeypatch, 40_000)
    assert watchdog.step() is False
    assert watchdog.dropped == []


def test_it_does_not_even_probe_while_comfortable(watchdog, monkeypatch):
    probes = []
    monkeypatch.setattr(reclaim, "probe", lambda *a, **k: probes.append(1) or 0)
    at(monkeypatch, 40_000)
    watchdog.step()
    assert probes == []


def test_it_reclaims_at_the_high_water_mark(watchdog, monkeypatch):
    at(monkeypatch, 75_000)
    assert watchdog.step() is True
    assert len(watchdog.dropped) == 1


def test_it_respects_the_cooldown(watchdog, monkeypatch):
    """A job that refills the cache immediately must not turn this into a thrash loop."""
    at(monkeypatch, 80_000)
    assert watchdog.step() is True
    assert watchdog.step() is False
    assert len(watchdog.dropped) == 1


def test_the_cooldown_expires(watchdog, monkeypatch):
    at(monkeypatch, 80_000)
    watchdog.step()
    watchdog.last_drop -= watchdog.cooldown + 1
    assert watchdog.step() is True
    assert len(watchdog.dropped) == 2


def test_it_tracks_the_peak_for_the_heartbeat(watchdog, monkeypatch):
    """The peak is what tells the operator whether the mark can be raised."""
    at(monkeypatch, 40_000)
    watchdog.step()
    at(monkeypatch, 61_000)
    watchdog.step()
    at(monkeypatch, 50_000)
    watchdog.step()
    assert watchdog.peak == 61_000


def test_it_reports_failure_when_it_cannot_drop(watchdog, monkeypatch):
    monkeypatch.setattr(reclaim, "drop_caches", lambda: False)
    at(monkeypatch, 80_000)
    assert watchdog.step() is False


def test_it_backs_off_when_it_cannot_drop(watchdog, monkeypatch):
    """Lack of privilege does not resolve on its own, so a misconfigured service must
    warn once a cooldown rather than once a poll."""
    attempts = []
    monkeypatch.setattr(reclaim, "drop_caches", lambda: attempts.append(1) or False)
    at(monkeypatch, 80_000)
    watchdog.step()
    watchdog.step()
    watchdog.step()
    assert len(attempts) == 1


# ------------------------------------------------------------- learning the real ceiling


def test_enfile_near_the_mark_lowers_the_ceiling_and_reclaims(watchdog, monkeypatch):
    """The configured ceiling is a guess.  Being refused an open is proof it was wrong,
    and the only proof available from inside the guest."""
    monkeypatch.setattr(reclaim, "probe", lambda *a, **k: errno.ENFILE)
    at(monkeypatch, 70_000)  # above 0.9*high, below high: would not have acted
    assert watchdog.step() is True
    assert watchdog.ceiling == 70_000
    assert watchdog.high == 52_500
    assert len(watchdog.dropped) == 1


def test_emfile_counts_as_exhaustion_too(watchdog, monkeypatch):
    monkeypatch.setattr(reclaim, "probe", lambda *a, **k: errno.EMFILE)
    at(monkeypatch, 70_000)
    assert watchdog.step() is True
    assert watchdog.ceiling == 70_000


def test_an_unrelated_probe_failure_is_not_exhaustion(watchdog, monkeypatch):
    """A missing probe file must not be mistaken for a wedged mount."""
    monkeypatch.setattr(reclaim, "probe", lambda *a, **k: errno.ENOENT)
    at(monkeypatch, 70_000)
    assert watchdog.step() is False
    assert watchdog.ceiling == 100_000


def test_the_ceiling_is_never_raised(watchdog):
    """A watchdog that could raise its own limit would creep towards the wall looking
    for it, and finding it is the outcome this exists to prevent."""
    assert watchdog.observe_exhaustion(150_000) is False
    assert watchdog.ceiling == 100_000
    assert watchdog.observe_exhaustion(100_000) is False
    assert watchdog.ceiling == 100_000
    assert watchdog.observe_exhaustion(0) is False
    assert watchdog.ceiling == 100_000


def test_a_learned_ceiling_survives_a_restart(tmp_path):
    path = str(tmp_path / "state.json")
    first = reclaim.Watchdog({"ceiling": 100_000}, state_path=path)
    first.observe_exhaustion(60_000)
    assert json.load(open(path))["ceiling"] == 60_000

    second = reclaim.Watchdog({"ceiling": 100_000}, state_path=path)
    assert second.ceiling == 60_000
    assert second.high == 45_000


def test_a_learned_ceiling_does_not_override_a_lower_configured_one(tmp_path):
    path = str(tmp_path / "state.json")
    reclaim.save_state({"ceiling": 90_000}, path)
    made = reclaim.Watchdog({"ceiling": 50_000}, state_path=path)
    assert made.ceiling == 50_000


def test_an_explicit_high_water_does_not_follow_the_ceiling_down(tmp_path):
    """A derived mark should track a corrected ceiling; one the operator set by hand is
    their decision to revisit."""
    path = str(tmp_path / "state.json")
    made = reclaim.Watchdog({"ceiling": 100_000, "high": 80_000}, state_path=path)
    made.observe_exhaustion(90_000)
    assert made.high == 80_000


# ---------------------------------------------------------------------------- the CLI


def test_check_is_quiet_and_succeeds_below_the_mark(monkeypatch, tmp_path):
    at(monkeypatch, 1000)
    monkeypatch.setattr(reclaim, "drop_caches", lambda: pytest.fail("reclaimed"))
    assert reclaim.main(["--check", "--config", str(tmp_path / "none.conf")]) == 0


def test_check_fails_above_the_mark(monkeypatch, tmp_path):
    at(monkeypatch, 10_000_000)
    monkeypatch.setattr(reclaim, "drop_caches", lambda: pytest.fail("reclaimed"))
    assert reclaim.main(["--check", "--config", str(tmp_path / "none.conf")]) == 1


def test_a_plain_run_below_the_mark_changes_nothing(monkeypatch, tmp_path):
    at(monkeypatch, 1000)
    monkeypatch.setattr(reclaim, "drop_caches", lambda: pytest.fail("reclaimed"))
    assert reclaim.main(["--config", str(tmp_path / "none.conf")]) == 0


def test_force_reclaims_below_the_mark(monkeypatch, tmp_path):
    calls = []
    at(monkeypatch, 1000)
    monkeypatch.setattr(reclaim, "drop_caches", lambda: calls.append(1) or True)
    monkeypatch.setattr(reclaim.os, "geteuid", lambda: 0)
    assert reclaim.main(["--force", "--config", str(tmp_path / "none.conf")]) == 0
    assert calls == [1]


def test_force_without_root_fails_loudly_rather_than_pretending(monkeypatch, tmp_path):
    """The old fallback allocated tens of gigabytes to fake memory pressure.  Saying so
    and exiting nonzero is better than quietly costing the machine its memory."""
    at(monkeypatch, 1000)
    monkeypatch.setattr(reclaim, "drop_caches", lambda: pytest.fail("reclaimed"))
    monkeypatch.setattr(reclaim.os, "geteuid", lambda: 1000)
    assert reclaim.main(["--force", "--config", str(tmp_path / "none.conf")]) == 1


# ------------------------------------------------------------------------- the installer


def test_install_writes_the_script_the_unit_and_a_config(tmp_path):
    target = str(tmp_path / "sbin" / "suvi-reclaimd")
    service = str(tmp_path / "suvi-reclaimd.service")
    config = str(tmp_path / "suvi-reclaimd.conf")
    reclaim.install(target, service, config)

    assert os.access(target, os.X_OK)
    assert open(target).read() == open(reclaim.__file__).read()
    unit = open(service).read()
    assert "--daemon" in unit and target in unit
    # The interpreter must be the system one, not the venv: the venv lives on the very
    # mount this recovers, and cannot be executed when that mount is refusing opens.
    assert "/usr/bin/python3 -I" in unit
    # The config it writes has to be one it can read back.
    assert reclaim.load_config(config) == {"ceiling": reclaim.DEFAULT_CEILING}


def test_install_never_clobbers_a_tuned_config(tmp_path):
    """The operator raises `high` from what the heartbeat reports.  Reinstalling after
    an update must not silently put that back to the default."""
    config = str(tmp_path / "suvi-reclaimd.conf")
    with open(config, "w") as handle:
        handle.write("high = 912345\n")
    reclaim.install(str(tmp_path / "sbin" / "d"), str(tmp_path / "u"), config)
    assert reclaim.load_config(config) == {"high": 912345}


def test_install_refuses_to_overwrite_the_original(tmp_path):
    with pytest.raises(SystemExit, match="refusing"):
        reclaim.install(reclaim.__file__, str(tmp_path / "unit"))