rose-ash/shared/sx/tests/test_io_detection.py

"""Tests for Phase 2 IO detection — component purity analysis.

Tests both the hand-written fallback (deps.py) and the bootstrapped
sx_ref.py implementation of IO reference scanning and transitive
IO classification.
"""

import os
import pytest
from shared.sx.parser import parse_all
from shared.sx.types import Component, Macro, Symbol
from shared.sx.deps import (
    _scan_io_refs_fallback,
    _transitive_io_refs_fallback,
    _compute_all_io_refs_fallback,
    compute_all_io_refs,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def make_env(*sx_sources: str) -> dict:
    """Parse and evaluate component definitions into an env dict."""
    from shared.sx.ref.sx_ref import eval_expr as _eval, trampoline as _trampoline
    env: dict = {}
    for source in sx_sources:
        exprs = parse_all(source)
        for expr in exprs:
            _trampoline(_eval(expr, env))
    return env


IO_NAMES = {"fetch-data", "call-action", "app-url", "config", "db-query"}


# ---------------------------------------------------------------------------
# _scan_io_refs_fallback — scan single AST for IO primitives
# ---------------------------------------------------------------------------

class TestScanIoRefs:
    def test_no_io_refs(self):
        env = make_env('(defcomp ~card (&key title) (div :class "p-4" title))')
        comp = env["~card"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == set()

    def test_direct_io_ref(self):
        env = make_env('(defcomp ~page (&key) (div (fetch-data "posts")))')
        comp = env["~page"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == {"fetch-data"}

    def test_multiple_io_refs(self):
        env = make_env(
            '(defcomp ~page (&key) (div (fetch-data "x") (config "y")))'
        )
        comp = env["~page"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == {"fetch-data", "config"}

    def test_io_in_nested_control_flow(self):
        env = make_env(
            '(defcomp ~page (&key show) '
            '  (if show (div (app-url "/")) (span "none")))'
        )
        comp = env["~page"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == {"app-url"}

    def test_io_in_dict_value(self):
        env = make_env(
            '(defcomp ~wrap (&key) (div {:data (db-query "x")}))'
        )
        comp = env["~wrap"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == {"db-query"}

    def test_non_io_symbol_ignored(self):
        """Symbols that aren't in the IO set should not be detected."""
        env = make_env(
            '(defcomp ~card (&key) (div (str "hello") (len "world")))'
        )
        comp = env["~card"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == set()

    def test_component_ref_not_io(self):
        """Component references (~name) should not appear as IO refs."""
        env = make_env(
            '(defcomp ~page (&key) (div (~card :title "hi")))',
            '(defcomp ~card (&key title) (div title))',
        )
        comp = env["~page"]
        refs = _scan_io_refs_fallback(comp.body, IO_NAMES)
        assert refs == set()


# ---------------------------------------------------------------------------
# _transitive_io_refs_fallback — follow deps to find all IO refs
# ---------------------------------------------------------------------------

class TestTransitiveIoRefs:
    def test_pure_component(self):
        env = make_env(
            '(defcomp ~card (&key title) (div title))',
        )
        refs = _transitive_io_refs_fallback("~card", env, IO_NAMES)
        assert refs == set()

    def test_direct_io(self):
        env = make_env(
            '(defcomp ~page (&key) (div (fetch-data "posts")))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert refs == {"fetch-data"}

    def test_transitive_io_through_dep(self):
        """IO ref in a dependency should propagate to the parent."""
        env = make_env(
            '(defcomp ~page (&key) (div (~nav)))',
            '(defcomp ~nav (&key) (nav (app-url "/home")))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert refs == {"app-url"}

    def test_multiple_transitive_io(self):
        """IO refs from multiple deps should be unioned."""
        env = make_env(
            '(defcomp ~page (&key) (div (~header) (~footer)))',
            '(defcomp ~header (&key) (nav (app-url "/")))',
            '(defcomp ~footer (&key) (footer (config "site-name")))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert refs == {"app-url", "config"}

    def test_deep_transitive_io(self):
        """IO refs should propagate through multiple levels."""
        env = make_env(
            '(defcomp ~page (&key) (div (~layout)))',
            '(defcomp ~layout (&key) (div (~sidebar)))',
            '(defcomp ~sidebar (&key) (nav (fetch-data "menu")))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert refs == {"fetch-data"}

    def test_circular_deps_no_infinite_loop(self):
        """Circular component references should not cause infinite recursion."""
        env = make_env(
            '(defcomp ~a (&key) (div (~b) (app-url "/")))',
            '(defcomp ~b (&key) (div (~a)))',
        )
        refs = _transitive_io_refs_fallback("~a", env, IO_NAMES)
        assert refs == {"app-url"}

    def test_without_tilde_prefix(self):
        """Should auto-add ~ prefix when not provided."""
        env = make_env(
            '(defcomp ~nav (&key) (nav (app-url "/")))',
        )
        refs = _transitive_io_refs_fallback("nav", env, IO_NAMES)
        assert refs == {"app-url"}

    def test_missing_dep_component(self):
        """Referencing a component not in env should not crash."""
        env = make_env(
            '(defcomp ~page (&key) (div (~unknown) (fetch-data "x")))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert refs == {"fetch-data"}

    def test_macro_io_detection(self):
        """IO refs in macros should be detected too."""
        env = make_env(
            '(defmacro ~with-data (body) (list (quote div) (list (quote fetch-data) "x") body))',
            '(defcomp ~page (&key) (div (~with-data (span "hi"))))',
        )
        refs = _transitive_io_refs_fallback("~page", env, IO_NAMES)
        assert "fetch-data" in refs


# ---------------------------------------------------------------------------
# _compute_all_io_refs_fallback — batch computation
# ---------------------------------------------------------------------------

class TestComputeAllIoRefs:
    def test_sets_io_refs_on_components(self):
        env = make_env(
            '(defcomp ~page (&key) (div (~nav) (fetch-data "x")))',
            '(defcomp ~nav (&key) (nav (app-url "/")))',
            '(defcomp ~card (&key title) (div title))',
        )
        _compute_all_io_refs_fallback(env, IO_NAMES)
        assert env["~page"].io_refs == {"fetch-data", "app-url"}
        assert env["~nav"].io_refs == {"app-url"}
        assert env["~card"].io_refs == set()

    def test_pure_components_get_empty_set(self):
        env = make_env(
            '(defcomp ~a (&key) (div "hello"))',
            '(defcomp ~b (&key) (span "world"))',
        )
        _compute_all_io_refs_fallback(env, IO_NAMES)
        assert env["~a"].io_refs == set()
        assert env["~b"].io_refs == set()

    def test_transitive_io_via_compute_all(self):
        """Transitive IO refs should be cached on the parent component."""
        env = make_env(
            '(defcomp ~page (&key) (div (~child)))',
            '(defcomp ~child (&key) (div (config "key")))',
        )
        _compute_all_io_refs_fallback(env, IO_NAMES)
        assert env["~page"].io_refs == {"config"}
        assert env["~child"].io_refs == {"config"}


# ---------------------------------------------------------------------------
# Public API dispatch — compute_all_io_refs
# ---------------------------------------------------------------------------

class TestPublicApiIoRefs:
    def test_fallback_mode(self):
        """Public API should work in fallback mode (SX_USE_REF != 1)."""
        env = make_env(
            '(defcomp ~page (&key) (div (fetch-data "x")))',
            '(defcomp ~leaf (&key) (span "pure"))',
        )
        old_val = os.environ.get("SX_USE_REF")
        try:
            os.environ.pop("SX_USE_REF", None)
            compute_all_io_refs(env, IO_NAMES)
            assert env["~page"].io_refs == {"fetch-data"}
            assert env["~leaf"].io_refs == set()
        finally:
            if old_val is not None:
                os.environ["SX_USE_REF"] = old_val

    def test_ref_mode(self):
        """Public API should work with bootstrapped sx_ref.py (SX_USE_REF=1)."""
        env = make_env(
            '(defcomp ~page (&key) (div (fetch-data "x")))',
            '(defcomp ~leaf (&key) (span "pure"))',
        )
        old_val = os.environ.get("SX_USE_REF")
        try:
            os.environ["SX_USE_REF"] = "1"
            compute_all_io_refs(env, IO_NAMES)
            # sx_ref returns lists, compute_all_io_refs converts as needed
            page_refs = env["~page"].io_refs
            leaf_refs = env["~leaf"].io_refs
            # May be list or set depending on backend
            assert "fetch-data" in page_refs
            assert len(leaf_refs) == 0
        finally:
            if old_val is not None:
                os.environ["SX_USE_REF"] = old_val
            else:
                os.environ.pop("SX_USE_REF", None)


# ---------------------------------------------------------------------------
# Bootstrapped sx_ref.py IO functions — direct testing
# ---------------------------------------------------------------------------

class TestSxRefIoFunctions:
    """Test the bootstrapped sx_ref.py IO functions directly."""

    def test_scan_io_refs(self):
        from shared.sx.ref.sx_ref import scan_io_refs
        env = make_env('(defcomp ~page (&key) (div (fetch-data "x") (config "y")))')
        comp = env["~page"]
        refs = scan_io_refs(comp.body, list(IO_NAMES))
        assert set(refs) == {"fetch-data", "config"}

    def test_scan_io_refs_no_match(self):
        from shared.sx.ref.sx_ref import scan_io_refs
        env = make_env('(defcomp ~card (&key title) (div title))')
        comp = env["~card"]
        refs = scan_io_refs(comp.body, list(IO_NAMES))
        assert refs == []

    def test_transitive_io_refs(self):
        from shared.sx.ref.sx_ref import transitive_io_refs
        env = make_env(
            '(defcomp ~page (&key) (div (~nav)))',
            '(defcomp ~nav (&key) (nav (app-url "/")))',
        )
        refs = transitive_io_refs("~page", env, list(IO_NAMES))
        assert set(refs) == {"app-url"}

    def test_transitive_io_refs_pure(self):
        from shared.sx.ref.sx_ref import transitive_io_refs
        env = make_env('(defcomp ~card (&key) (div "hi"))')
        refs = transitive_io_refs("~card", env, list(IO_NAMES))
        assert refs == []

    def test_compute_all_io_refs(self):
        from shared.sx.ref.sx_ref import compute_all_io_refs as ref_compute
        env = make_env(
            '(defcomp ~page (&key) (div (~nav) (fetch-data "x")))',
            '(defcomp ~nav (&key) (nav (app-url "/")))',
            '(defcomp ~card (&key) (div "pure"))',
        )
        ref_compute(env, list(IO_NAMES))
        page_refs = env["~page"].io_refs
        nav_refs = env["~nav"].io_refs
        card_refs = env["~card"].io_refs
        assert "fetch-data" in page_refs
        assert "app-url" in page_refs
        assert "app-url" in nav_refs
        assert len(card_refs) == 0

    def test_component_pure_p(self):
        from shared.sx.ref.sx_ref import component_pure_p
        env = make_env(
            '(defcomp ~pure-card (&key) (div "hello"))',
            '(defcomp ~io-card (&key) (div (fetch-data "x")))',
        )
        io_list = list(IO_NAMES)
        assert component_pure_p("~pure-card", env, io_list) is True
        assert component_pure_p("~io-card", env, io_list) is False

    def test_component_pure_p_transitive(self):
        """A component is impure if any transitive dep uses IO."""
        from shared.sx.ref.sx_ref import component_pure_p
        env = make_env(
            '(defcomp ~page (&key) (div (~child)))',
            '(defcomp ~child (&key) (div (config "key")))',
        )
        io_list = list(IO_NAMES)
        assert component_pure_p("~page", env, io_list) is False
        assert component_pure_p("~child", env, io_list) is False


# ---------------------------------------------------------------------------
# Parity: fallback vs bootstrapped produce same results
# ---------------------------------------------------------------------------

class TestFallbackVsRefParity:
    """Ensure fallback Python and bootstrapped sx_ref.py agree."""

    def _check_parity(self, *sx_sources: str):
        """Run both implementations and verify io_refs match."""
        from shared.sx.ref.sx_ref import compute_all_io_refs as ref_compute

        # Run fallback
        env_fb = make_env(*sx_sources)
        _compute_all_io_refs_fallback(env_fb, IO_NAMES)

        # Run bootstrapped
        env_ref = make_env(*sx_sources)
        ref_compute(env_ref, list(IO_NAMES))

        # Compare all components
        for key in env_fb:
            if isinstance(env_fb[key], Component):
                fb_refs = env_fb[key].io_refs or set()
                ref_refs = env_ref[key].io_refs
                # Normalize: fallback returns set, ref returns list/set
                assert set(fb_refs) == set(ref_refs), (
                    f"Mismatch for {key}: fallback={fb_refs}, ref={set(ref_refs)}"
                )

    def test_parity_pure_components(self):
        self._check_parity(
            '(defcomp ~a (&key) (div "hello"))',
            '(defcomp ~b (&key) (span (~a)))',
        )

    def test_parity_io_components(self):
        self._check_parity(
            '(defcomp ~page (&key) (div (~header) (fetch-data "x")))',
            '(defcomp ~header (&key) (nav (app-url "/")))',
            '(defcomp ~footer (&key) (footer "static"))',
        )

    def test_parity_deep_chain(self):
        self._check_parity(
            '(defcomp ~a (&key) (div (~b)))',
            '(defcomp ~b (&key) (div (~c)))',
            '(defcomp ~c (&key) (div (config "x")))',
        )

    def test_parity_mixed(self):
        self._check_parity(
            '(defcomp ~layout (&key) (div (~nav) (~content) (~footer)))',
            '(defcomp ~nav (&key) (nav (app-url "/")))',
            '(defcomp ~content (&key) (main "pure content"))',
            '(defcomp ~footer (&key) (footer (config "name")))',
        )