nexa / tests /test_env.py
ascii: make the fallback reachable, and the clone survive the gateway
3432acf about 1 month ago
24.6 kB
"""The documented environment variables, none of which had ever been set.
README, ARCHITECTURE and `nexa help` between them promise:
    NEXA_EVIDENCE=0          NEXA_EVIDENCE_DIR      NEXA_EVIDENCE_KEEP
    LOCALAI_TELEMETRY=0      LOCALAI_TELEMETRY_CONTENT=1
    LOCALAI_TELEMETRY_FILE   NO_COLOR / TERM=dumb
plus OLLAMA_HOST, which bin/endpoint.py added. Documenting a switch nobody has
ever flipped is a promise, not a feature.
evidence.py and telemetry.py read their settings into module constants at
import. That is right for a CLI that runs one process per command, and useless
to a test living in a long-lived interpreter -- so every test here either
rebuilds the module under a chosen environment or crosses a process boundary.
The subprocess ones are the ones that matter: they set the variable the way a
user does and then look at what happened on disk.
"""
import importlib
import io
import json
import os
import subprocess
import sys

import pytest

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# Modules whose behaviour is fixed at import. Restored after every test that
# rebuilds one, so a reconfigured module cannot leak into the next test.
ENV_BOUND = ("evidence", "telemetry")


@pytest.fixture()
def with_env(monkeypatch):
    """Rebuild an env-bound module under a chosen environment.
    The environment stays in force for the body of the test: an earlier version
    of this restored it in a finally block and reloaded the module again on the
    way out, which handed the test back a module configured with the defaults it
    was trying to override. Every assertion passed for the wrong reason.
    """
    def configure(module_name, environ):
        for key, value in environ.items():
            if value is None:
                monkeypatch.delenv(key, raising=False)
            else:
                monkeypatch.setenv(key, value)
        return importlib.reload(importlib.import_module(module_name))

    yield configure

    # monkeypatch has already put the environment back by the time this runs.
    for name in ENV_BOUND:
        if name in sys.modules:
            importlib.reload(sys.modules[name])


# --------------------------------------------------------------- NEXA_EVIDENCE

def test_evidence_off_writes_nothing(with_env, tmp_path):
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "0",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path)})
    evidence.write(evidence.new_bundle("t-1", "odoo", "q", "m", {}, {}))
    assert list(tmp_path.iterdir()) == [], "NEXA_EVIDENCE=0 still wrote a bundle"
    assert evidence.status()["enabled"] is False
    assert evidence.status()["directory"] is None


def test_evidence_dir_is_honoured(with_env, tmp_path):
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path)})
    evidence.write(evidence.new_bundle("t-2", "odoo", "q", "m", {}, {}))
    assert [p.name for p in tmp_path.iterdir()] == ["t-2.json"]


def test_evidence_keep_prunes_the_oldest(with_env, tmp_path):
    """The pruning promise, at a number small enough to watch.
    NEXA_EVIDENCE_KEEP defaults to 200, so nothing on this machine had ever
    reached it: this is the first time the branch has executed.
    """
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path),
                                     "NEXA_EVIDENCE_KEEP": "3"})
    for n in range(8):
        evidence.write(evidence.new_bundle("t-{0:02d}".format(n), "odoo",
                                           "q", "m", {}, {}))
    remaining = sorted(p.name for p in tmp_path.iterdir())
    assert len(remaining) == 3, remaining
    # Pruning by count is only useful if it prunes the right end.
    assert "t-07.json" in remaining
    assert "t-00.json" not in remaining


def test_pruning_does_not_depend_on_timestamp_resolution(with_env, tmp_path,
                                                        monkeypatch):
    """The same promise, on the filesystem CI actually runs on.
    Linux stamps inodes from a clock that only moves once a tick, so eight
    bundles written inside one millisecond share a single mtime. sort() is
    stable, so the tie fell through to os.listdir order -- hash order on ext4 --
    and the three survivors were an arbitrary subset. One CI run pruned the
    bundle write() had that moment written and returned its path regardless.
    Windows stamps at sub-millisecond resolution, so the test above passes on
    the machine this is written on whether the bug is there or not, which is
    why it sat unnoticed. This one ties every mtime and hands the directory
    back oldest-first, the order that makes a stable sort prune the newest end.
    """
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path),
                                     "NEXA_EVIDENCE_KEEP": "3"})
    real_listdir = os.listdir
    monkeypatch.setattr(os.path, "getmtime", lambda path: 1789520000.0)
    monkeypatch.setattr(os, "listdir", lambda path: sorted(real_listdir(path)))

    written = None
    for n in range(8):
        written = evidence.write(evidence.new_bundle("t-{0:02d}".format(n),
                                                     "odoo", "q", "m", {}, {}))

    # iterdir() goes through scandir, so it sees the real directory, not the
    # one patched above.
    remaining = sorted(p.name for p in tmp_path.iterdir())
    assert remaining == ["t-05.json", "t-06.json", "t-07.json"], remaining
    assert written and os.path.exists(written), "it pruned what it just wrote"


def test_evidence_status_reports_the_settings_in_force(with_env, tmp_path):
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path),
                                     "NEXA_EVIDENCE_KEEP": "7"})
    status = evidence.status()
    assert status["enabled"] is True
    assert status["keep"] == 7
    assert os.path.normcase(status["directory"]) == os.path.normcase(str(tmp_path))


# ------------------------------------------------------------ LOCALAI_TELEMETRY

def test_telemetry_off_writes_nothing(with_env, tmp_path):
    target = tmp_path / "telemetry.jsonl"
    telemetry = with_env("telemetry", {"LOCALAI_TELEMETRY": "0",
                                       "LOCALAI_TELEMETRY_FILE": str(target)})
    telemetry.emit(telemetry.EVENT_RUN, {"outcome": "verified_fast"})
    assert not target.exists()
    assert telemetry.status()["enabled"] is False
    # None, not the path. Naming a file nothing writes to would make
    # `nexa report status` point at a stale artefact as though it were live.
    assert telemetry.status()["file"] is None


def test_telemetry_file_is_honoured(with_env, tmp_path):
    target = tmp_path / "elsewhere.jsonl"
    telemetry = with_env("telemetry", {"LOCALAI_TELEMETRY": "1",
                                       "LOCALAI_TELEMETRY_FILE": str(target)})
    telemetry.emit(telemetry.EVENT_RUN, {"outcome": "verified_fast"})
    rows = [json.loads(line) for line in io.open(str(target), encoding="utf-8")
            if line.strip()]
    assert len(rows) == 1
    assert rows[0]["outcome"] == "verified_fast"


def test_content_capture_is_off_by_default(with_env, tmp_path):
    telemetry = with_env("telemetry", {
        "LOCALAI_TELEMETRY": "1",
        "LOCALAI_TELEMETRY_CONTENT": None,
        "LOCALAI_TELEMETRY_FILE": str(tmp_path / "t.jsonl")})
    assert telemetry.content("the question text") is None
    assert telemetry.status()["content_capture"] is False


def test_content_capture_on_returns_the_text(with_env, tmp_path):
    telemetry = with_env("telemetry", {
        "LOCALAI_TELEMETRY": "1",
        "LOCALAI_TELEMETRY_CONTENT": "1",
        "LOCALAI_TELEMETRY_FILE": str(tmp_path / "t.jsonl")})
    assert telemetry.content("the question text") == "the question text"
    assert telemetry.status()["content_capture"] is True


def test_evidence_and_telemetry_share_the_content_switch(with_env, tmp_path):
    """One variable, both subsystems. Two switches for "record what I typed" is
    one switch too many to get right under pressure."""
    evidence = with_env("evidence", {"LOCALAI_TELEMETRY_CONTENT": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path)})
    assert "the question" in json.dumps(
        evidence.new_bundle("t-3", "odoo", "the question", "m", {}, {}))


def test_content_omitted_from_the_bundle_by_default(with_env, tmp_path):
    evidence = with_env("evidence", {"LOCALAI_TELEMETRY_CONTENT": "0",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path)})
    assert "the question" not in json.dumps(
        evidence.new_bundle("t-4", "odoo", "the question", "m", {}, {}))


# ---------------------------------------------------------------- OLLAMA_HOST

@pytest.mark.parametrize("value,expected", [
    (None, "http://127.0.0.1:11434"),
    ("", "http://127.0.0.1:11434"),
    ("   ", "http://127.0.0.1:11434"),
    ("11500", "http://127.0.0.1:11500"),
    ("box:11434", "http://box:11434"),
    ("http://box:11434/", "http://box:11434"),
    ("https://gpu.internal", "https://gpu.internal"),
])
def test_ollama_host_spellings(value, expected):
    """Every spelling Ollama's own server accepts. A bare port is how its
    documentation says "same host, other port"."""
    import endpoint
    env = {} if value is None else {"OLLAMA_HOST": value}
    assert endpoint.base_url(env) == expected


def test_ollama_host_source_is_reported():
    import endpoint
    assert endpoint.describe({})[1] == "default"
    assert endpoint.describe({"OLLAMA_HOST": "box:1"})[1] == "OLLAMA_HOST"
    # Whitespace is not an override. Calling it one would send `nexa doctor`
    # hunting for a server at an address nobody meant to set.
    assert endpoint.describe({"OLLAMA_HOST": "  "})[1] == "default"


# ------------------------------------------------- NO_COLOR / TERM, end to end

def cli(args, env=None):
    """A real subprocess. COLOR is decided at import against sys.stdout, so an
    in-process call would be measuring pytest's capture, not a terminal."""
    environment = dict(os.environ, PYTHONIOENCODING="utf-8")
    environment.update(env or {})
    proc = subprocess.run([sys.executable, os.path.join(ROOT, "nexa.py")] + args,
                          capture_output=True, text=True, encoding="utf-8",
                          errors="replace", cwd=ROOT, env=environment)
    return proc.returncode, proc.stdout or ""


@pytest.mark.parametrize("env", [
    {"NO_COLOR": "1"},
    {"TERM": "dumb"},
    {"NO_COLOR": "1", "TERM": "dumb"},
])
def test_documented_colour_opt_outs_emit_no_escapes(env):
    code, out = cli(["rules", "--stack", "odoo"], env)
    assert code == 0
    assert "\033" not in out, "an ANSI escape survived " + repr(env)


def test_colour_is_off_when_stdout_is_not_a_terminal():
    """A pipe is not a terminal, so colour is off even with nothing set.
    Escape codes landing in a redirected file is the complaint that made
    NO_COLOR a standard in the first place."""
    code, out = cli(["rules", "--stack", "odoo"])
    assert code == 0
    assert "\033" not in out


def _art_rows(art):
    """The art's rows, indentation intact, without the blank lines the triple-
    quoted literal starts and ends with. The same shape test_cli.py uses, kept
    local because these tests read the art out of a subprocess's stdout rather
    than out of the module."""
    return [row for row in art.splitlines() if row.strip()]


def test_a_legacy_code_page_does_not_crash_the_banner():
    """chcp 437, in the form that is portable to a test.
    This used to be unable to assert which banner appeared: reconfigure()
    re-encodes a pipe whatever PYTHONIOENCODING asked for, so _enable_utf8
    succeeded and the cp437 declaration was overridden. A declared encoding is
    now read as a statement about the far end, so the fallback is reached from
    outside a monkeypatch and the test can say so.
    """
    code, out = cli([], {"PYTHONIOENCODING": "cp437:replace"})
    assert code == 0
    assert "NEXA" in out
    import nexa
    for row in _art_rows(nexa.BANNER_ASCII):
        assert row in out, "the ASCII wordmark did not reach a cp437 stdout"


def test_a_declared_ascii_stdout_gets_the_ascii_wordmark():
    """The CI job's case, and the one the fallback is named for.
    Strict `ascii` rather than `ascii:replace`: under :replace a glyph that
    escaped the UNICODE gate comes back as `?` and nothing fails, which is how
    a fallback rots while its job stays green.
    """
    code, out = cli([], {"PYTHONIOENCODING": "ascii"})
    assert code == 0
    assert out.isascii(), "a non-ASCII character survived a declared ASCII stdout"
    import nexa
    for row in _art_rows(nexa.BANNER_ASCII):
        assert row in out


def test_a_declared_utf8_still_gets_the_unicode_wordmark():
    """The common case, asserted so the narrowing above cannot swallow it."""
    import nexa
    code, out = cli([], {"PYTHONIOENCODING": "utf-8"})
    assert code == 0
    for row in _art_rows(nexa.BANNER_UNICODE):
        assert row in out


def test_an_undeclared_code_page_is_still_rescued():
    """The Windows console this was written for declares nothing.
    cp1252 as the machine's code page is not a statement about the far end, so
    it is still reconfigured to UTF-8. The check keys on PYTHONIOENCODING, not
    on the encoding stdout happens to report -- otherwise every Windows console
    would drop to the fallback and the rescue would be dead code.
    In-process because `cli` always names an encoding, and the situation here
    is the one where nobody has.
    """
    import nexa

    class MachineCodePage(object):
        """Reports cp1252 because that is the machine's code page, and agrees
        to be re-encoded, which is what a real console does."""
        encoding = "cp1252"
        reconfigured = False

        def reconfigure(self, **kwargs):
            MachineCodePage.reconfigured = True

    monkey = MachineCodePage()
    saved_out, saved_err = nexa.sys.stdout, nexa.sys.stderr
    declared = os.environ.pop("PYTHONIOENCODING", None)
    try:
        nexa.sys.stdout = nexa.sys.stderr = monkey
        assert nexa._enable_utf8() is True
        assert MachineCodePage.reconfigured, "the console was never rescued"
    finally:
        nexa.sys.stdout, nexa.sys.stderr = saved_out, saved_err
        if declared is not None:
            os.environ["PYTHONIOENCODING"] = declared


def test_a_bare_error_handler_declares_no_codec():
    """`:replace` is a handler and no encoding. Read as a codec it raises
    LookupError, which would send every such run to the fallback."""
    import nexa
    for value, expected in (
            (":replace", ""), ("", ""), ("utf-8", "utf-8"),
            ("utf-8:replace", "utf-8"), ("  ascii : strict ", "ascii")):
        os.environ["PYTHONIOENCODING"] = value
        try:
            assert nexa._declared_encoding() == expected, value
        finally:
            del os.environ["PYTHONIOENCODING"]


def test_ascii_banner_when_stdout_cannot_be_made_utf8(monkeypatch):
    """The other half: when reconfigure fails, the ASCII shapes are used.
    Forced rather than simulated with a code page, because the situation this
    guards is a stdout that REFUSES to become UTF-8, and every stdout reachable
    from a test happily agrees to.
    """
    import nexa
    assert nexa.BANNER_ASCII.strip(), "there has to be something to fall back to"
    for char in "█╗║╔╚═╝":
        assert char not in nexa.BANNER_ASCII
    class LegacyConsole(object):
        """A stdout on a code page that cannot encode the banner, and that
        refuses to be re-encoded. sys.stdout.encoding is read-only, so the
        stream is replaced rather than patched."""
        encoding = "cp437"

        def reconfigure(self, **kwargs):
            raise OSError("console is stuck on its code page")

        def write(self, text):
            return len(text)

        def flush(self):
            pass

    monkeypatch.setattr(nexa.sys, "stdout", LegacyConsole())
    assert nexa._enable_utf8() is False

    # And a stdout that agrees to become UTF-8 reports success, so the two
    # halves of the branch are both covered rather than one asserted twice.
    class ModernConsole(LegacyConsole):
        def reconfigure(self, **kwargs):
            return None

    monkeypatch.setattr(nexa.sys, "stdout", ModernConsole())
    assert nexa._enable_utf8() is True


def test_evidence_dir_end_to_end(tmp_path):
    """The variable as a user sets it: exported, then a command run.
    An empty directory is not a failure, so `nexa evidence` says so and
    exits 0 -- and it has to be reading the directory it was told about.
    """
    code, out = cli(["evidence"], {"NEXA_EVIDENCE_DIR": str(tmp_path)})
    assert code == 0
    assert "no evidence" in out.lower()


# --------------------------------------------------- NEXA_EMPIRICAL_EXEC

FENCE = "`" * 3


def _answer(code):
    return "{0}python\n{1}\n{0}\n".format(FENCE, code)


CLEAN = _answer("""from decimal import Decimal
total = Decimal("0.1") + Decimal("0.2")
assert total == Decimal("0.3")""")

RAISES = _answer(
    'values = [1, 2, 3]\n'
    'assert sum(values) == 7, "the totals do not agree"')

REACHES_OUT = _answer("""import os
print(os.listdir("."))""")

LOOPS = _answer("""while True:
    pass""")


def test_execution_is_off_by_default(monkeypatch):
    """The default has to be no. Running model-written code on a machine used
    for client work is a decision, not a checkbox."""
    import layers
    monkeypatch.delenv("NEXA_EMPIRICAL_EXEC", raising=False)
    result = layers.layer_empirical(CLEAN, "odoo")
    gaps = dict(result.not_checked)
    assert "execution" in gaps
    assert "off by default" in gaps["execution"]
    assert "isolated execution" not in result.checked


def test_a_clean_snippet_runs_when_asked(monkeypatch):
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    result = layers.layer_empirical(CLEAN, "odoo")
    assert "isolated execution" in result.checked
    assert [f for f in result.findings if "raised on execution" in f.message] == []


def test_a_failing_snippet_is_reported_as_a_finding(monkeypatch):
    """The point of running it at all: an assertion the model got wrong shows
    up as a finding rather than as a confident paragraph."""
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    result = layers.layer_empirical(RAISES, "odoo")
    raised = [f for f in result.findings if "raised on execution" in f.message]
    assert raised, [f.message for f in result.findings]
    assert raised[0].severity == "high"
    assert "totals do not agree" in raised[0].message


def test_anything_outside_the_allowlist_is_refused(monkeypatch):
    """os is not on the list, so the snippet is not run -- and the refusal is
    recorded as a gap rather than passed off as a clean execution."""
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    result = layers.layer_empirical(REACHES_OUT, "odoo")
    gaps = dict(result.not_checked)
    assert "execution" in gaps
    assert "outside the execution allowlist" in gaps["execution"]
    assert "isolated execution" not in result.checked


def test_an_alias_does_not_walk_past_the_allowlist(monkeypatch):
    """`import os as math` binds a name that IS on the allowlist.
    The check used to ask semantics.imported_names(), which answers "what does
    this module BIND" -- the right question for resolving an undefined name and
    the wrong one for an allowlist, which needs "what does this REACH FOR". The
    two disagree in both directions, and this is the direction that mattered.
    """
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    aliased = _answer("import os as math\n\nprint(math.getcwd())")
    gaps = dict(layers.layer_empirical(aliased, "odoo").not_checked)
    assert "outside the execution allowlist" in gaps.get("execution", "")
    assert "'os'" in gaps["execution"], "name the module it actually reaches for"


def test_a_from_import_of_an_allowed_module_is_allowed(monkeypatch):
    """The other direction of the same bug: `from decimal import Decimal` was
    refused although decimal is on the list, so the feature was far narrower
    than its own comment described."""
    import semantics
    tree, _ = semantics.parse_python("from decimal import Decimal")
    assert semantics.imported_modules(tree) == {"decimal"}
    assert semantics.imported_names(tree) == {"Decimal"}, \
        "imported_names keeps its own meaning; this is a second function"


def test_a_relative_import_can_never_match_the_allowlist():
    """`from . import x` resolves to a package the snippet is not part of."""
    import semantics
    tree, _ = semantics.parse_python("from . import helpers")
    assert semantics.imported_modules(tree) == {"."}


def test_an_odoo_answer_is_never_executed(monkeypatch):
    """Every Odoo answer imports odoo, so none of them is ever run. That is the
    correct outcome and not a limitation to engineer around: there is no useful
    sense in which an ORM method executes without an ORM."""
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    answer = _answer("from odoo import api, models\n\nx = 1")
    gaps = dict(layers.layer_empirical(answer, "odoo").not_checked)
    assert "outside the execution allowlist" in gaps.get("execution", "")


def test_a_runaway_snippet_hits_the_timeout(monkeypatch):
    """Five seconds, enforced by the parent. A generated infinite loop must not
    be able to hang the verifier that is checking it."""
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    result = layers.layer_empirical(LOOPS, "odoo")
    timed_out = [f for f in result.findings if "did not finish" in f.message]
    assert timed_out, [f.message for f in result.findings]


def test_the_child_is_isolated_and_runs_outside_the_repository(monkeypatch):
    """-I ignores PYTHONPATH and the user site directory, -B writes no
    __pycache__, and the working directory is not this checkout. None of that
    makes it a security boundary -- layers.py says so -- but all three are
    claims, and a claim nothing checks is a claim that drifts."""
    import subprocess
    import layers
    monkeypatch.setenv("NEXA_EMPIRICAL_EXEC", "1")
    seen = {}
    real_run = subprocess.run

    def record(command, **kwargs):
        seen["command"] = command
        seen["kwargs"] = kwargs
        return real_run(command, **kwargs)

    monkeypatch.setattr(layers.subprocess if hasattr(layers, "subprocess")
                        else subprocess, "run", record)
    layers.layer_empirical(CLEAN, "odoo")
    assert seen, "the snippet was never launched"
    assert "-I" in seen["command"]
    assert "-B" in seen["command"]
    assert seen["kwargs"].get("timeout") == 5
    assert os.path.normcase(seen["kwargs"].get("cwd", "")) != os.path.normcase(ROOT)


# ----------------------------------------- a setting written wrong is not fatal

def test_a_malformed_retention_setting_does_not_take_the_tool_down(with_env,
                                                                   tmp_path):
    """int() at module scope meant NEXA_EVIDENCE_KEEP=abc raised a ValueError
    out of `import evidence` -- and so out of every command that imports it, as
    a traceback. Every other switch in this file degrades quietly, and
    retention is no more important than any of them."""
    evidence = with_env("evidence", {"NEXA_EVIDENCE_DIR": str(tmp_path),
                                     "NEXA_EVIDENCE_KEEP": "abc"})
    assert evidence._KEEP == evidence.DEFAULT_KEEP


def test_zero_retention_keeps_the_bundle_it_just_wrote(with_env, tmp_path):
    """0 made write() prune what it had that moment written and still return
    the path, so a run printed an evidence id for a file already gone. Turning
    bundles off is what NEXA_EVIDENCE=0 is for."""
    evidence = with_env("evidence", {"NEXA_EVIDENCE": "1",
                                     "NEXA_EVIDENCE_DIR": str(tmp_path),
                                     "NEXA_EVIDENCE_KEEP": "0"})
    path = evidence.write(evidence.new_bundle("t-00", "odoo", "q", "m", {}, {}))
    assert path and os.path.exists(path), "it deleted the record of the run"