| """Shared setup for the Nexa test suite. |
| |
| Nexa is stdlib-only and runs as scripts, not as an installed package, so the |
| repository root and bin/ both go on sys.path exactly as the CLI puts them there. |
| Importing nexa also reconfigures stdout, which is harmless under pytest and is |
| the same thing that happens in a real run. |
| """ |
| import os |
| import sys |
|
|
| import pytest |
|
|
| TESTS = os.path.dirname(os.path.abspath(__file__)) |
| ROOT = os.path.dirname(TESTS) |
| BIN = os.path.join(ROOT, "bin") |
| for _path in (ROOT, BIN, TESTS): |
| if _path not in sys.path: |
| sys.path.insert(0, _path) |
|
|
|
|
| @pytest.fixture(scope="session") |
| def root(): |
| return ROOT |
|
|
|
|
| @pytest.fixture(scope="session") |
| def all_rules(): |
| import corpus |
| rules = corpus.load_rules() |
| if not rules: |
| pytest.skip("corpus is empty; run `nexa build` first") |
| return rules |
|
|
|
|
| @pytest.fixture(scope="session") |
| def odoo_rules(all_rules): |
| return [r for r in all_rules if r["stack"] == "odoo"] |
|
|
|
|
| @pytest.fixture() |
| def bundle(odoo_rules): |
| """A complete evidence bundle, retrieval included. |
| |
| Passing None, or an bundle with no recorded retrieval, makes every verdict |
| in every test depend on a provenance finding that has nothing to do with |
| what is under test. Worth noting that the provenance layer caught exactly |
| that when this fixture was first written and shipped incomplete -- which is |
| the layer working, so the fixture is what changed. |
| """ |
| import evidence |
| b = evidence.new_bundle("test-000000000000", "odoo", "q", "test-model", |
| {"temperature": 0}, {"layers": "1.0.0", |
| "semantics": "1.0.0"}) |
| hits = [(0.9, r["id"], r["severity"], r["rule"]) for r in odoo_rules[:4]] |
| evidence.record_retrieval(b, hits, {odoo_rules[0]["id"]}) |
| return b |
|
|
|
|
| class FakeOllama(object): |
| """Stands in for corpus._post: replays canned answers, records the calls. |
| |
| The whole system reaches Ollama through that one function, so replacing it |
| makes the nine layers and the escalation topology testable in milliseconds |
| with no GPU, no model, and no network -- and, more to the point, |
| deterministically. A pipeline that escalates only when a 7B happens to write |
| something wrong cannot be tested against a live model at all. |
| """ |
|
|
| def __init__(self, answers, embedding=None): |
| self.answers = list(answers) |
| self.embedding = embedding or [0.1] * 768 |
| self.calls = [] |
|
|
| def __call__(self, path, payload, stream=False): |
| self.calls.append((path, payload.get("model"), payload.get("options"))) |
| if path == "/api/embed": |
| count = len(payload.get("input") or [""]) |
| return {"embeddings": [list(self.embedding) for _ in range(count)]} |
| if path == "/api/generate": |
| text = self.answers.pop(0) if self.answers else "" |
| return self._stream(text) |
| raise AssertionError("unexpected Ollama path in a test: " + path) |
|
|
| @staticmethod |
| def _stream(text): |
| """Byte lines, the way urlopen hands them back, ending with a done frame |
| carrying the counters report perf reads.""" |
| import json |
| yield json.dumps({"response": text, "done": False}).encode("utf-8") |
| yield json.dumps({ |
| "response": "", "done": True, |
| "eval_count": 160, "eval_duration": 4_000_000_000, |
| "prompt_eval_count": 900, "load_duration": 0, |
| }).encode("utf-8") |
|
|
| @property |
| def generate_calls(self): |
| return [c for c in self.calls if c[0] == "/api/generate"] |
|
|
| @property |
| def models_used(self): |
| return [c[1] for c in self.generate_calls] |
|
|
|
|
| @pytest.fixture() |
| def fake_ollama(monkeypatch): |
| """Install a FakeOllama over corpus._post. Returns the installer.""" |
| import corpus |
|
|
| def install(answers, embedding=None): |
| fake = FakeOllama(answers, embedding) |
| monkeypatch.setattr(corpus, "_post", fake) |
| return fake |
|
|
| return install |
|
|
|
|
| @pytest.fixture() |
| def seeded_index(monkeypatch, tmp_path): |
| """A retrieval index built from the committed corpus, with flat vectors. |
| |
| The real bin/index.sqlite is gitignored and rebuilt by `nexa build`, which |
| needs a model. A fresh clone and CI therefore have none, and anything that |
| retrieves gets an empty list -- which is how twelve pipeline tests came to |
| depend on whoever ran them having built one first. |
| |
| The rule JSON is committed, so the index can be rebuilt from it here. Every |
| vector is identical, so similarity contributes nothing and retrieval orders |
| by the severity boost alone. That is enough to exercise routing, which is |
| what those tests assert. It is deliberately NOT enough to say anything |
| about ranking: `nexa eval --retrieval` measures that, against a real index. |
| """ |
| import json |
| import corpus |
| index = tmp_path / "index.sqlite" |
| monkeypatch.setattr(corpus, "DB", str(index)) |
| flat = json.dumps([0.1] * 768) |
| db = corpus.connect() |
| for rule in corpus.load_rules(): |
| db.execute("INSERT OR REPLACE INTO rules VALUES (?,?,?,?,?)", |
| (rule["id"], rule["stack"], rule["severity"], |
| corpus.context_document(rule), flat)) |
| db.commit() |
| db.close() |
| return index |
|
|
|
|
| @pytest.fixture() |
| def real_embedding(): |
| """One genuine vector lifted from the built index. |
| |
| Retrieval scores a query against stored rule vectors; a made-up vector still |
| returns rules, but a real one keeps the ordering meaningful when a test |
| cares which rule came back. |
| """ |
| import json |
| import sqlite3 |
| import corpus |
| from fixtures.retrieval import indexed_rule_count |
| if not indexed_rule_count(): |
| pytest.skip("no retrieval index; run `nexa build` first") |
| db = sqlite3.connect(corpus.DB) |
| row = db.execute("SELECT vec FROM rules LIMIT 1").fetchone() |
| db.close() |
| if not row: |
| pytest.skip("retrieval index has no rules; run `nexa build` first") |
| return json.loads(row[0]) |
|
|
