nexa / tests /test_failure.py

Claude Opus 5

ci: the first pipeline run found four things, all real

0190f14 about 2 months ago

12.3 kB

"""What happens when the things this tool depends on misbehave.
Everything here reaches one HTTP endpoint, one SQLite file and one directory of
JSON, and until this file existed every test assumed all three behaved. The open
items listed the cases worth injecting; these are them.
The Ollama cases run against a REAL socket rather than a patched function.
bin/endpoint.py resolves the server address per call from OLLAMA_HOST, so a test
can stand up a server that refuses, stalls, 404s or returns rubbish, point the
whole system at it, and watch what a user would see. A monkeypatched _post would
prove the error handling handles the exceptions the test invented; a socket
proves it handles the ones urllib actually raises.
The first run of this found the failure that mattered: with the server down,
`nexa ask` printed a 30-line traceback ending in a Windows error number.
"""
import io
import json
import os
import socket
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest

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


# ------------------------------------------------------------ a bad Ollama

class Misbehaving(BaseHTTPRequestHandler):
    """Serves whatever the test told it to. mode is set on the server."""

    protocol_version = "HTTP/1.1"

    def log_message(self, *args):
        pass                                # pytest output is not a web log

    def do_POST(self):
        mode = self.server.mode
        length = int(self.headers.get("Content-Length") or 0)
        self.rfile.read(length)

        if mode == "not-json":
            body = b"<html>a proxy login page, not Ollama</html>"
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
        elif mode == "no-such-model":
            body = json.dumps({"error": "model 'ai-odoo' not found"}).encode()
            self.send_response(404)
            self.send_header("Content-Type", "application/json")
        elif mode == "server-error":
            body = json.dumps({"error": "llama runner process has terminated"}).encode()
            self.send_response(500)
            self.send_header("Content-Type", "application/json")
        elif mode == "truncated-stream":
            # A stream that stops mid-answer: the runner was killed while
            # generating, which is what "model deleted mid-session" looks like
            # from this end.
            body = (json.dumps({"response": "partial", "done": False}).encode()
                    + b"\n" + b'{"response": "and then nothi')
            self.send_response(200)
            self.send_header("Content-Type", "application/x-ndjson")
        else:
            raise AssertionError("unknown mode " + mode)

        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


@pytest.fixture()
def bad_ollama(monkeypatch):
    """Start a server that misbehaves in a named way, and point Nexa at it."""
    started = []

    def start(mode):
        server = HTTPServer(("127.0.0.1", 0), Misbehaving)
        server.mode = mode
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        started.append((server, thread))
        monkeypatch.setenv("OLLAMA_HOST", "127.0.0.1:{0}".format(server.server_port))
        return server

    yield start
    for server, thread in started:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def closed_port():
    """A port with nothing on it. Bound and released, so it is genuinely free
    rather than merely unlikely."""
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port


# ------------------------------------------------------------------ the cases

def test_server_stopped_is_a_sentence_not_a_traceback(monkeypatch):
    import corpus
    import endpoint
    monkeypatch.setenv("OLLAMA_HOST", "127.0.0.1:{0}".format(closed_port()))
    with pytest.raises(endpoint.OllamaError) as caught:
        corpus.embed(["anything"])
    message = str(caught.value)
    assert "not responding" in message
    assert "OLLAMA_HOST" in message, "say where it looked, and why there"
    assert "Traceback" not in message
    assert "WinError" not in message and "Errno" not in message


def test_a_missing_model_names_the_model(bad_ollama, monkeypatch):
    """Model deleted mid-session. Ollama 404s, and 'HTTP Error 404' tells
    nobody which of the five models went missing."""
    import corpus
    import endpoint
    bad_ollama("no-such-model")
    with pytest.raises(endpoint.OllamaError) as caught:
        corpus._post("/api/generate", {"model": "ai-odoo", "prompt": "x"})
    message = str(caught.value)
    assert "ai-odoo" in message
    assert "nexa build" in message, "name the command that fixes it"


def test_a_non_json_reply_says_so(bad_ollama):
    """Something else listening on 11434 -- a proxy, a dev server, a captive
    portal. The bytes parse as neither JSON nor an error."""
    import corpus
    import endpoint
    bad_ollama("not-json")
    with pytest.raises(endpoint.OllamaError) as caught:
        corpus._post("/api/embed", {"model": "nomic-embed-text", "input": ["x"]})
    assert "not JSON" in str(caught.value)


def test_a_server_error_carries_ollamas_own_words(bad_ollama):
    """A 500 usually means the runner died. Ollama says so in the body, and
    repeating it beats inventing a summary."""
    import corpus
    import endpoint
    bad_ollama("server-error")
    with pytest.raises(endpoint.OllamaError) as caught:
        corpus._post("/api/generate", {"model": "ai-odoo", "prompt": "x"})
    message = str(caught.value)
    assert "500" in message
    assert "runner process has terminated" in message


def test_a_timeout_is_reported_as_a_timeout(monkeypatch):
    """The 900-second timeout, without waiting 900 seconds for it."""
    import endpoint
    explained = endpoint.explain(socket.timeout("timed out"))
    assert isinstance(explained, endpoint.OllamaError)
    message = str(explained)
    assert "timeout" in message
    assert "doctor" in message, "point at what would show a slow load"


def test_the_cli_exits_1_with_a_message_when_ollama_is_down(monkeypatch):
    """End to end, through the CLI, which is where the traceback appeared.
    Exit 1, not 2: the command was called correctly and could not finish.
    """
    env = dict(os.environ, OLLAMA_HOST="127.0.0.1:{0}".format(closed_port()),
               PYTHONIOENCODING="utf-8", NEXA_EVIDENCE="0", LOCALAI_TELEMETRY="0")
    proc = subprocess.run([sys.executable, os.path.join(ROOT, "nexa.py"),
                           "ask", "odoo", "anything at all"],
                          capture_output=True, text=True, encoding="utf-8",
                          errors="replace", cwd=ROOT, env=env)
    out = (proc.stdout or "") + (proc.stderr or "")
    assert proc.returncode == 1, out
    assert "Traceback" not in out, "this is the defect that motivated the file"

    # Which sentence appears depends on whether this machine has a built index.
    # With one, the run reaches Ollama and reports it unreachable. Without one
    # -- a fresh clone, or CI, where index.sqlite is gitignored and needs a
    # model to build -- retrieval stops first and says so. Both are sentences
    # naming a next step, which is the contract; the strict wording is pinned
    # by test_server_stopped_is_a_sentence_not_a_traceback, which needs no index.
    from fixtures.retrieval import indexed_rule_count
    if indexed_rule_count():
        assert "not responding" in out
    else:
        assert "No rules indexed" in out


def test_advise_also_fails_cleanly(monkeypatch):
    """rag.py is the one tool that does not go through corpus._post, so it
    needs its own conversion and its own guard."""
    env = dict(os.environ, OLLAMA_HOST="127.0.0.1:{0}".format(closed_port()),
               PYTHONIOENCODING="utf-8")
    proc = subprocess.run([sys.executable, os.path.join(ROOT, "nexa.py"),
                           "advise", "anything at all"],
                          capture_output=True, text=True, encoding="utf-8",
                          errors="replace", cwd=ROOT, env=env)
    out = (proc.stdout or "") + (proc.stderr or "")
    assert proc.returncode == 1, out
    assert "Traceback" not in out


# -------------------------------------------------------- the other two files

def test_a_corrupt_index_is_reported_not_crashed(monkeypatch, tmp_path):
    """bin/index.sqlite is rebuilt by `nexa build` and is not in the
    repository, so a truncated or half-written one is a normal thing to meet."""
    import corpus
    broken = tmp_path / "index.sqlite"
    broken.write_bytes(b"SQLite format 3\x00" + b"\xff" * 400)
    monkeypatch.setattr(corpus, "DB", str(broken))
    import sqlite3
    with pytest.raises(sqlite3.DatabaseError):
        corpus.retrieve("anything", "odoo")


def test_a_missing_corpus_is_empty_not_an_exception(monkeypatch, tmp_path):
    """load_rules returning [] is what every caller already checks for. An
    exception here would take down `nexa rules`, which has a message for the
    empty case and would never reach it."""
    import corpus
    monkeypatch.setattr(corpus, "RULES_DIR", str(tmp_path / "gone"))
    assert corpus.load_rules() == []


def test_an_unwritable_evidence_directory_does_not_lose_the_answer(monkeypatch,
                                                                   tmp_path):
    """Disk full, or a read-only directory. A bundle is a record OF the run;
    failing to write one must not fail the run, because the answer the user
    asked for has already been produced."""
    import importlib
    monkeypatch.setenv("NEXA_EVIDENCE", "1")
    monkeypatch.setenv("NEXA_EVIDENCE_DIR", str(tmp_path / "evidence"))
    evidence = importlib.reload(importlib.import_module("evidence"))

    # Built before the disk fills: new_bundle reads the corpus in order to hash
    # it, and the case under test is the WRITE failing, not the read.
    bundle = evidence.new_bundle("t-disk-full", "odoo", "q", "m", {}, {})

    def full(*args, **kwargs):
        raise OSError(28, "No space left on device")

    monkeypatch.setattr(evidence.io, "open", full)
    evidence.write(bundle)              # must not raise
    importlib.reload(evidence)


def _refuse(*args, **kwargs):
    raise OSError(13, "Permission denied")


def test_an_unreadable_corpus_leaves_the_hash_unknown(monkeypatch):
    """The same principle one level down. corpus_sha256 is called while a
    bundle is being built, so an OSError there propagated out of new_bundle and
    ended a run that had already produced its answer."""
    import evidence
    monkeypatch.setattr(evidence.io, "open", _refuse)
    assert evidence.corpus_sha256() is None


def test_telemetry_survives_an_unwritable_file(monkeypatch, tmp_path):
    """Same contract, other subsystem: 'telemetry must not break the tool' is
    written in telemetry.py and had never been tested."""
    import importlib
    monkeypatch.setenv("LOCALAI_TELEMETRY", "1")
    monkeypatch.setenv("LOCALAI_TELEMETRY_FILE", str(tmp_path / "nope" / "t.jsonl"))
    telemetry = importlib.reload(importlib.import_module("telemetry"))
    telemetry.emit(telemetry.EVENT_RUN, {"outcome": "verified_fast"})   # no raise
    importlib.reload(telemetry)


def test_a_partially_written_telemetry_line_is_skipped(tmp_path, monkeypatch):
    """Two processes appending, one killed mid-line. report.load already
    tolerates it; this pins that it still does."""
    import importlib
    path = tmp_path / "telemetry.jsonl"
    path.write_text(
        json.dumps({"event": "localai.run", "outcome": "verified_fast"}) + "\n"
        + '{"event": "localai.run", "outcome": "verified_esc',
        encoding="utf-8")
    monkeypatch.setenv("LOCALAI_TELEMETRY", "1")
    monkeypatch.setenv("LOCALAI_TELEMETRY_FILE", str(path))
    importlib.reload(importlib.import_module("telemetry"))
    report = importlib.reload(importlib.import_module("report"))
    events = report.load()
    assert len(events) == 1, "the torn line should be skipped, not fatal"
    importlib.reload(importlib.import_module("telemetry"))
    importlib.reload(report)

	"""What happens when the things this tool depends on misbehave.

	Everything here reaches one HTTP endpoint, one SQLite file and one directory of
	JSON, and until this file existed every test assumed all three behaved. The open
	items listed the cases worth injecting; these are them.

	The Ollama cases run against a REAL socket rather than a patched function.
	bin/endpoint.py resolves the server address per call from OLLAMA_HOST, so a test
	can stand up a server that refuses, stalls, 404s or returns rubbish, point the
	whole system at it, and watch what a user would see. A monkeypatched _post would
	prove the error handling handles the exceptions the test invented; a socket
	proves it handles the ones urllib actually raises.

	The first run of this found the failure that mattered: with the server down,
	`nexa ask` printed a 30-line traceback ending in a Windows error number.
	"""
	import io
	import json
	import os
	import socket
	import subprocess
	import sys
	import threading
	from http.server import BaseHTTPRequestHandler, HTTPServer

	import pytest

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


	# ------------------------------------------------------------ a bad Ollama

	class Misbehaving(BaseHTTPRequestHandler):
	"""Serves whatever the test told it to. mode is set on the server."""

	protocol_version = "HTTP/1.1"

	def log_message(self, *args):
	pass # pytest output is not a web log

	def do_POST(self):
	mode = self.server.mode
	length = int(self.headers.get("Content-Length") or 0)
	self.rfile.read(length)

	if mode == "not-json":
	body = b"<html>a proxy login page, not Ollama</html>"
	self.send_response(200)
	self.send_header("Content-Type", "text/html")
	elif mode == "no-such-model":
	body = json.dumps({"error": "model 'ai-odoo' not found"}).encode()
	self.send_response(404)
	self.send_header("Content-Type", "application/json")
	elif mode == "server-error":
	body = json.dumps({"error": "llama runner process has terminated"}).encode()
	self.send_response(500)
	self.send_header("Content-Type", "application/json")
	elif mode == "truncated-stream":
	# A stream that stops mid-answer: the runner was killed while
	# generating, which is what "model deleted mid-session" looks like
	# from this end.
	body = (json.dumps({"response": "partial", "done": False}).encode()
	+ b"\n" + b'{"response": "and then nothi')
	self.send_response(200)
	self.send_header("Content-Type", "application/x-ndjson")
	else:
	raise AssertionError("unknown mode " + mode)

	self.send_header("Content-Length", str(len(body)))
	self.end_headers()
	self.wfile.write(body)


	@pytest.fixture()
	def bad_ollama(monkeypatch):
	"""Start a server that misbehaves in a named way, and point Nexa at it."""
	started = []

	def start(mode):
	server = HTTPServer(("127.0.0.1", 0), Misbehaving)
	server.mode = mode
	thread = threading.Thread(target=server.serve_forever, daemon=True)
	thread.start()
	started.append((server, thread))
	monkeypatch.setenv("OLLAMA_HOST", "127.0.0.1:{0}".format(server.server_port))
	return server

	yield start
	for server, thread in started:
	server.shutdown()
	server.server_close()
	thread.join(timeout=5)


	def closed_port():
	"""A port with nothing on it. Bound and released, so it is genuinely free
	rather than merely unlikely."""
	sock = socket.socket()
	sock.bind(("127.0.0.1", 0))
	port = sock.getsockname()[1]
	sock.close()
	return port


	# ------------------------------------------------------------------ the cases

	def test_server_stopped_is_a_sentence_not_a_traceback(monkeypatch):
	import corpus
	import endpoint
	monkeypatch.setenv("OLLAMA_HOST", "127.0.0.1:{0}".format(closed_port()))
	with pytest.raises(endpoint.OllamaError) as caught:
	corpus.embed(["anything"])
	message = str(caught.value)
	assert "not responding" in message
	assert "OLLAMA_HOST" in message, "say where it looked, and why there"
	assert "Traceback" not in message
	assert "WinError" not in message and "Errno" not in message


	def test_a_missing_model_names_the_model(bad_ollama, monkeypatch):
	"""Model deleted mid-session. Ollama 404s, and 'HTTP Error 404' tells
	nobody which of the five models went missing."""
	import corpus
	import endpoint
	bad_ollama("no-such-model")
	with pytest.raises(endpoint.OllamaError) as caught:
	corpus._post("/api/generate", {"model": "ai-odoo", "prompt": "x"})
	message = str(caught.value)
	assert "ai-odoo" in message
	assert "nexa build" in message, "name the command that fixes it"


	def test_a_non_json_reply_says_so(bad_ollama):
	"""Something else listening on 11434 -- a proxy, a dev server, a captive
	portal. The bytes parse as neither JSON nor an error."""
	import corpus
	import endpoint
	bad_ollama("not-json")
	with pytest.raises(endpoint.OllamaError) as caught:
	corpus._post("/api/embed", {"model": "nomic-embed-text", "input": ["x"]})
	assert "not JSON" in str(caught.value)


	def test_a_server_error_carries_ollamas_own_words(bad_ollama):
	"""A 500 usually means the runner died. Ollama says so in the body, and
	repeating it beats inventing a summary."""
	import corpus
	import endpoint
	bad_ollama("server-error")
	with pytest.raises(endpoint.OllamaError) as caught:
	corpus._post("/api/generate", {"model": "ai-odoo", "prompt": "x"})
	message = str(caught.value)
	assert "500" in message
	assert "runner process has terminated" in message


	def test_a_timeout_is_reported_as_a_timeout(monkeypatch):
	"""The 900-second timeout, without waiting 900 seconds for it."""
	import endpoint
	explained = endpoint.explain(socket.timeout("timed out"))
	assert isinstance(explained, endpoint.OllamaError)
	message = str(explained)
	assert "timeout" in message
	assert "doctor" in message, "point at what would show a slow load"


	def test_the_cli_exits_1_with_a_message_when_ollama_is_down(monkeypatch):
	"""End to end, through the CLI, which is where the traceback appeared.

	Exit 1, not 2: the command was called correctly and could not finish.
	"""
	env = dict(os.environ, OLLAMA_HOST="127.0.0.1:{0}".format(closed_port()),
	PYTHONIOENCODING="utf-8", NEXA_EVIDENCE="0", LOCALAI_TELEMETRY="0")
	proc = subprocess.run([sys.executable, os.path.join(ROOT, "nexa.py"),
	"ask", "odoo", "anything at all"],
	capture_output=True, text=True, encoding="utf-8",
	errors="replace", cwd=ROOT, env=env)
	out = (proc.stdout or "") + (proc.stderr or "")
	assert proc.returncode == 1, out
	assert "Traceback" not in out, "this is the defect that motivated the file"

	# Which sentence appears depends on whether this machine has a built index.
	# With one, the run reaches Ollama and reports it unreachable. Without one
	# -- a fresh clone, or CI, where index.sqlite is gitignored and needs a
	# model to build -- retrieval stops first and says so. Both are sentences
	# naming a next step, which is the contract; the strict wording is pinned
	# by test_server_stopped_is_a_sentence_not_a_traceback, which needs no index.
	from fixtures.retrieval import indexed_rule_count
	if indexed_rule_count():
	assert "not responding" in out
	else:
	assert "No rules indexed" in out


	def test_advise_also_fails_cleanly(monkeypatch):
	"""rag.py is the one tool that does not go through corpus._post, so it
	needs its own conversion and its own guard."""
	env = dict(os.environ, OLLAMA_HOST="127.0.0.1:{0}".format(closed_port()),
	PYTHONIOENCODING="utf-8")
	proc = subprocess.run([sys.executable, os.path.join(ROOT, "nexa.py"),
	"advise", "anything at all"],
	capture_output=True, text=True, encoding="utf-8",
	errors="replace", cwd=ROOT, env=env)
	out = (proc.stdout or "") + (proc.stderr or "")
	assert proc.returncode == 1, out
	assert "Traceback" not in out


	# -------------------------------------------------------- the other two files

	def test_a_corrupt_index_is_reported_not_crashed(monkeypatch, tmp_path):
	"""bin/index.sqlite is rebuilt by `nexa build` and is not in the
	repository, so a truncated or half-written one is a normal thing to meet."""
	import corpus
	broken = tmp_path / "index.sqlite"
	broken.write_bytes(b"SQLite format 3\x00" + b"\xff" * 400)
	monkeypatch.setattr(corpus, "DB", str(broken))
	import sqlite3
	with pytest.raises(sqlite3.DatabaseError):
	corpus.retrieve("anything", "odoo")


	def test_a_missing_corpus_is_empty_not_an_exception(monkeypatch, tmp_path):
	"""load_rules returning [] is what every caller already checks for. An
	exception here would take down `nexa rules`, which has a message for the
	empty case and would never reach it."""
	import corpus
	monkeypatch.setattr(corpus, "RULES_DIR", str(tmp_path / "gone"))
	assert corpus.load_rules() == []


	def test_an_unwritable_evidence_directory_does_not_lose_the_answer(monkeypatch,
	tmp_path):
	"""Disk full, or a read-only directory. A bundle is a record OF the run;
	failing to write one must not fail the run, because the answer the user
	asked for has already been produced."""
	import importlib
	monkeypatch.setenv("NEXA_EVIDENCE", "1")
	monkeypatch.setenv("NEXA_EVIDENCE_DIR", str(tmp_path / "evidence"))
	evidence = importlib.reload(importlib.import_module("evidence"))

	# Built before the disk fills: new_bundle reads the corpus in order to hash
	# it, and the case under test is the WRITE failing, not the read.
	bundle = evidence.new_bundle("t-disk-full", "odoo", "q", "m", {}, {})

	def full(args, *kwargs):
	raise OSError(28, "No space left on device")

	monkeypatch.setattr(evidence.io, "open", full)
	evidence.write(bundle) # must not raise
	importlib.reload(evidence)


	def _refuse(args, *kwargs):
	raise OSError(13, "Permission denied")


	def test_an_unreadable_corpus_leaves_the_hash_unknown(monkeypatch):
	"""The same principle one level down. corpus_sha256 is called while a
	bundle is being built, so an OSError there propagated out of new_bundle and
	ended a run that had already produced its answer."""
	import evidence
	monkeypatch.setattr(evidence.io, "open", _refuse)
	assert evidence.corpus_sha256() is None


	def test_telemetry_survives_an_unwritable_file(monkeypatch, tmp_path):
	"""Same contract, other subsystem: 'telemetry must not break the tool' is
	written in telemetry.py and had never been tested."""
	import importlib
	monkeypatch.setenv("LOCALAI_TELEMETRY", "1")
	monkeypatch.setenv("LOCALAI_TELEMETRY_FILE", str(tmp_path / "nope" / "t.jsonl"))
	telemetry = importlib.reload(importlib.import_module("telemetry"))
	telemetry.emit(telemetry.EVENT_RUN, {"outcome": "verified_fast"}) # no raise
	importlib.reload(telemetry)


	def test_a_partially_written_telemetry_line_is_skipped(tmp_path, monkeypatch):
	"""Two processes appending, one killed mid-line. report.load already
	tolerates it; this pins that it still does."""
	import importlib
	path = tmp_path / "telemetry.jsonl"
	path.write_text(
	json.dumps({"event": "localai.run", "outcome": "verified_fast"}) + "\n"
	+ '{"event": "localai.run", "outcome": "verified_esc',
	encoding="utf-8")
	monkeypatch.setenv("LOCALAI_TELEMETRY", "1")
	monkeypatch.setenv("LOCALAI_TELEMETRY_FILE", str(path))
	importlib.reload(importlib.import_module("telemetry"))
	report = importlib.reload(importlib.import_module("report"))
	events = report.load()
	assert len(events) == 1, "the torn line should be skipped, not fatal"
	importlib.reload(importlib.import_module("telemetry"))
	importlib.reload(report)