"""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"