test(import): cover parsers, importer and a seeded fuzz corpus; import 100 entries in the UI flow

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

3 files changed, 185 insertions, 4 deletions

diff --git a/tools/gen_fuzz_corpus.py b/tools/gen_fuzz_corpus.py
new file mode 100644
index 0000000..87d11ae
--- /dev/null
+++ b/tools/gen_fuzz_corpus.py
@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+# Author:  cirakg & claude ai
+# Summary: Generate the seeded fuzz corpus for the import parsers: mutated Aegis JSON and otpauth URIs.
+"""Seeded fuzz inputs for the Monkey C import tests (README 17.3).
+
+Starts from small valid documents (fake data only) and applies random
+mutations: truncation, deletion, insertion of random or structural bytes,
+byte flips, duplicated spans, deep nesting and long strings. The output is
+deterministic for a given seed, so the corpus in git is reproducible:
+
+    gen_fuzz_corpus.py [--seed 4242] [--out test/data/FuzzCorpus.mc]
+
+Inputs are written as hex, because many aren't valid UTF-8; the tests feed
+the bytes to the tokenizer and the Aegis reader, and the valid-UTF-8 ones
+to the whole importer as text.
+"""
+
+from __future__ import annotations
+
+import argparse
+import random
+from pathlib import Path
+
+SECRET = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"  # RFC 4226 test key in Base32, not a real secret
+AEGIS_SEED = (
+    '{"version": 1, "header": {"slots": null, "params": null}, "db": {"version": 3, "entries": ['
+    '{"type": "totp", "uuid": "u1", "name": "alice", "issuer": "GitHub", "note": "", "favorite": true, '
+    '"icon": null, "info": {"secret": "' + SECRET + '", "algo": "SHA1", "digits": 6, "period": 30}, '
+    '"groups": ["g1"]}, '
+    '{"type": "hotp", "name": "b\\u00e9b", "issuer": "VPN", "info": {"secret": "' + SECRET + '", '
+    '"algo": "SHA256", "digits": 8, "period": 30, "counter": 7}}], '
+    '"groups": [{"uuid": "g1", "name": "Work"}]}}'
+)
+URI_SEEDS = [
+    "otpauth://totp/GitHub:alice%40example.com?secret=" + SECRET + "&issuer=GitHub&digits=6",
+    "otpauth://hotp/VPN:bob?secret=" + SECRET + "&counter=5;algorithm=SHA256",
+    "otpauth://steam/Steam:carol?secret=" + SECRET,
+]
+STRUCTURAL = [
+    *(b"{", b"}", b"[", b"]", b'"', b":", b",", b"\\", b"\\u", b"\\ud83d"),
+    *(b"null", b"-", b"1e9", b"%", b"&", b"=", b"?"),
+]
+
+
+def mutate(data: bytes, rng: random.Random) -> bytes:
+    """One to three random mutations of data."""
+    out = bytearray(data)
+    for _ in range(rng.randint(1, 3)):
+        choice = rng.randrange(7)
+        at = rng.randrange(len(out) + 1) if out else 0
+        if choice == 0 and out:  # truncate
+            del out[at:]
+        elif choice == 1 and out:  # delete a span
+            del out[at : at + rng.randint(1, 8)]
+        elif choice == 2:  # insert random bytes
+            out[at:at] = bytes(rng.randrange(256) for _ in range(rng.randint(1, 4)))
+        elif choice == 3:  # insert a structural token
+            out[at:at] = rng.choice(STRUCTURAL)
+        elif choice == 4 and out:  # flip a byte
+            pos = rng.randrange(len(out))
+            out[pos] ^= 1 << rng.randrange(8)
+        elif choice == 5 and out:  # duplicate a span
+            span = out[at : at + rng.randint(1, 30)]
+            out[at:at] = span
+        else:  # deep nesting or a long string
+            long_string = b'"' + b"A" * rng.randint(520, 1100) + b'"'
+            out[at:at] = b"[" * rng.randint(10, 30) if rng.random() < 0.5 else long_string
+    return bytes(out)
+
+
+def corpus(seed: int, count_json: int, count_uri: int) -> tuple[list[bytes], list[bytes]]:
+    rng = random.Random(seed)
+    base = AEGIS_SEED.encode()
+    jsons = [base] + [mutate(base, rng) for _ in range(count_json - 1)]
+    mutated = [mutate(rng.choice(URI_SEEDS).encode(), rng) for _ in range(count_uri - len(URI_SEEDS))]
+    uris = [u.encode() for u in URI_SEEDS] + mutated
+    return jsons, uris
+
+
+def render(jsons: list[bytes], uris: list[bytes], seed: int) -> str:
+    def table(name: str, items: list[bytes]) -> str:
+        last = len(items) - 1
+        rows = "\n".join(
+            f'        "{item.hex()}"{"" if i == last else ","} // lint: allow-literal' for i, item in enumerate(items)
+        )
+        return f"    const {name} = [\n{rows}\n    ];\n"
+
+    return (
+        "// Author:  cirakg & claude ai\n"
+        "// Summary: Seeded fuzz inputs for the import parsers (generated by tools/gen_fuzz_corpus.py; do not edit).\n"
+        f"//\n// Seed {seed}. Hex because many inputs are not valid UTF-8. Fake data only.\n\n"
+        "import Toybox.Lang;\n\n(:test)\nmodule FuzzCorpus {\n"
+        + table("JSON", jsons)
+        + "\n"
+        + table("URIS", uris)
+        + "}\n"
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--seed", type=int, default=4242)
+    parser.add_argument("--out", type=Path, default=Path("test/data/FuzzCorpus.mc"))
+    parser.add_argument("--json", type=int, default=50)
+    parser.add_argument("--uris", type=int, default=50)
+    args = parser.parse_args(argv)
+    jsons, uris = corpus(args.seed, args.json, args.uris)
+    args.out.write_text(render(jsons, uris, args.seed), encoding="utf-8")
+    print(f"wrote {len(jsons)} JSON and {len(uris)} URI inputs to {args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tools/tests/test_gen_fuzz_corpus.py b/tools/tests/test_gen_fuzz_corpus.py
new file mode 100644
index 0000000..5ce8c99
--- /dev/null
+++ b/tools/tests/test_gen_fuzz_corpus.py
@@ -0,0 +1,41 @@
+# Author:  cirakg & claude ai
+# Summary: Tests for the fuzz corpus generator: determinism, mutation, valid Monkey C output, and the committed corpus.
+from __future__ import annotations
+
+import random
+from pathlib import Path
+
+import gen_fuzz_corpus
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_same_seed_gives_the_same_corpus() -> None:
+    assert gen_fuzz_corpus.corpus(7, 10, 10) == gen_fuzz_corpus.corpus(7, 10, 10)
+    assert gen_fuzz_corpus.corpus(7, 10, 10) != gen_fuzz_corpus.corpus(8, 10, 10)
+
+
+def test_corpus_starts_with_the_valid_seeds() -> None:
+    jsons, uris = gen_fuzz_corpus.corpus(1, 5, 5)
+    assert jsons[0] == gen_fuzz_corpus.AEGIS_SEED.encode()
+    assert uris[: len(gen_fuzz_corpus.URI_SEEDS)] == [u.encode() for u in gen_fuzz_corpus.URI_SEEDS]
+    assert len(jsons) == 5 and len(uris) == 5
+
+
+def test_mutate_changes_the_input() -> None:
+    rng = random.Random(3)
+    base = gen_fuzz_corpus.AEGIS_SEED.encode()
+    assert sum(gen_fuzz_corpus.mutate(base, rng) != base for _ in range(20)) >= 19
+
+
+def test_render_puts_commas_outside_the_comments() -> None:
+    text = gen_fuzz_corpus.render([b"{}", b"[]"], [b"x"], 1)
+    assert '"7b7d", // lint: allow-literal' in text
+    assert '"5b5d" // lint: allow-literal' in text
+    assert text.startswith("// Author:  cirakg & claude ai\n// Summary:")
+
+
+def test_committed_corpus_matches_the_generator() -> None:
+    jsons, uris = gen_fuzz_corpus.corpus(4242, 50, 50)
+    committed = (ROOT / "test" / "data" / "FuzzCorpus.mc").read_text(encoding="utf-8")
+    assert committed == gen_fuzz_corpus.render(jsons, uris, 4242)
diff --git a/tools/ui_flow.py b/tools/ui_flow.py
index f43cf81..3e82231 100644
--- a/tools/ui_flow.py
+++ b/tools/ui_flow.py
@@ -19,7 +19,8 @@ drawing each screen, so every step waits for the screen it expects instead
 of sleeping. Menus are native Menu2 views that print nothing; for those the
 script waits a fixed time.
 
-The flow covers: the list, a TOTP entry, a HOTP entry and its next code, the
+The flow covers: the results of an Aegis import sent through the settings
+form (the fixture sets the form, as the phone would), the list, a TOTP entry, a HOTP entry and its next code, the
 PIN screen, a wrong PIN, a correct PIN opening a locked entry, auto-lock
 closing it, the menu, the About screen (watches with buttons; touch-only
 menus have no focus to move) and the lockout after 5 wrong PINs.
@@ -46,6 +47,8 @@ import sim_screenshot as shots
 EXIT_OK, EXIT_DIFFERENT, EXIT_FLOW = 0, 1, 2
 SCREEN_TIMEOUT_S = 20.0
 AUTO_LOCK_TIMEOUT_S = 25.0  # the UI-test build locks after 10 s without input
+IMPORT_TIMEOUT_S = 90.0  # the 100-entry fixture import runs on a timer after start-up
+IMPORT_MEMORY_BUDGET = 0.70  # README 18: peak memory at most 70 % of the watch-app memory
 MENU_SETTLE_S = 1.5  # Menu2 has no "shown" line
 SETTLE_S = 0.4  # after a "shown" line, before the screenshot
 CLICK_PAUSE_S = 0.25
@@ -225,6 +228,12 @@ class Flow:
         else:
             self.sim.swipe(self._w() // 2, self._h() * 3 // 4, self._w() // 2, self._h() // 4)
 
+    def prev(self) -> None:
+        if self.sim.has("up"):
+            self.sim.press("up")
+        else:
+            self.sim.swipe(self._w() // 2, self._h() // 4, self._w() // 2, self._h() * 3 // 4)
+
     def back(self) -> None:
         self.sim.press("esc")
 
@@ -305,9 +314,26 @@ class Flow:
         self.results[name] = shots.check_golden(screen, label, out, golden)
 
 
+def flow_import(f: Flow) -> None:
+    """The fixture's 100-entry Aegis import through the settings form opens its
+    results; its peak memory must stay within the budget."""
+    f.expect("results", IMPORT_TIMEOUT_S)
+    stats = f.log.latest(r"DIAG import entries=(\d+) wall_ms=(\d+) mem_peak=(\d+) mem_total=(\d+)")
+    if stats is None:
+        raise FlowError("the import didn't report its statistics")
+    entries, wall_ms, peak, total = (int(v) for v in stats.groups())
+    fraction = peak / total
+    print(f"[{f.sim.device_dir.name}] import of {entries} entries: {wall_ms} ms, peak memory {fraction:.1%}")
+    if fraction > IMPORT_MEMORY_BUDGET:
+        raise FlowError(f"import peak memory {fraction:.1%} is over {IMPORT_MEMORY_BUDGET:.0%}")
+    f.shot("import_results")
+    f.act()
+    f.back()
+    f.expect("list")
+
+
 def flow_entries(f: Flow) -> None:
     """The list, a TOTP entry, a HOTP entry and its next code."""
-    f.expect("list")
     f.shot("list")
     f.act()
     f.open_focused_row()  # GitHub, TOTP
@@ -333,8 +359,7 @@ def flow_entries(f: Flow) -> None:
 
 def flow_locked_entry(f: Flow) -> None:
     """Bank is locked: PIN screen, a wrong PIN, the right PIN, then auto-lock."""
-    for _ in range(6):  # wrap around to Bank (index 2 of 7)
-        f.next()
+    f.prev()  # from VPN back to Bank
     f.act()
     f.open_focused_row()
     f.expect("pin")
@@ -391,6 +416,7 @@ def flow_lockout(f: Flow) -> None:
 
 def run_flow(f: Flow) -> None:
     """The scripted session; see the module docstring."""
+    flow_import(f)
     flow_entries(f)
     flow_locked_entry(f)
     flow_menu(f)