"""Keeping the heavyweight imports out of a pod that is only serving pages.

Measured in the container: a bare interpreter is 8.9 MiB, and importing
yfinance takes it to 116.6 MiB. So `pricefeed` imports yfinance on first use and
`statements` imports pdfplumber on first use, and these tests stop either
drifting back to module scope — an invisible regression, because the app works
either way and only the memory graph changes.

What that buys in a running deployment with the feed on: nothing, since the
startup catch-up run imports it anyway (decisions.md #17). It protects the
install that turns the feed off, every CLI entry point, and the suite itself —
which is why these spawn fresh interpreters rather than trusting `sys.modules`
in a process the suite has already dirtied.

`memory.release()` is tested for honesty about what it could not do rather than
for a number, because the number depends on the allocator and the platform.
"""

from __future__ import annotations

import subprocess
import sys
import textwrap
from pathlib import Path

import pytest

from app import memory

APP_ROOT = Path(__file__).resolve().parent.parent
HEAVY = ("yfinance", "pandas", "numpy", "pdfplumber")


def _in_fresh_interpreter(body: str) -> str:
    """Run a snippet in a new process, with the app importable.

    A fresh interpreter is the whole point: by the time this test file runs,
    another test may already have imported anything, so asking `sys.modules` in
    THIS process would prove nothing about what starting the app costs.
    """
    script = textwrap.dedent(f"""
        import os, sys
        sys.path[:0] = [{str(APP_ROOT)!r}, {str(APP_ROOT.parent.parent / "libs" / "appcore")!r}]
        os.environ.setdefault("APP_CONFIG_FILE", {str(APP_ROOT / "tests" / "config.test.yaml")!r})
        os.environ.setdefault("APP_DATABASE__TYPE", "sqlite")
        os.environ.setdefault("APP_DATABASE__PATH", "/tmp/pf-memtest.db")
        {body}
    """)
    result = subprocess.run(
        [sys.executable, "-c", script], capture_output=True, text=True, timeout=180
    )
    assert result.returncode == 0, f"probe failed:\n{result.stdout}\n{result.stderr}"
    return result.stdout.strip()


# --------------------------------------------------------------------------- #
# The imports stay lazy
# --------------------------------------------------------------------------- #

@pytest.mark.slow
def test_importing_the_app_does_not_load_the_heavyweights():
    """The regression this whole task is about. If someone moves `import
    yfinance` back to the top of pricefeed.py, every pod grows ~108 MiB at
    startup and nothing else changes — no test fails, no page breaks. This one
    fails."""
    loaded = _in_fresh_interpreter("""
        import app.main  # noqa: F401
        print(",".join(sorted(m for m in %r if m in sys.modules)))
    """ % (HEAVY,))

    assert loaded == "", f"app import pulled in: {loaded}"


@pytest.mark.slow
def test_the_feed_loads_yfinance_when_it_actually_runs():
    """The mirror: lazy must mean deferred, not broken. `_yahoo()` has to
    really import the library when something needs it."""
    loaded = _in_fresh_interpreter("""
        from app import pricefeed
        assert pricefeed._yf is None, "cached before first use"
        module = pricefeed._yahoo()
        assert module is not None
        assert pricefeed._yf is module, "second call must reuse the import"
        print("yfinance" if "yfinance" in sys.modules else "missing")
    """)

    assert loaded == "yfinance"


@pytest.mark.slow
def test_statements_loads_pdfplumber_only_when_parsing():
    loaded = _in_fresh_interpreter("""
        from app import statements
        before = "pdfplumber" in sys.modules
        try:
            statements.extract_text(b"not a pdf")
        except Exception:
            pass  # it is not a PDF; we only care that the import happened
        print(f"{before},{'pdfplumber' in sys.modules}")
    """)

    assert loaded == "False,True"


# --------------------------------------------------------------------------- #
# release()
# --------------------------------------------------------------------------- #

def test_release_collects_and_reports_what_it_did():
    # Cycles only the collector can free, so "collected" has something to count;
    # `>= 0` alone could not fail.
    for _ in range(50):
        cycle: dict = {}
        cycle["self"] = cycle
    del cycle

    result = memory.release("a test")

    assert result["collected"] >= 50
    assert isinstance(result["trimmed"], bool)
    # rss/freed are None off Linux; the shape must be stable either way so the
    # log line and any future dashboard can rely on it.
    assert set(result) == {"collected", "trimmed", "rss_before", "rss_after", "freed"}


def test_release_never_raises_when_the_platform_cannot_trim(monkeypatch):
    """`malloc_trim` is a glibc extension: absent on musl and on macOS. A feed
    run must not fail because the allocator wouldn't co-operate — the work it
    just did is already committed."""
    monkeypatch.setattr(memory.ctypes.util, "find_library", lambda _name: None)

    result = memory.release()

    assert result["trimmed"] is False


def test_release_survives_a_libc_without_the_symbol(monkeypatch):
    class _NoTrim:
        def __getattr__(self, name):
            raise AttributeError(name)

    monkeypatch.setattr(memory.ctypes, "CDLL", lambda _name: _NoTrim())

    assert memory.release()["trimmed"] is False


def test_rss_is_read_from_the_cgroup_when_there_is_one(monkeypatch, tmp_path):
    """The cgroup figure, not VmRSS: that is the number the limit is enforced
    against, and the one that gets the pod OOM-killed."""
    current = tmp_path / "memory.current"
    current.write_text(str(200 * 1048576), encoding="utf-8")
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(current))

    assert memory.rss_mib() == 200.0


def test_rss_falls_back_to_proc_when_there_is_no_cgroup(monkeypatch, tmp_path):
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(tmp_path / "absent"))

    value = memory.rss_mib()

    # Linux has /proc/self/status and returns a number; macOS has neither and
    # returns None. Both are correct — what must not happen is an exception.
    assert value is None or value > 0


def test_a_garbled_cgroup_file_is_ignored_rather_than_fatal(monkeypatch, tmp_path):
    current = tmp_path / "memory.current"
    current.write_text("not a number", encoding="utf-8")
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(current))

    value = memory.rss_mib()

    assert value is None or value > 0  # fell through to /proc, or gave up


def test_the_limit_is_read_when_one_is_set(monkeypatch, tmp_path):
    (tmp_path / "memory.current").write_text("1", encoding="utf-8")
    (tmp_path / "memory.max").write_text(str(512 * 1048576), encoding="utf-8")
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(tmp_path / "memory.current"))

    assert memory.limit_mib() == 512.0


def test_an_unlimited_cgroup_reports_no_limit(monkeypatch, tmp_path):
    (tmp_path / "memory.current").write_text("1", encoding="utf-8")
    (tmp_path / "memory.max").write_text("max", encoding="utf-8")
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(tmp_path / "memory.current"))

    assert memory.limit_mib() is None


def test_the_limit_is_none_outside_a_cgroup(monkeypatch, tmp_path):
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(tmp_path / "nothing" / "memory.current"))

    assert memory.limit_mib() is None


def test_loaded_heavyweights_reports_what_is_in_memory():
    """Used in the startup log line, so a 190 MiB pod can be explained without
    attaching a profiler to it."""
    import pandas  # noqa: F401  — deliberately loaded, to have something to find

    loaded = memory.loaded_heavyweights()

    assert "pandas" in loaded
    assert loaded == sorted(loaded)


def test_log_snapshot_does_not_raise_without_a_cgroup(monkeypatch, tmp_path, caplog):
    monkeypatch.setattr(memory, "_CGROUP_CURRENT", str(tmp_path / "absent"))

    memory.log_snapshot("in a test")  # must be safe to call anywhere


# --------------------------------------------------------------------------- #
# Wiring
# --------------------------------------------------------------------------- #

def test_a_feed_run_releases_memory_afterwards(session_factory, monkeypatch):
    """The trim has to be attached to the thing that allocates. A feed run that
    doesn't release leaves the pod at its high-water mark until it restarts."""
    import datetime as dt
    from types import SimpleNamespace

    from app import pricefeed, tenancy
    from app.settings import PriceFeedSettings

    calls = []
    monkeypatch.setattr(memory, "release", lambda context="": calls.append(context))
    monkeypatch.setattr(pricefeed, "_closes", lambda *a, **k: [])
    monkeypatch.setattr(pricefeed, "_yf", SimpleNamespace(Ticker=lambda _s: None))
    settings = SimpleNamespace(
        price_feed=PriceFeedSettings(backfill_start=dt.date(2020, 1, 1))
    )

    with tenancy.unscoped_session(session_factory) as session:
        pricefeed.run_feed(session, settings)

    assert calls == ["feed run"]