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

#!/usr/bin/env python3
"""Run SX test specs against the Python SX evaluator.

The Python evaluator parses and evaluates test specs — SX tests itself.
This script provides only platform functions (error catching, reporting).

Usage:
    python shared/sx/tests/run.py                    # run all available specs
    python shared/sx/tests/run.py eval               # run only test-eval.sx
    python shared/sx/tests/run.py eval parser router  # run specific specs
    python shared/sx/tests/run.py --legacy           # run monolithic test.sx
"""
from __future__ import annotations

import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_PROJECT = os.path.abspath(os.path.join(_HERE, "..", "..", ".."))
sys.path.insert(0, _PROJECT)

from shared.sx.parser import parse_all
from shared.sx.evaluator import _eval, _trampoline, _call_lambda
from shared.sx.types import Symbol, Keyword, Lambda, NIL

# --- Test state ---
suite_stack: list[str] = []
passed = 0
failed = 0
test_num = 0


def try_call(thunk):
    """Call an SX thunk, catching errors."""
    try:
        _trampoline(_eval([thunk], env))
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": str(e)}


def report_pass(name):
    global passed, test_num
    test_num += 1
    passed += 1
    full_name = " > ".join(suite_stack + [name])
    print(f"ok {test_num} - {full_name}")


def report_fail(name, error):
    global failed, test_num
    test_num += 1
    failed += 1
    full_name = " > ".join(suite_stack + [name])
    print(f"not ok {test_num} - {full_name}")
    print(f"  # {error}")


def push_suite(name):
    suite_stack.append(name)


def pop_suite():
    suite_stack.pop()


# --- Parser platform functions ---

def sx_parse(source):
    """Parse SX source string into list of AST expressions."""
    return parse_all(source)


def sx_serialize(val):
    """Serialize an AST value to SX source text."""
    if val is None or val is NIL:
        return "nil"
    if isinstance(val, bool):
        return "true" if val else "false"
    if isinstance(val, (int, float)):
        return str(val)
    if isinstance(val, str):
        escaped = val.replace("\\", "\\\\").replace('"', '\\"')
        return f'"{escaped}"'
    if isinstance(val, Symbol):
        return val.name
    if isinstance(val, Keyword):
        return f":{val.name}"
    if isinstance(val, list):
        inner = " ".join(sx_serialize(x) for x in val)
        return f"({inner})"
    if isinstance(val, dict):
        parts = []
        for k, v in val.items():
            parts.append(f":{k}")
            parts.append(sx_serialize(v))
        return "{" + " ".join(parts) + "}"
    return str(val)


def make_symbol(name):
    return Symbol(name)


def make_keyword(name):
    return Keyword(name)


def symbol_name(sym):
    if isinstance(sym, Symbol):
        return sym.name
    return str(sym)


def keyword_name(kw):
    if isinstance(kw, Keyword):
        return kw.name
    return str(kw)


# --- Render platform function ---

def render_html(sx_source):
    """Parse SX source and render to HTML via the bootstrapped evaluator."""
    try:
        from shared.sx.ref.sx_ref import render_to_html as _render_to_html
    except ImportError:
        raise RuntimeError("render-to-html not available — sx_ref.py not built")
    exprs = parse_all(sx_source)
    render_env = dict(env)
    result = ""
    for expr in exprs:
        result += _render_to_html(expr, render_env)
    return result


# --- Spec registry ---

SPECS = {
    "eval":   {"file": "test-eval.sx",   "needs": []},
    "parser": {"file": "test-parser.sx", "needs": ["sx-parse"]},
    "router": {"file": "test-router.sx", "needs": []},
    "render": {"file": "test-render.sx", "needs": ["render-html"]},
    "deps":   {"file": "test-deps.sx",   "needs": []},
    "engine": {"file": "test-engine.sx", "needs": []},
    "orchestration": {"file": "test-orchestration.sx", "needs": []},
}

REF_DIR = os.path.join(_HERE, "..", "ref")


def eval_file(filename, env):
    """Load and evaluate an SX file."""
    filepath = os.path.join(REF_DIR, filename)
    if not os.path.exists(filepath):
        print(f"# SKIP {filename} (file not found)")
        return
    with open(filepath) as f:
        src = f.read()
    exprs = parse_all(src)
    for expr in exprs:
        _trampoline(_eval(expr, env))


# --- Build env ---
env = {
    "try-call": try_call,
    "report-pass": report_pass,
    "report-fail": report_fail,
    "push-suite": push_suite,
    "pop-suite": pop_suite,
    # Parser platform functions
    "sx-parse": sx_parse,
    "sx-serialize": sx_serialize,
    "make-symbol": make_symbol,
    "make-keyword": make_keyword,
    "symbol-name": symbol_name,
    "keyword-name": keyword_name,
    # Render platform function
    "render-html": render_html,
    # Extra primitives needed by spec modules (router.sx, deps.sx)
    "for-each-indexed": "_deferred",  # replaced below
    "dict-set!": "_deferred",
    "dict-has?": "_deferred",
    "dict-get": "_deferred",
    "append!": "_deferred",
    "inc": lambda n: n + 1,
    # Component accessor for affinity (Phase 7)
    "component-affinity": lambda c: getattr(c, 'affinity', 'auto'),
}


def _call_sx(fn, args, caller_env):
    """Call an SX lambda or native function with args."""
    if isinstance(fn, Lambda):
        return _trampoline(_call_lambda(fn, list(args), caller_env))
    return fn(*args)


def _for_each_indexed(fn, coll):
    """for-each-indexed that respects set! in lambda closures.

    The hand-written evaluator copies envs on lambda calls, which breaks
    set! mutation of outer scope. We eval directly in the closure dict
    to match the bootstrapped semantics (cell-based mutation).
    """
    if isinstance(fn, Lambda):
        closure = fn.closure
        for i, item in enumerate(coll or []):
            # Bind params directly in the closure (no copy)
            for p, v in zip(fn.params, [i, item]):
                closure[p] = v
            _trampoline(_eval(fn.body, closure))
    else:
        for i, item in enumerate(coll or []):
            fn(i, item)
    return NIL


def _dict_set(d, k, v):
    if isinstance(d, dict):
        d[k] = v
    return NIL


def _dict_has(d, k):
    return isinstance(d, dict) and k in d


def _dict_get(d, k):
    if isinstance(d, dict):
        return d.get(k, NIL)
    return NIL


def _append_mut(lst, item):
    if isinstance(lst, list):
        lst.append(item)
    return NIL


env["for-each-indexed"] = _for_each_indexed
env["dict-set!"] = _dict_set
env["dict-has?"] = _dict_has
env["dict-get"] = _dict_get
env["append!"] = _append_mut


def _load_router_from_bootstrap(env):
    """Load router functions from the bootstrapped sx_ref.py.

    The hand-written evaluator can't run router.sx faithfully because
    set! inside lambda closures doesn't propagate to outer scopes
    (the evaluator uses dict copies, not cells). The bootstrapped code
    compiles set! to cell-based mutation, so we import from there.
    """
    try:
        from shared.sx.ref.sx_ref import (
            split_path_segments,
            parse_route_pattern,
            match_route_segments,
            match_route,
            find_matching_route,
            make_route_segment,
        )
        env["split-path-segments"] = split_path_segments
        env["parse-route-pattern"] = parse_route_pattern
        env["match-route-segments"] = match_route_segments
        env["match-route"] = match_route
        env["find-matching-route"] = find_matching_route
        env["make-route-segment"] = make_route_segment
    except ImportError:
        # Fallback: eval router.sx directly (may fail on set! scoping)
        eval_file("router.sx", env)


def _load_deps_from_bootstrap(env):
    """Load deps functions from the bootstrapped sx_ref.py."""
    try:
        from shared.sx.ref.sx_ref import (
            scan_refs,
            scan_components_from_source,
            transitive_deps,
            compute_all_deps,
            components_needed,
            page_component_bundle,
            page_css_classes,
            scan_io_refs,
            transitive_io_refs,
            compute_all_io_refs,
            component_pure_p,
            render_target,
            page_render_plan,
        )
        env["scan-refs"] = scan_refs
        env["scan-components-from-source"] = scan_components_from_source
        env["transitive-deps"] = transitive_deps
        env["compute-all-deps"] = compute_all_deps
        env["components-needed"] = components_needed
        env["page-component-bundle"] = page_component_bundle
        env["page-css-classes"] = page_css_classes
        env["scan-io-refs"] = scan_io_refs
        env["transitive-io-refs"] = transitive_io_refs
        env["compute-all-io-refs"] = compute_all_io_refs
        env["component-pure?"] = component_pure_p
        env["render-target"] = render_target
        env["page-render-plan"] = page_render_plan
        env["test-env"] = lambda: env
    except ImportError:
        eval_file("deps.sx", env)
        env["test-env"] = lambda: env


def _load_engine_from_bootstrap(env):
    """Load engine pure functions from the bootstrapped sx_ref.py."""
    try:
        from shared.sx.ref.sx_ref import (
            parse_time,
            parse_trigger_spec,
            default_trigger,
            parse_swap_spec,
            parse_retry_spec,
            next_retry_ms,
            filter_params,
        )
        env["parse-time"] = parse_time
        env["parse-trigger-spec"] = parse_trigger_spec
        env["default-trigger"] = default_trigger
        env["parse-swap-spec"] = parse_swap_spec
        env["parse-retry-spec"] = parse_retry_spec
        env["next-retry-ms"] = next_retry_ms
        env["filter-params"] = filter_params
    except ImportError:
        eval_file("engine.sx", env)


def _load_orchestration(env):
    """Load orchestration.sx with mocked platform functions for testing.

    Orchestration defines many browser-wiring functions (DOM, fetch, etc.)
    but the Phase 7c/7d tests only exercise the cache, optimistic, and
    offline functions.  Lambda bodies referencing DOM/fetch are never called,
    so we only need to mock the functions actually invoked by the tests:
    now-ms, log-info, log-warn, execute-action, try-rerender-page.
    """
    _mock_ts = [1000]  # mutable so mock can advance time

    def _mock_now_ms():
        return _mock_ts[0]

    def _noop(*_a, **_kw):
        return NIL

    def _mock_execute_action(action, payload, on_success, on_error):
        """Mock: immediately calls on_success with payload as 'server truth'."""
        on_success(payload)
        return NIL

    def _dict_delete(d, k):
        if isinstance(d, dict) and k in d:
            del d[k]
        return NIL

    env["now-ms"] = _mock_now_ms
    env["log-info"] = _noop
    env["log-warn"] = _noop
    env["execute-action"] = _mock_execute_action
    env["try-rerender-page"] = _noop
    env["persist-offline-data"] = _noop
    env["retrieve-offline-data"] = lambda: NIL
    env["dict-delete!"] = _dict_delete
    # DOM / browser stubs (never called by tests, but referenced in lambdas
    # that the evaluator might try to resolve at call time)
    for stub in [
        "try-parse-json", "dom-dispatch", "dom-query-selector",
        "dom-get-attribute", "dom-set-attribute", "dom-set-text-content",
        "dom-append", "dom-insert-html-adjacent", "dom-remove",
        "dom-outer-html", "dom-inner-html", "dom-create-element",
        "dom-set-inner-html", "dom-morph", "dom-get-tag",
        "dom-query-selector-all", "dom-add-event-listener",
        "dom-set-timeout", "dom-prevent-default", "dom-closest",
        "dom-matches", "dom-get-id", "dom-set-id", "dom-form-data",
        "dom-is-form", "browser-location-href", "browser-push-state",
        "browser-replace-state", "sx-hydrate-elements", "render-to-dom",
        "hoist-head-elements-full", "url-pathname",
    ]:
        if stub not in env:
            env[stub] = _noop

    # Load engine.sx first (orchestration depends on it)
    _load_engine_from_bootstrap(env)
    # Load orchestration.sx
    eval_file("orchestration.sx", env)


def _load_forms_from_bootstrap(env):
    """Load forms functions (including streaming protocol) from sx_ref.py."""
    try:
        from shared.sx.ref.sx_ref import (
            stream_chunk_id,
            stream_chunk_bindings,
            normalize_binding_key,
            bind_stream_chunk,
            validate_stream_data,
        )
        env["stream-chunk-id"] = stream_chunk_id
        env["stream-chunk-bindings"] = stream_chunk_bindings
        env["normalize-binding-key"] = normalize_binding_key
        env["bind-stream-chunk"] = bind_stream_chunk
        env["validate-stream-data"] = validate_stream_data
    except ImportError:
        eval_file("forms.sx", env)


def main():
    global passed, failed, test_num

    args = sys.argv[1:]

    # Legacy mode
    if args and args[0] == "--legacy":
        print("TAP version 13")
        eval_file("test.sx", env)
    else:
        # Determine which specs to run
        specs_to_run = args if args else list(SPECS.keys())

        print("TAP version 13")

        # Always load framework first
        eval_file("test-framework.sx", env)

        for spec_name in specs_to_run:
            spec = SPECS.get(spec_name)
            if not spec:
                print(f"# SKIP unknown spec: {spec_name}")
                continue

            # Check platform requirements
            can_run = True
            for need in spec["needs"]:
                if need not in env:
                    print(f"# SKIP {spec_name} (missing: {need})")
                    can_run = False
                    break
            if not can_run:
                continue

            # Load prerequisite spec modules
            if spec_name == "eval":
                _load_forms_from_bootstrap(env)
            if spec_name == "router":
                _load_router_from_bootstrap(env)
            if spec_name == "deps":
                _load_deps_from_bootstrap(env)
            if spec_name == "engine":
                _load_engine_from_bootstrap(env)
            if spec_name == "orchestration":
                _load_orchestration(env)

            print(f"# --- {spec_name} ---")
            eval_file(spec["file"], env)

    # Summary
    print()
    print(f"1..{test_num}")
    print(f"# tests {passed + failed}")
    print(f"# pass  {passed}")
    if failed > 0:
        print(f"# fail  {failed}")
        sys.exit(1)


if __name__ == "__main__":
    main()