diff --git a/ops/ax/README.md b/ops/ax/README.md index 2ae9e8c..6d65c9f 100644 --- a/ops/ax/README.md +++ b/ops/ax/README.md @@ -154,16 +154,29 @@ writes objects and their tags, and it gets no `Put*` on versioning or lifecycle: "s3:GetLifecycleConfiguration"], "Resource": ["arn:aws:s3:::sast-memory"]}, {"Effect": "Allow", "Action": ["s3:GetObject", "s3:GetObjectVersion", "s3:PutObject", "s3:DeleteObject", - "s3:GetObjectTagging", "s3:PutObjectTagging"], + "s3:GetObjectTagging", "s3:PutObjectTagging", "s3:GetObjectVersionTagging"], "Resource": ["arn:aws:s3:::sast-memory/*"]}]} Measured state on 2026-10-02 (agent `sast-memory-agent`): versioning was `Enabled` and a 30-day rule on `runs/` -was in place. `verify_bucket` still refused, because `GetObjectTagging` returned AccessDenied for the agent. +was in place. `verify_bucket` first refused because `GetObjectTagging` returned AccessDenied for the agent; with +the three tagging permissions above (`s3:GetObjectTagging`, `s3:PutObjectTagging`, `s3:GetObjectVersionTagging`) +the real-server contract tests pass. -Known RustFS limit (measured 2026-10-02): Select infers an object's JSON schema from its leading rows. A -`where` field that is missing there, even if row 5000 has it, fails with `EvaluatorBindingDoesNotExist` instead -of matching. The store raises a `MemoryStoreError` that names the field. It does not answer "no rows", since -that answer could silently drop matches. +**Fixed schema: every row carries every queryable field.** RustFS's Select infers an object's JSON schema from its +first 1000 rows (measured 2026-10-02). Two failures follow: + +- a `where` field missing from those rows, even if row 5000 has it, fails with `EvaluatorBindingDoesNotExist`; +- a column that is `null` in all of those rows and set in a later one fails the whole object with + `JSONParsingError`, whatever the `where` names (999 leading nulls answer, 1000 fail). A column holding two JSON + types (a string, then a number) fails the same way at any size. + +So `plane/memory.py` declares per kind the fields a `where` may name (`QUERY_FIELDS`, besides the fields every +row has). They are strings, and the store writes all of them on every row at write time, `""` where one does not +apply (never `null`). `rows()` refuses an unknown kind, a `where` on a field no kind declares, and `where` = null, +before any read: a new query declares its field first. Rows written before the rule are rewritten once with +`ousast plane memory-normalise [--store URL] [--dry-run]`, which is idempotent and prints its counts; on a bucket +it writes new object versions, and versioning keeps the old ones. Run it when a store moves to RustFS. The store +never answers "no rows" for a failed Select; it raises a `MemoryStoreError` naming the object and the cause. The real-server contract tests (`OUSAST_MEMORY_TEST_MINIO=1`, bucket `OUSAST_MEMORY_TEST_BUCKET`, else `MINIO_BUCKET`) verify the bucket at its root and write only under a fresh `contract-/` prefix. They never diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index f1fad28..c111556 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -287,6 +287,11 @@ def _main(argv: list[str] | None) -> int: plane_remember.add_argument("--plane", type=Path, default=Path("plane"), help="where runs/.yaml names population and split") plane_remember.add_argument("--population", help="the population the run measured (default: the Run's annotation)") plane_remember.add_argument("--split", help="the split of that population (default: the Run's annotation)") + plane_normalise = plane_sub.add_parser( + "memory-normalise", help='rewrite the store\'s rows to the fixed schema (every queryable field, "" where absent); idempotent' + ) + plane_normalise.add_argument("--store", help="the store URL (default OUSAST_MEMORY): file:/// or minio://[/]") + plane_normalise.add_argument("--dry-run", action="store_true", help="count what would be rewritten, write nothing") plane_engine = plane_sub.add_parser( "alerts-engine", help="a Run's `alerts` for PHP and languages quick mode does not cover, from the engine image on this host" ) @@ -1384,6 +1389,8 @@ def _plane(args: argparse.Namespace) -> int: if args.plane_command == "remember": manifest = args.plane / "runs" / f"{args.run}.yaml" return _plane_memory("ingest", manifest if manifest.is_file() else None, run=args.run, population=args.population, split=args.split) + if args.plane_command == "memory-normalise": + return _plane_memory_normalise(args) if args.plane_command == "status": print(reconciler.status(args.run, units=args.units)) return 0 @@ -1442,6 +1449,25 @@ def _plane_alerts_engine(args: argparse.Namespace) -> int: return 0 +def _plane_memory_normalise(args: argparse.Namespace) -> int: + """``ousast plane memory-normalise``: one pass over every row object; prints the counts.""" + from .plane import memory + + try: + store = memory.open_store(args.store) + got = store.normalise(dry_run=args.dry_run) + except (memory.MemoryStoreError, OSError, ValueError) as exc: + print(f"memory-normalise failed: {exc}", file=sys.stderr) + return 1 + verb = "would rewrite" if args.dry_run else "rewrote" + print( + f"memory {store.describe()}: {verb} {got['objects_rewritten']}/{got['objects']} objects; " + f'{got["rows_changed"]}/{got["rows"]} rows filled {got["fields_filled"]} absent or null fields with "" ' + f"({', '.join(memory.QUERYABLE)})" + ) + return 0 + + def _plane_memory( step: str, run_manifest: Path | None, *, run: str | None = None, population: str | None = None, split: str | None = None ) -> int: diff --git a/src/openultrasast/plane/memory.py b/src/openultrasast/plane/memory.py index ac7dc6b..809c932 100644 --- a/src/openultrasast/plane/memory.py +++ b/src/openultrasast/plane/memory.py @@ -18,6 +18,11 @@ credentials come from ``.env`` (``MINIO_ENDPOINT``, ``MINIO_ACCESS_KEY``, ``MINI enables versioning and the ``runs/`` expiry rule once, and opening a ``MinioStore`` verifies them and S3 Select (:meth:`MinioStore.verify_bucket`), refusing to start otherwise. +Every row carries every queryable field (``QUERY_FIELDS``: strings, ``""`` where one does not apply, never ``null``): +the store fills them in on every write, ``rows`` refuses a ``where`` field no kind declared, and ``ousast plane +memory-normalise`` rewrites rows written before the rule. RustFS's S3 Select infers the schema from an object's first +1000 rows only. + Ingest is idempotent: an ``index.jsonl`` hit on (run, task, sha256) skips the delivery, and rows carry a deterministic ``id``, so a repeated row replaces itself. Every object is written whole (``FileStore``: a temporary file renamed) and the index last, so an interrupted ingest leaves nothing partial that a repeat would not repair. @@ -68,6 +73,33 @@ _BLOB_PREFIX = re.compile(r"[a-z]+(?:/[A-Za-z0-9._-]+)?") _BLOB_NAME = re.compile(r"[0-9a-f]{64}") BLOB_SUFFIX = {"excerpts": ".txt", "labels": ".jsonl"} # every other prefix holds JSON ROW_FIELDS = ("id", "kind", "repo", "pin", "run", "task", "population", "split", "image") +# The fixed schema: every row carries every queryable field. RustFS's S3 Select infers an object's schema from its +# first 1000 rows: a ``where`` field absent there raises ``EvaluatorBindingDoesNotExist`` although later rows carry it +# (measured 2026-10-02: field only at row 5000, and absent everywhere), and a column that is ``null`` in all of them +# but set later fails the whole object with ``JSONParsingError``, whatever the ``where`` (measured 2026-10-02: 999 +# leading nulls answer, 1000 fail; "" placeholders answer at 5000). So a queryable field is a string, and a row it +# does not apply to carries the empty string, never ``null``. Per kind, the fields a server-side filter +# (``rows(..., where=...)``) may name besides ``ROW_FIELDS`` (which every row carries already); a new query declares +# its field here first. One object holds every kind of a repository + pin, so the store writes the union of these +# fields (``QUERYABLE``) on every row (``normalise_row``). +QUERY_FIELDS: Mapping[str, tuple[str, ...]] = { + "facts": ("candidates_digest",), # seed(): candidates_digest + image + "verdict": ("candidate", "final"), + "unit_cost": (), + "alert": (), + "coverage": (), + "proposal_outcome": (), + "features": (), # benchmarks/learn/build_features.py: run + task + "label": (), + "decision": (), + "experiment": (), + "arm_outcome": (), + "experiment_result": (), + "example": ("profile",), # learn.examples.load_examples +} +if tuple(QUERY_FIELDS) != KINDS: # pragma: no cover -- a kind added without its queryable fields + raise RuntimeError("QUERY_FIELDS must declare every kind of KINDS, in order") +QUERYABLE = tuple(sorted({f for fields in QUERY_FIELDS.values() for f in fields} - set(ROW_FIELDS))) TAG_FIELDS = ("repo", "pin", "kind", "family", "run", "population", "split") MIN_FREE_BYTES = 1 << 30 MINIO_KEYS = ("MINIO_ENDPOINT", "MINIO_ACCESS_KEY", "MINIO_SECRET_KEY") @@ -145,6 +177,37 @@ def validate_row(row: object, where: str = "row") -> dict[str, Any]: return row +def normalise_row(row: Mapping[str, Any], where: str = "row") -> dict[str, Any]: + """``row`` with every ``QUERYABLE`` field present: absent or ``null`` becomes ``""`` (see ``QUERY_FIELDS``); + a string is kept; any other type is refused, since one column of two types fails the server's Select.""" + out = dict(row) + for name in QUERYABLE: + value = out.get(name) + if value is None: + out[name] = "" + elif not isinstance(value, str): + raise MemoryStoreError(f"{where}: queryable field {name!r} must be a string, got {type(value).__name__}") + return out + + +def check_query(kind: str | None, where: Mapping[str, Any] | None) -> None: + """Refuse a filter the fixed schema does not cover: an unknown kind, or a ``where`` field that ``kind`` (any kind + when None) does not declare in ``QUERY_FIELDS``; the server could not bind it on every object.""" + if kind is not None and kind not in QUERY_FIELDS: + raise MemoryStoreError(f"rows: unknown kind {kind!r} (one of {', '.join(KINDS)})") + declared = set(ROW_FIELDS) | set(QUERY_FIELDS[kind] if kind is not None else QUERYABLE) + undeclared = sorted(set(where or {}) - declared) + if undeclared: + scope = f"kind {kind!r}" if kind is not None else "any kind" + raise MemoryStoreError( + f"rows: where field(s) {undeclared} are not queryable for {scope}: declare them in memory.QUERY_FIELDS " + '(every row then carries them, "" where they do not apply) and run `ousast plane memory-normalise`' + ) + nulls = sorted(name for name, value in (where or {}).items() if value is None) + if nulls: + raise MemoryStoreError(f'rows: where {nulls} = null never matches: a queryable field that does not apply is ""') + + def _line(row: Mapping[str, Any]) -> str: return json.dumps(row, sort_keys=True, separators=(",", ":")) @@ -272,7 +335,9 @@ class MemoryStore(ABC): def rows( self, repo: str | None = None, pin: str | None = None, kind: str | None = None, where: Mapping[str, Any] | None = None ) -> list[Record]: - """Rows filtered by repository, pin, kind and top-level equality on ``where``, in key then id order.""" + """Rows filtered by repository, pin, kind and top-level equality on ``where``, in key then id order. An unknown + kind or a ``where`` field the kind does not declare in ``QUERY_FIELDS`` is refused (:func:`check_query`).""" + check_query(kind, where) prefix = f"repos/{repo_dir(repo)}/" if repo else "repos/" keys = [k for k in self._keys(prefix) if k.endswith(".jsonl") and (pin is None or k.endswith(f"/{pin}.jsonl"))] wanted = {**(where or {}), **({"kind": kind} if kind else {})} @@ -318,13 +383,13 @@ class MemoryStore(ABC): """Merge rows into their repository + pin objects by ``id`` (a repeated id replaces the stored row).""" groups: dict[str, dict[str, dict[str, Any]]] = {} for index, row in enumerate(rows): - checked = validate_row(dict(row), f"row {index}") + checked = normalise_row(validate_row(dict(row), f"row {index}"), f"row {index}") groups.setdefault(f"repos/{repo_dir(checked['repo'])}/{checked['pin']}.jsonl", {})[checked["id"]] = checked if groups: self._check_space() for key, new in sorted(groups.items()): found = self._get(key) - merged = {str(r["id"]): r for r in (_parse_lines(found[0], key) if found else [])} + merged = {str(r["id"]): normalise_row(r) for r in (_parse_lines(found[0], key) if found else [])} merged.update(new) ordered = [merged[i] for i in sorted(merged)] self._put(key, "".join(_line(r) + "\n" for r in ordered).encode("utf-8"), _labels(ordered)) @@ -338,7 +403,7 @@ class MemoryStore(ABC): if found is None or not unwanted: return 0 rows = _parse_lines(found[0], key) - kept = [r for r in rows if str(r["id"]) not in unwanted] + kept = [normalise_row(r) for r in rows if str(r["id"]) not in unwanted] if len(kept) == len(rows): return 0 if kept: @@ -365,6 +430,31 @@ class MemoryStore(ABC): self._put("index.jsonl", "".join(_line(e) + "\n" for e in [*index, entry]).encode("utf-8")) return IngestResult(run, task, len(checked), False, kinds) + def normalise(self, *, dry_run: bool = False) -> dict[str, int]: + """Rewrite every row object to the fixed schema (every ``QUERYABLE`` field a string, ``""`` where it was + absent or ``null``), once, before a store moves to a server that infers its schema (RustFS). Idempotent: an + object already in the schema is not rewritten. On a versioned bucket a rewrite is a new object version; the + old stays citable. A queryable field of another type fails naming its object and row. Returns counts: + objects seen and rewritten, rows seen and changed, fields filled.""" + out = {"objects": 0, "objects_rewritten": 0, "rows": 0, "rows_changed": 0, "fields_filled": 0} + for key in sorted(k for k in self._keys("repos/") if k.endswith(".jsonl")): + found = self._get(key) + if found is None: + continue + rows = _parse_lines(found[0], key) + fixed = [normalise_row(r, f"{key}:{n}") for n, r in enumerate(rows, 1)] + filled = [sum(r.get(f) is None for f in QUERYABLE) for r in rows] + out["objects"] += 1 + out["rows"] += len(rows) + out["rows_changed"] += sum(1 for n in filled if n) + out["fields_filled"] += sum(filled) + if any(filled): + out["objects_rewritten"] += 1 + if not dry_run: + self._check_space() + self._put(key, "".join(_line(r) + "\n" for r in fixed).encode("utf-8"), _labels(fixed)) + return out + def _now() -> str: return datetime.now(UTC).isoformat(timespec="seconds") @@ -852,6 +942,8 @@ __all__ = [ "KINDS", "MEMORY_KEY_ANNOTATION", "POPULATION_ANNOTATION", + "QUERYABLE", + "QUERY_FIELDS", "SNAPSHOT_ANNOTATION", "SPLIT_ANNOTATION", "FileStore", @@ -863,10 +955,12 @@ __all__ = [ "ExpiryRule", "Record", "SdkClient", + "check_query", "counts", "image_digest", "ingest", "memory_key", + "normalise_row", "open_store", "parse_memory_key", "repo_key", diff --git a/tests/test_plane_memory.py b/tests/test_plane_memory.py index 655f492..f56e56b 100644 --- a/tests/test_plane_memory.py +++ b/tests/test_plane_memory.py @@ -70,6 +70,7 @@ class FakeObjects: self.mode, self.objects, self.versions = select, {}, {} self.status, self.rules, self.tags_denied = versioning, RUNS_RULES if rules is None else rules, False self.selects: list[str] = [] + self.schema_rows = 2 # RustFS: a `where` field must be present in the object's leading rows, else it is unbound self.gets: list[str] = [] self.calls: list[tuple[Any, ...]] = [] @@ -104,9 +105,16 @@ class FakeObjects: return b"" if self.mode == "unbound": # RustFS: a field absent from the object's leading rows raise type("S3Error", (Exception,), {"code": "EvaluatorBindingDoesNotExist"})("A column name ... does not exist") + rows = [json.loads(line) for line in self.objects[key][0].splitlines() if line.strip()] + for name in re.findall(r"s\.\"(\w+)\"", expression): + if any(name not in r for r in rows[: self.schema_rows]): # RustFS infers the schema from the leading rows + raise type("S3Error", (Exception,), {"code": "EvaluatorBindingDoesNotExist"})(f"column {name} does not exist") + leading = rows[: self.schema_rows] + typeless = {k for r in leading for k, v in r.items() if v is None and all(x.get(k) is None for x in leading)} + if any(r.get(k) is not None for r in rows[self.schema_rows :] for k in typeless): # null-typed, set later + raise type("MinioException", (Exception,), {})("JSONParsingError: An error occurred while parsing the JSON file") clauses = re.findall(r"s\.\"(\w+)\" = ('(?:[^']|'')*'|true|false|-?[\d.]+)", expression) want = {n: v[1:-1].replace("''", "'") if v.startswith("'") else json.loads(v) for n, v in clauses} - rows = [json.loads(line) for line in self.objects[key][0].splitlines() if line.strip()] return b"".join(json.dumps(r).encode() + b"\n" for r in rows if all(r.get(k) == v for k, v in want.items())) def presign(self, method: str, key: str, expires: timedelta) -> str: @@ -185,7 +193,7 @@ def test_rows_filter_by_repo_pin_kind_and_where_with_their_object_version(store: assert len(store.rows()) == 5 assert {r.row["candidate"] for r in store.rows("https://github.com/o/r", kind="verdict")} == {"a.py::run", "a.py::x", "b.py::y"} assert [r.row["candidate"] for r in store.rows("github.com/o/r", PIN, "verdict", {"final": "disputed"})] == ["a.py::x"] - assert [r.row["kind"] for r in store.rows(where={"usd": 0.5})] == ["unit_cost"] + assert [r.row["candidate"] for r in store.rows(where={"final": "disputed"})] == ["a.py::x"] assert store.rows(where={"final": "no such"}) == [] record = store.rows("github.com/o/r", PIN, "verdict", {"candidate": "a.py::run"})[0] assert record.key == f"repos/github.com__o__r/{PIN}.jsonl" and record.version @@ -222,6 +230,112 @@ def test_presigned_urls_name_the_object(store: MemoryStore) -> None: assert url.startswith(("file://", "http://", "https://")) and "runs/r1/x/units.jsonl" in url +# --- the fixed schema: every row carries every queryable field ------------------------------------------------------ + + +def _late(field_kind: str = "example", n: int = 6) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """``n`` alert rows without any queryable field, and one ``field_kind`` row whose id sorts after them all (the + object is written in id order), so a declared field appears only after the object's leading rows.""" + alerts = [row("alert", f"rule-{i}", rule=f"r{i}") for i in range(n)] + return alerts, {**row(field_kind, "late", profile="static"), "id": "f" * 64} + + +def test_rows_written_without_optional_fields_gain_them_as_empty_strings(store: MemoryStore) -> None: + store.put_rows([verdict("a.py::run", "agreed"), row("unit_cost", "a:a.py", **{"pass": "a", "usd": 0.5})]) + store.ingest_rows("r2", "c-remember", [row("alert", "rule-1", run="r2", rule="r1", profile=None)]) + rows = {r.row["kind"]: r.row for r in store.rows()} + assert all(set(memory.QUERYABLE) <= set(r) for r in rows.values()) + assert rows["unit_cost"]["final"] == "" and rows["alert"]["candidate"] == "" and rows["alert"]["profile"] == "", "null too" + assert rows["verdict"]["final"] == "agreed" and rows["verdict"]["candidate"] == "a.py::run", "present values stay" + assert rows["unit_cost"]["usd"] == 0.5, "a field nobody queries is not touched" + store.drop_rows("github.com/o/r", PIN, [rows["alert"]["id"]]) + assert all(set(memory.QUERYABLE) <= set(r.row) for r in store.rows()) + + +def test_a_queryable_field_of_another_type_is_refused_on_write(store: MemoryStore) -> None: + with pytest.raises(MemoryStoreError, match=r"queryable field 'final' must be a string, got int"): + store.put_row(verdict("a.py::run", 1)) # type: ignore[arg-type] + assert store.rows() == [] + + +def test_a_declared_field_absent_from_the_leading_rows_is_queryable(store: MemoryStore) -> None: + alerts, late = _late(n=5000) # RustFS measured: a field first at row 5000 was unbound before the rule + store.put_rows([*alerts, late]) + assert [r.row["id"] for r in store.rows(kind="example", where={"profile": "static"})] == [late["id"]] + assert [r.row["id"] for r in store.rows(where={"profile": "static"})] == [late["id"]], "any kind: the union is written" + assert store.rows(kind="verdict", where={"final": "agreed"}) == [] + assert len(store.rows(where={"run": "r1", "profile": ""})) == len(alerts), "'' is the empty value" + + +def test_the_fake_client_mimics_rustfs_and_the_normaliser_repairs_a_pre_rule_object() -> None: + client = FakeObjects("works") + store = MinioStore(client, "bucket") + alerts, late = _late() + key = f"repos/github.com__o__r/{PIN}.jsonl" + legacy = sorted([*alerts, late], key=lambda r: str(r["id"])) # written before the rule: no queryable fields + client.put(key, "".join(json.dumps(r) + "\n" for r in legacy).encode(), {}) + with pytest.raises(MemoryStoreError, match=r"EvaluatorBindingDoesNotExist.*\['kind', 'profile'\] is absent"): + store.rows(kind="example", where={"profile": "static"}) + nulls = [{**r, "profile": None} for r in alerts] + [late] # the schema rows hold null, a later row a string + client.put(key, "".join(json.dumps(r) + "\n" for r in nulls).encode(), {}) + with pytest.raises(MemoryStoreError, match=r"JSONParsingError"): + store.rows(kind="example", where={"run": "r1"}) + assert store.normalise() == {"objects": 1, "objects_rewritten": 1, "rows": 7, "rows_changed": 7, + "fields_filled": 7 * len(memory.QUERYABLE) - 1} # fmt: skip + assert client.versions[key] == 3, "a rewrite is a new object version; versioning keeps the old" + assert [r.row["id"] for r in store.rows(kind="example", where={"profile": "static"})] == [late["id"]] + + +def test_a_where_on_an_undeclared_field_an_unknown_kind_or_null_is_refused_before_any_read() -> None: + client = FakeObjects("works") + store = _filled(client) + client.selects.clear() + with pytest.raises(MemoryStoreError, match=r"\['profile'\] are not queryable for kind 'verdict'.*QUERY_FIELDS"): + store.rows(kind="verdict", where={"profile": "static"}) + with pytest.raises(MemoryStoreError, match=r"\['usd'\] are not queryable for any kind"): + store.rows(where={"usd": 0.5}) + with pytest.raises(MemoryStoreError, match="unknown kind 'vibes'"): + store.rows(kind="vibes") + with pytest.raises(MemoryStoreError, match=r"\['final'\] = null never matches"): + store.rows(kind="verdict", where={"final": None}) + assert client.selects == [] + assert memory.QUERYABLE == ("candidate", "candidates_digest", "final", "profile") + + +@pytest.mark.parametrize("backend", ["file", "fake-minio"]) +def test_the_normaliser_is_idempotent_and_keeps_every_value(backend: str, tmp_path: Path) -> None: + store: MemoryStore = FileStore(tmp_path / "memory") if backend == "file" else MinioStore(FakeObjects(), "bucket", "p") + legacy = [verdict("a.py::run", "agreed"), row("unit_cost", "a:a.py", **{"pass": "a", "usd": 0.5, "final": None})] + store._put(f"repos/github.com__o__r/{PIN}.jsonl", "".join(json.dumps(r) + "\n" for r in legacy).encode()) + current = memory.normalise_row(row("alert", "x", pin=OTHER_PIN)) + store._put(f"repos/github.com__o__r/{OTHER_PIN}.jsonl", (json.dumps(current) + "\n").encode()) + first = store.normalise() + assert first == {"objects": 2, "objects_rewritten": 1, "rows": 3, "rows_changed": 2, "fields_filled": 2 + 4} + assert store.normalise() == {**first, "objects_rewritten": 0, "rows_changed": 0, "fields_filled": 0} + got = {r.row["kind"]: r.row for r in store.rows(repo="github.com/o/r", pin=PIN)} + assert got["verdict"] == memory.normalise_row(legacy[0]) and got["unit_cost"]["usd"] == 0.5 + + +def test_the_normaliser_names_a_queryable_field_of_another_type(tmp_path: Path) -> None: + store = FileStore(tmp_path / "memory") + store._put(f"repos/github.com__o__r/{PIN}.jsonl", (json.dumps({**verdict("a.py::run", "agreed"), "candidate": 3}) + "\n").encode()) + with pytest.raises(MemoryStoreError, match=rf"repos/github.com__o__r/{PIN}.jsonl:1: queryable field 'candidate'"): + store.normalise() + + +def test_the_memory_normalise_command_prints_its_counts(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + from openultrasast.cli import main + + store = FileStore(tmp_path / "memory") + store._put(f"repos/github.com__o__r/{PIN}.jsonl", (json.dumps(verdict("a.py::run", "agreed")) + "\n").encode()) + url = f"file://{tmp_path / 'memory'}" + assert main(["plane", "memory-normalise", "--store", url, "--dry-run"]) == 0 + assert 'would rewrite 1/1 objects; 1/1 rows filled 2 absent or null fields with ""' in capsys.readouterr().out + assert main(["plane", "memory-normalise", "--store", url]) == 0 + assert main(["plane", "memory-normalise", "--store", url]) == 0 + assert "rewrote 0/1 objects; 0/1 rows filled 0 absent or null fields" in capsys.readouterr().out.splitlines()[-1] + + # --- MinIO specifics on the fake client ---------------------------------------------------------------------------- @@ -260,8 +374,8 @@ def test_an_unbound_field_on_rustfs_raises_naming_the_schema_inference_never_ret client = FakeObjects("works") store = _filled(client) client.mode = "unbound" - with pytest.raises(MemoryStoreError, match=r"EvaluatorBindingDoesNotExist.*\['usd'\] is absent from the object's leading rows"): - store.rows(where={"usd": 0.5}) + with pytest.raises(MemoryStoreError, match=r"EvaluatorBindingDoesNotExist.*\['final'\] is absent from the object's leading rows"): + store.rows(where={"final": "agreed"}) # --- bucket verification: the admin configures once, the store only reads ------------------------------------------