self-directed-learning/test/test_ump_decay.py

#!/usr/bin/env python3
"""
test_ump_decay.py — Tests for ump_decay.py. Stdlib only. Prints PASS/FAIL.
Exit 1 if any FAIL.

Each test uses a temporary fixture file under tmp/ (auto-cleaned).
"""

from __future__ import annotations

import json
import shutil
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

SCRIPT = Path("/root/ump-recall/scripts/ump_decay.py")
sys.path.insert(0, str(SCRIPT.parent))
import ump_decay  # type: ignore  # noqa: E402


# ---- Mini test harness ----------------------------------------------------

_results: list[tuple[str, bool, str]] = []


def record(name: str, ok: bool, detail: str = "") -> None:
    _results.append((name, ok, detail))
    flag = "PASS" if ok else "FAIL"
    line = f"  [{flag}] {name}"
    if detail and not ok:
        line += f"\n         {detail}"
    print(line)


def assert_eq(name: str, got, want, tol: float = 0.0) -> None:
    if isinstance(got, float) and isinstance(want, (int, float)):
        ok = abs(got - want) <= tol
    else:
        ok = got == want
    detail = "" if ok else f"got={got!r} want={want!r}"
    record(name, ok, detail)


def assert_true(name: str, cond: bool, detail: str = "") -> None:
    record(name, bool(cond), detail)


# ---- Fixture builders -----------------------------------------------------

def make_rec(
    *,
    urn: str,
    kind: str,
    confidence: float | None,
    status: str | None,
    created: datetime,
    modified: datetime | None = None,
    include_lifecycle: bool = True,
) -> dict:
    rec: dict = {
        "id": urn,
        "kind": kind,
        "time": {"created": created.isoformat().replace("+00:00", "Z")},
    }
    if modified is not None:
        rec["time"]["modified"] = modified.isoformat().replace("+00:00", "Z")
    if include_lifecycle:
        rec["lifecycle"] = {}
        if confidence is not None:
            rec["lifecycle"]["confidence"] = confidence
        if status is not None:
            rec["lifecycle"]["status"] = status
    return rec


# ---- Tests ----------------------------------------------------------------

def test_1_semantic_one_month_factor() -> None:
    """1-month-old semantic record with conf=0.8 decays by factor ~0.97."""
    # Use a frozen 'now' by patching datetime via apply_decay — but apply_decay
    # uses datetime.now() internally, so we test the math via direct call:
    # we'll set the created date so that "now" yields ~30 days.
    now = datetime.now(timezone.utc)
    created = now - timedelta(days=30)
    rec = make_rec(
        urn="urn:t1",
        kind="semantic",
        confidence=0.8,
        status="active",
        created=created,
    )
    out, _report = ump_decay.apply_decay([rec], dry_run=False)
    # λ=0.001, 30 days → factor = exp(-0.03) ≈ 0.9704
    factor = out[0]["lifecycle"]["confidence"] / 0.8
    assert_true(
        "Test 1: 30-day semantic factor in (0.96, 0.98)",
        0.96 < factor < 0.98,
        detail=f"factor={factor:.4f}, conf={out[0]['lifecycle']['confidence']:.4f}",
    )


def test_2_episodic_30_days_noticeable() -> None:
    """30-day-old episodic record decays noticeably (~0.74)."""
    now = datetime.now(timezone.utc)
    created = now - timedelta(days=30)
    rec = make_rec(
        urn="urn:t2",
        kind="episodic",
        confidence=1.0,
        status="active",
        created=created,
    )
    out, _ = ump_decay.apply_decay([rec], dry_run=False)
    # λ=0.01, 30 days → exp(-0.3) ≈ 0.7408
    conf = out[0]["lifecycle"]["confidence"]
    assert_eq("Test 2: 30-day episodic confidence ≈ 0.74",
              conf, 0.7408, tol=0.01)


def test_3_identity_no_decay() -> None:
    """Identity record never decays meaningfully (1.0 → ~0.997 over 30 days)."""
    now = datetime.now(timezone.utc)
    created_30 = now - timedelta(days=30)
    rec = make_rec(
        urn="urn:t3",
        kind="identity",
        confidence=1.0,
        status="active",
        created=created_30,
    )
    out, _ = ump_decay.apply_decay([rec], dry_run=False)
    conf = out[0]["lifecycle"]["confidence"]
    assert_eq("Test 3: 30-day identity ≈ 0.997", conf, 0.997, tol=0.001)


def test_4_archive_threshold() -> None:
    """A record with confidence < 0.2 after decay → status flips to 'archived'."""
    now = datetime.now(timezone.utc)
    # Working memory λ=0.05.  conf=1.0, 60 days: exp(-3) ≈ 0.0498 → floor 0.05.
    # That's < 0.2 → archived.
    created = now - timedelta(days=60)
    rec = make_rec(
        urn="urn:t4",
        kind="working",
        confidence=1.0,
        status="active",
        created=created,
    )
    out, report = ump_decay.apply_decay([rec], dry_run=False)
    assert_eq("Test 4: low-confidence status → 'archived'",
              out[0]["lifecycle"]["status"], "archived")
    assert_eq("Test 4: report['after']['archived'] == 1",
              report["after"]["archived"], 1)
    assert_true("Test 4: archive_candidates populated",
                len(report["archive_candidates"]) == 1)


def test_5_candidate_to_active() -> None:
    """Candidate record with confidence crossing 0.5 → status flips to 'active'."""
    now = datetime.now(timezone.utc)
    # Semantic, λ=0.001, freshly created (0 days) → conf unchanged at 0.8.
    # 0.8 >= 0.5 → promote.
    created = now - timedelta(days=0)
    rec = make_rec(
        urn="urn:t5",
        kind="semantic",
        confidence=0.8,
        status="candidate",
        created=created,
    )
    out, report = ump_decay.apply_decay([rec], dry_run=False)
    assert_eq("Test 5: candidate with conf>=0.5 → 'active'",
              out[0]["lifecycle"]["status"], "active")
    assert_eq("Test 5: promotion_candidates populated",
              len(report["promotion_candidates"]), 1)


def test_6_floor_at_0_05() -> None:
    """A record that would decay to 0.001 stays at 0.05 (floor)."""
    now = datetime.now(timezone.utc)
    # Working memory 200 days old: exp(-10) ≈ 4.5e-5 → floor at 0.05.
    created = now - timedelta(days=200)
    rec = make_rec(
        urn="urn:t6",
        kind="working",
        confidence=1.0,
        status="active",
        created=created,
    )
    out, _ = ump_decay.apply_decay([rec], dry_run=False)
    assert_eq("Test 6: floor clamps new_confidence to 0.05",
              out[0]["lifecycle"]["confidence"], 0.05)


def test_7_atomic_write(tmpdir: Path) -> None:
    """write_records replaces old file cleanly; .tmp is removed."""
    target = tmpdir / "memory.ump.json"
    records = [
        make_rec(
            urn="urn:t7",
            kind="semantic",
            confidence=0.5,
            status="active",
            created=datetime.now(timezone.utc),
        )
    ]
    target.write_text(json.dumps(records))

    out, _ = ump_decay.apply_decay([dict(r) for r in records], dry_run=False)
    ump_decay.write_records(target, out)

    assert_true("Test 7: target file exists after write", target.exists())
    assert_true("Test 7: .tmp file removed by os.replace",
                not target.with_suffix(target.suffix + ".tmp").exists())
    assert_true("Test 7: written file is valid JSON array",
                isinstance(json.loads(target.read_text()), list))
    assert_eq("Test 7: round-trip preserves urn",
              json.loads(target.read_text())[0]["id"], "urn:t7")


def test_8_missing_lifecycle_defaults() -> None:
    """Records missing 'lifecycle' get defaults (confidence=1.0, status='active')."""
    now = datetime.now(timezone.utc)
    rec = {
        "id": "urn:t8",
        "kind": "semantic",
        "time": {"created": now.isoformat().replace("+00:00", "Z")},
    }
    out, _ = ump_decay.apply_decay([rec], dry_run=False)
    # Decay applied to a 0-day record: 1.0 * exp(-λ*0) should equal 1.0, but
    # float math yields ~1.0 - epsilon. Allow tiny tolerance.
    assert_eq("Test 8: missing lifecycle.confidence defaults ≈ 1.0",
              out[0]["lifecycle"]["confidence"], 1.0, tol=1e-9)
    assert_eq("Test 8: missing lifecycle.status → 'active'",
              out[0]["lifecycle"]["status"], "active")


def test_9_tombstoned_skipped() -> None:
    """Tombstoned records are skipped — confidence unchanged."""
    now = datetime.now(timezone.utc)
    created = now - timedelta(days=365)
    rec = make_rec(
        urn="urn:t9",
        kind="semantic",
        confidence=0.42,
        status="tombstoned",
        created=created,
    )
    out, report = ump_decay.apply_decay([rec], dry_run=False)
    assert_eq("Test 9: tombstoned confidence unchanged",
              out[0]["lifecycle"]["confidence"], 0.42)
    assert_eq("Test 9: tombstoned status unchanged",
              out[0]["lifecycle"]["status"], "tombstoned")
    assert_eq("Test 9: skipped_tombstoned count",
              report["skipped_tombstoned"], 1)


def test_10_modified_preferred_over_created() -> None:
    """time.modified is used for reference time when present."""
    now = datetime.now(timezone.utc)
    # Created 100 days ago, but modified 5 days ago.
    created = now - timedelta(days=100)
    modified = now - timedelta(days=5)
    rec = make_rec(
        urn="urn:t10",
        kind="semantic",
        confidence=1.0,
        status="active",
        created=created,
        modified=modified,
    )
    out, report = ump_decay.apply_decay([rec], dry_run=False)
    # λ=0.001, 5 days → exp(-0.005) ≈ 0.9950
    conf = out[0]["lifecycle"]["confidence"]
    assert_eq("Test 10: time.modified drives decay (5 days, semantic)",
              conf, 0.9950, tol=0.001)
    # days_since should reflect modified (5), not created (100).
    assert_eq("Test 10: days_since reflects modified, not created",
              round(report["changes"][0]["days_since"]), 5)


# ---- Runner ---------------------------------------------------------------

def main() -> int:
    print(f"Running ump_decay tests against {SCRIPT}\n")
    with tempfile.TemporaryDirectory(prefix="ump_decay_test_") as td:
        tmpdir = Path(td)

        test_1_semantic_one_month_factor()
        test_2_episodic_30_days_noticeable()
        test_3_identity_no_decay()
        test_4_archive_threshold()
        test_5_candidate_to_active()
        test_6_floor_at_0_05()
        test_7_atomic_write(tmpdir)
        test_8_missing_lifecycle_defaults()
        test_9_tombstoned_skipped()
        test_10_modified_preferred_over_created()

    passed = sum(1 for _, ok, _ in _results if ok)
    total = len(_results)
    print(f"\n{passed}/{total} tests passed")
    return 0 if passed == total else 1


if __name__ == "__main__":
    sys.exit(main())