From 8fd4112e9f9b3a3dea33393bc66873e1b1ad4e8d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 12:21:38 +0900 Subject: [PATCH 01/14] test(browser): require bounded Chromium process-tree RSS --- .../test_agent_task_pinned_chrome_contract.py | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py index 87faf198..4b68965b 100644 --- a/tests/test_agent_task_pinned_chrome_contract.py +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -86,6 +86,26 @@ def test_agent_task_records_real_bounded_resource_evidence(self) -> None: with self.subTest(expected=expected): self.assertIn(expected, runner) + def test_agent_task_records_bounded_chromium_process_tree_rss(self) -> None: + """The evidence runner must measure a bounded root-plus-descendant process set.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_process_tree_contract") + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + "MAX_BROWSER_PROCESS_TREE_SIZE", + "_parse_linux_children_process_ids", + "_discover_linux_process_tree_ids", + "_sample_linux_process_set_rss_bytes", + ): + with self.subTest(expected=expected): + self.assertIn(expected, namespace) + for expected in ( + '"chromium_process_count"', + '"chromium_process_set_rss_bytes"', + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: """Runner-side RSS evidence must not accept ambiguous proc status input.""" @@ -104,6 +124,26 @@ def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: with self.assertRaises((ValueError, OverflowError)): parser(malformed) + def test_linux_children_parser_is_bounded_unique_and_positive(self) -> None: + """Process-tree discovery must reject ambiguous or unbounded kernel child lists.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_children_contract") + parser = namespace["_parse_linux_children_process_ids"] + limit = namespace["MAX_BROWSER_PROCESS_TREE_SIZE"] + self.assertEqual(parser(""), ()) + self.assertEqual(parser("12 34\n"), (12, 34)) + for malformed in ( + "0\n", + "12 12\n", + "12 child\n", + "-1\n", + "12 34 trailing!\n", + " ".join(str(index) for index in range(1, limit + 2)), + ): + with self.subTest(malformed=malformed[:120]): + with self.assertRaises(ValueError): + parser(malformed) + def test_agent_task_fixture_runs_under_the_existing_pinned_chrome_job(self) -> None: """No floating browser or second workflow may be introduced for this slice.""" From f73f2a66ec2cb6bdbb66f27b7e56ad9dc29022c3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:09:15 +0900 Subject: [PATCH 02/14] test(browser): measure bounded Chromium process-tree RSS --- scripts/ci/run_mv3_compatibility.py | 95 +++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 1aa680d5..2086ed33 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -42,6 +42,7 @@ FIXTURE_TIMEOUT_SECONDS = 20.0 MAX_WEBDRIVER_RESPONSE_BYTES = 1_048_576 MAX_PROC_STATUS_CHARACTERS = 65_536 +MAX_BROWSER_PROCESS_TREE_SIZE = 256 MAX_U64 = (1 << 64) - 1 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf" PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.") @@ -241,6 +242,86 @@ def _sample_linux_process_rss_bytes(process_id: int) -> int: return _parse_linux_proc_status_rss_bytes(status_text) +def _parse_linux_children_process_ids(children_text: str) -> tuple[int, ...]: + """Parse one bounded Linux ``children`` list into unique positive process IDs.""" + + fields = children_text.split() + if len(fields) > MAX_BROWSER_PROCESS_TREE_SIZE: + raise ValueError("Linux child process list exceeded the bounded process-tree size") + process_ids: list[int] = [] + seen: set[int] = set() + for field in fields: + if not field.isascii() or not field.isdigit(): + raise ValueError("malformed Linux child process identifier") + process_id = int(field, 10) + if process_id <= 0 or process_id in seen: + raise ValueError("Linux child process identifiers must be unique and positive") + seen.add(process_id) + process_ids.append(process_id) + return tuple(process_ids) + + +def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: + """Discover one bounded root-plus-descendant Linux process tree from ``/proc``.""" + + if ( + isinstance(root_process_id, bool) + or not isinstance(root_process_id, int) + or root_process_id <= 0 + ): + raise ValueError("invalid Linux root process identifier") + + discovered: list[int] = [] + queued: list[int] = [root_process_id] + known: set[int] = {root_process_id} + while queued: + process_id = queued.pop(0) + discovered.append(process_id) + if len(discovered) > MAX_BROWSER_PROCESS_TREE_SIZE: + raise ValueError("Linux process tree exceeded the bounded process-tree size") + + children_path = ( + pathlib.Path("/proc") + / str(process_id) + / "task" + / str(process_id) + / "children" + ) + with children_path.open("r", encoding="utf-8", errors="strict") as children_file: + children_text = children_file.read(MAX_PROC_STATUS_CHARACTERS + 1) + if len(children_text) > MAX_PROC_STATUS_CHARACTERS: + raise RuntimeError("Linux proc children exceeded the bounded text limit") + children = _parse_linux_children_process_ids(children_text) + for child_process_id in children: + if child_process_id in known: + raise ValueError("Linux process tree contained a duplicate process identifier") + if len(known) >= MAX_BROWSER_PROCESS_TREE_SIZE: + raise ValueError("Linux process tree exceeded the bounded process-tree size") + known.add(child_process_id) + queued.append(child_process_id) + + return tuple(discovered) + + +def _sample_linux_process_set_rss_bytes(process_ids: tuple[int, ...]) -> int: + """Sample and sum one exact bounded Linux process set without silent overflow.""" + + if not process_ids or len(process_ids) > MAX_BROWSER_PROCESS_TREE_SIZE: + raise ValueError("invalid Linux process set size") + if len(set(process_ids)) != len(process_ids): + raise ValueError("Linux process set identifiers must be unique") + + total_rss_bytes = 0 + for process_id in process_ids: + if isinstance(process_id, bool) or not isinstance(process_id, int) or process_id <= 0: + raise ValueError("invalid Linux process identifier") + rss_bytes = _sample_linux_process_rss_bytes(process_id) + if rss_bytes > MAX_U64 - total_rss_bytes: + raise OverflowError("Linux process-set RSS exceeds u64 byte range") + total_rss_bytes += rss_bytes + return total_rss_bytes + + def _wait_for_extension_evidence( driver_port: int, session_id: str, @@ -652,7 +733,13 @@ def _run_agent_task_browser_pass( raise RuntimeError(f"Agent Task state post-condition failed: {state!r}") if text != AGENT_TASK_INPUT_VALUE: raise RuntimeError("Agent Task result did not match the synthetic typed value") + + chromium_process_ids = _discover_linux_process_tree_ids(browser_process_id) browser_process_rss_bytes = _sample_linux_process_rss_bytes(browser_process_id) + chromium_process_set_rss_bytes = _sample_linux_process_set_rss_bytes( + chromium_process_ids + ) + chromium_process_count = len(chromium_process_ids) task_duration_ms = round((time.monotonic() - started) * 1000, 3) if task_duration_ms <= 0: raise RuntimeError("Agent Task measured a non-positive task duration") @@ -664,6 +751,8 @@ def _run_agent_task_browser_pass( "submit_semantics_verified": True, "extensions_disabled": True, "browser_process_rss_bytes": browser_process_rss_bytes, + "chromium_process_count": chromium_process_count, + "chromium_process_set_rss_bytes": chromium_process_set_rss_bytes, "semantic_observation_bytes": semantic_observation_bytes, "action_latency_ms": action_latency_ms, "task_duration_ms": task_duration_ms, @@ -720,6 +809,8 @@ def _run_agent_task_trial( "submit_semantics_verified": result["submit_semantics_verified"], "extensions_disabled": result["extensions_disabled"], "browser_process_rss_bytes": result["browser_process_rss_bytes"], + "chromium_process_count": result["chromium_process_count"], + "chromium_process_set_rss_bytes": result["chromium_process_set_rss_bytes"], "semantic_observation_bytes": result["semantic_observation_bytes"], "action_latency_ms": result["action_latency_ms"], "task_duration_ms": result["task_duration_ms"], @@ -853,6 +944,10 @@ def main() -> int: and trial.get("profile_cleaned") is True and isinstance(trial.get("browser_process_rss_bytes"), int) and trial["browser_process_rss_bytes"] > 0 + and isinstance(trial.get("chromium_process_count"), int) + and 0 < trial["chromium_process_count"] <= MAX_BROWSER_PROCESS_TREE_SIZE + and isinstance(trial.get("chromium_process_set_rss_bytes"), int) + and trial["chromium_process_set_rss_bytes"] > 0 and isinstance(trial.get("semantic_observation_bytes"), int) and trial["semantic_observation_bytes"] > 0 and isinstance(trial.get("action_latency_ms"), (int, float)) From a64d3b16967ff99b228ae0b44ff74a7494f52fc8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:11:07 +0900 Subject: [PATCH 03/14] test(browser): reject unreliable live proc children discovery --- .../test_agent_task_pinned_chrome_contract.py | 33 ++++++++++--------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py index 4b68965b..aa0ec599 100644 --- a/tests/test_agent_task_pinned_chrome_contract.py +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -93,12 +93,15 @@ def test_agent_task_records_bounded_chromium_process_tree_rss(self) -> None: runner = RUNNER.read_text(encoding="utf-8") for expected in ( "MAX_BROWSER_PROCESS_TREE_SIZE", - "_parse_linux_children_process_ids", + "MAX_PROC_PROCESS_SCAN_SIZE", + "_parse_linux_proc_status_process_identity", + "_snapshot_linux_process_parent_ids", "_discover_linux_process_tree_ids", "_sample_linux_process_set_rss_bytes", ): with self.subTest(expected=expected): self.assertIn(expected, namespace) + self.assertNotIn("_parse_linux_children_process_ids", namespace) for expected in ( '"chromium_process_count"', '"chromium_process_set_rss_bytes"', @@ -124,23 +127,23 @@ def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: with self.assertRaises((ValueError, OverflowError)): parser(malformed) - def test_linux_children_parser_is_bounded_unique_and_positive(self) -> None: - """Process-tree discovery must reject ambiguous or unbounded kernel child lists.""" + def test_linux_status_identity_parser_is_strict_and_positive(self) -> None: + """Parent-map discovery must parse one unambiguous process identity per status.""" - namespace = runpy.run_path(str(RUNNER), run_name="agent_task_children_contract") - parser = namespace["_parse_linux_children_process_ids"] - limit = namespace["MAX_BROWSER_PROCESS_TREE_SIZE"] - self.assertEqual(parser(""), ()) - self.assertEqual(parser("12 34\n"), (12, 34)) + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_parent_map_contract") + parser = namespace["_parse_linux_proc_status_process_identity"] + self.assertEqual(parser("Name:\tchrome\nPid:\t34\nPPid:\t12\n"), (34, 12)) + self.assertEqual(parser("Name:\tinit\nPid:\t1\nPPid:\t0\n"), (1, 0)) for malformed in ( - "0\n", - "12 12\n", - "12 child\n", - "-1\n", - "12 34 trailing!\n", - " ".join(str(index) for index in range(1, limit + 2)), + "Name:\tchrome\nPid:\t34\n", + "Name:\tchrome\nPPid:\t12\n", + "Pid:\t0\nPPid:\t12\n", + "Pid:\t34\nPPid:\t-1\n", + "Pid:\tchild\nPPid:\t12\n", + "Pid:\t34\nPid:\t35\nPPid:\t12\n", + "Pid:\t34\nPPid:\t12\nPPid:\t13\n", ): - with self.subTest(malformed=malformed[:120]): + with self.subTest(malformed=malformed): with self.assertRaises(ValueError): parser(malformed) From 675e5737f6cc507ae85e2fc5d56f43a17d4d22d2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:13:22 +0900 Subject: [PATCH 04/14] fix(browser): discover Chromium process set from proc status --- scripts/ci/run_mv3_compatibility.py | 124 ++++++++++++++++++---------- 1 file changed, 81 insertions(+), 43 deletions(-) diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 2086ed33..290bb7bd 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -43,6 +43,7 @@ MAX_WEBDRIVER_RESPONSE_BYTES = 1_048_576 MAX_PROC_STATUS_CHARACTERS = 65_536 MAX_BROWSER_PROCESS_TREE_SIZE = 256 +MAX_PROC_PROCESS_SCAN_SIZE = 32_768 MAX_U64 = (1 << 64) - 1 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf" PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.") @@ -229,6 +230,35 @@ def _parse_linux_proc_status_rss_bytes(status_text: str) -> int: return rss_values[0] +def _parse_linux_proc_status_process_identity(status_text: str) -> tuple[int, int]: + """Parse exactly one positive ``Pid`` and one non-negative ``PPid`` from status.""" + + parsed: dict[str, int] = {} + for line in status_text.splitlines(): + if not (line.startswith("Pid:") or line.startswith("PPid:")): + continue + fields = line.split() + if len(fields) != 2 or fields[0] not in {"Pid:", "PPid:"}: + raise ValueError("malformed Linux process identity field") + label = fields[0] + if label in parsed: + raise ValueError("duplicate Linux process identity field") + raw_process_id = fields[1] + if not raw_process_id.isascii() or not raw_process_id.isdigit(): + raise ValueError("malformed Linux process identity value") + parsed[label] = int(raw_process_id, 10) + + if set(parsed) != {"Pid:", "PPid:"}: + raise ValueError("Linux proc status must contain exactly one Pid and PPid") + process_id = parsed["Pid:"] + parent_process_id = parsed["PPid:"] + if process_id <= 0: + raise ValueError("Linux process identifier must be positive") + if parent_process_id < 0: + raise ValueError("Linux parent process identifier must be non-negative") + return process_id, parent_process_id + + def _sample_linux_process_rss_bytes(process_id: int) -> int: """Read one attributed Linux process RSS through a bounded ``/proc`` status file.""" @@ -242,27 +272,45 @@ def _sample_linux_process_rss_bytes(process_id: int) -> int: return _parse_linux_proc_status_rss_bytes(status_text) -def _parse_linux_children_process_ids(children_text: str) -> tuple[int, ...]: - """Parse one bounded Linux ``children`` list into unique positive process IDs.""" +def _snapshot_linux_process_parent_ids() -> dict[int, int]: + """Read one bounded best-effort Linux PID/PPID snapshot from process status files.""" - fields = children_text.split() - if len(fields) > MAX_BROWSER_PROCESS_TREE_SIZE: - raise ValueError("Linux child process list exceeded the bounded process-tree size") - process_ids: list[int] = [] - seen: set[int] = set() - for field in fields: - if not field.isascii() or not field.isdigit(): - raise ValueError("malformed Linux child process identifier") - process_id = int(field, 10) - if process_id <= 0 or process_id in seen: - raise ValueError("Linux child process identifiers must be unique and positive") - seen.add(process_id) - process_ids.append(process_id) - return tuple(process_ids) + proc_root = pathlib.Path("/proc") + process_entries: list[tuple[int, pathlib.Path]] = [] + for entry in proc_root.iterdir(): + raw_process_id = entry.name + if not raw_process_id.isascii() or not raw_process_id.isdigit(): + continue + process_id = int(raw_process_id, 10) + if process_id <= 0: + continue + process_entries.append((process_id, entry)) + if len(process_entries) > MAX_PROC_PROCESS_SCAN_SIZE: + raise RuntimeError("Linux proc process scan exceeded the bounded entry limit") + + parent_ids: dict[int, int] = {} + for expected_process_id, entry in sorted(process_entries): + status_path = entry / "status" + try: + with status_path.open("r", encoding="utf-8", errors="strict") as status_file: + status_text = status_file.read(MAX_PROC_STATUS_CHARACTERS + 1) + except FileNotFoundError: + continue + if len(status_text) > MAX_PROC_STATUS_CHARACTERS: + raise RuntimeError("Linux proc status exceeded the bounded text limit") + process_id, parent_process_id = _parse_linux_proc_status_process_identity( + status_text + ) + if process_id != expected_process_id: + raise RuntimeError("Linux proc status identity did not match its directory") + if process_id in parent_ids: + raise RuntimeError("Linux proc process snapshot contained a duplicate PID") + parent_ids[process_id] = parent_process_id + return parent_ids def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: - """Discover one bounded root-plus-descendant Linux process tree from ``/proc``.""" + """Discover one bounded root-plus-descendant process set from a PID/PPID snapshot.""" if ( isinstance(root_process_id, bool) @@ -271,35 +319,25 @@ def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: ): raise ValueError("invalid Linux root process identifier") - discovered: list[int] = [] - queued: list[int] = [root_process_id] - known: set[int] = {root_process_id} - while queued: - process_id = queued.pop(0) - discovered.append(process_id) - if len(discovered) > MAX_BROWSER_PROCESS_TREE_SIZE: - raise ValueError("Linux process tree exceeded the bounded process-tree size") - - children_path = ( - pathlib.Path("/proc") - / str(process_id) - / "task" - / str(process_id) - / "children" + parent_ids = _snapshot_linux_process_parent_ids() + if root_process_id not in parent_ids: + raise RuntimeError("Linux process snapshot did not contain the browser root PID") + + discovered = [root_process_id] + known = {root_process_id} + while True: + children = sorted( + process_id + for process_id, parent_process_id in parent_ids.items() + if parent_process_id in known and process_id not in known ) - with children_path.open("r", encoding="utf-8", errors="strict") as children_file: - children_text = children_file.read(MAX_PROC_STATUS_CHARACTERS + 1) - if len(children_text) > MAX_PROC_STATUS_CHARACTERS: - raise RuntimeError("Linux proc children exceeded the bounded text limit") - children = _parse_linux_children_process_ids(children_text) - for child_process_id in children: - if child_process_id in known: - raise ValueError("Linux process tree contained a duplicate process identifier") + if not children: + break + for process_id in children: if len(known) >= MAX_BROWSER_PROCESS_TREE_SIZE: raise ValueError("Linux process tree exceeded the bounded process-tree size") - known.add(child_process_id) - queued.append(child_process_id) - + known.add(process_id) + discovered.append(process_id) return tuple(discovered) From 623ac23790535f3025a82af4f357e9fc2a1b3c36 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:16:23 +0900 Subject: [PATCH 05/14] test(browser): require one sampled process evidence snapshot --- .../test_agent_task_pinned_chrome_contract.py | 26 ++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py index aa0ec599..9d24c1ac 100644 --- a/tests/test_agent_task_pinned_chrome_contract.py +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -87,7 +87,7 @@ def test_agent_task_records_real_bounded_resource_evidence(self) -> None: self.assertIn(expected, runner) def test_agent_task_records_bounded_chromium_process_tree_rss(self) -> None: - """The evidence runner must measure a bounded root-plus-descendant process set.""" + """The evidence runner must measure one bounded sampled process-set snapshot.""" namespace = runpy.run_path(str(RUNNER), run_name="agent_task_process_tree_contract") runner = RUNNER.read_text(encoding="utf-8") @@ -95,20 +95,40 @@ def test_agent_task_records_bounded_chromium_process_tree_rss(self) -> None: "MAX_BROWSER_PROCESS_TREE_SIZE", "MAX_PROC_PROCESS_SCAN_SIZE", "_parse_linux_proc_status_process_identity", - "_snapshot_linux_process_parent_ids", + "_snapshot_linux_process_evidence", "_discover_linux_process_tree_ids", "_sample_linux_process_set_rss_bytes", ): with self.subTest(expected=expected): self.assertIn(expected, namespace) self.assertNotIn("_parse_linux_children_process_ids", namespace) + self.assertNotIn("_snapshot_linux_process_parent_ids", namespace) for expected in ( '"chromium_process_count"', '"chromium_process_set_rss_bytes"', + '"failure_type"', ): with self.subTest(expected=expected): self.assertIn(expected, runner) + def test_process_tree_and_rss_use_one_sampled_process_snapshot(self) -> None: + """Descendant RSS must come from the same bounded status snapshot as lineage.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_process_snapshot") + discover = namespace["_discover_linux_process_tree_ids"] + sample = namespace["_sample_linux_process_set_rss_bytes"] + evidence = { + 10: (1, 100), + 20: (10, 200), + 30: (20, 300), + 40: (999, 400), + } + process_ids = discover(10, evidence) + self.assertEqual(process_ids, (10, 20, 30)) + self.assertEqual(sample(process_ids, evidence), 600) + with self.assertRaises(ValueError): + sample((10, 50), evidence) + def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: """Runner-side RSS evidence must not accept ambiguous proc status input.""" @@ -128,7 +148,7 @@ def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: parser(malformed) def test_linux_status_identity_parser_is_strict_and_positive(self) -> None: - """Parent-map discovery must parse one unambiguous process identity per status.""" + """Process snapshot discovery must parse one unambiguous identity per status.""" namespace = runpy.run_path(str(RUNNER), run_name="agent_task_parent_map_contract") parser = namespace["_parse_linux_proc_status_process_identity"] From 1d12a1ce5939aded2ea149ed88413e48e77c8d3b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:20:14 +0900 Subject: [PATCH 06/14] fix(browser): sample Chromium lineage and RSS in one proc sweep --- scripts/ci/run_mv3_compatibility.py | 64 +++++++++++++++++++---------- 1 file changed, 43 insertions(+), 21 deletions(-) diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 290bb7bd..10c66af5 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -272,8 +272,8 @@ def _sample_linux_process_rss_bytes(process_id: int) -> int: return _parse_linux_proc_status_rss_bytes(status_text) -def _snapshot_linux_process_parent_ids() -> dict[int, int]: - """Read one bounded best-effort Linux PID/PPID snapshot from process status files.""" +def _snapshot_linux_process_evidence() -> dict[int, tuple[int, int | None]]: + """Capture one bounded best-effort PID/PPID/RSS sweep from Linux proc status.""" proc_root = pathlib.Path("/proc") process_entries: list[tuple[int, pathlib.Path]] = [] @@ -288,7 +288,7 @@ def _snapshot_linux_process_parent_ids() -> dict[int, int]: if len(process_entries) > MAX_PROC_PROCESS_SCAN_SIZE: raise RuntimeError("Linux proc process scan exceeded the bounded entry limit") - parent_ids: dict[int, int] = {} + process_evidence: dict[int, tuple[int, int | None]] = {} for expected_process_id, entry in sorted(process_entries): status_path = entry / "status" try: @@ -303,14 +303,24 @@ def _snapshot_linux_process_parent_ids() -> dict[int, int]: ) if process_id != expected_process_id: raise RuntimeError("Linux proc status identity did not match its directory") - if process_id in parent_ids: + if process_id in process_evidence: raise RuntimeError("Linux proc process snapshot contained a duplicate PID") - parent_ids[process_id] = parent_process_id - return parent_ids - - -def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: - """Discover one bounded root-plus-descendant process set from a PID/PPID snapshot.""" + try: + rss_bytes: int | None = _parse_linux_proc_status_rss_bytes(status_text) + except ValueError as exc: + if "VmRSS must be positive" in str(exc) or "exactly one VmRSS" in str(exc): + rss_bytes = None + else: + raise + process_evidence[process_id] = (parent_process_id, rss_bytes) + return process_evidence + + +def _discover_linux_process_tree_ids( + root_process_id: int, + process_evidence: dict[int, tuple[int, int | None]], +) -> tuple[int, ...]: + """Discover one bounded root-plus-descendant set from sampled process evidence.""" if ( isinstance(root_process_id, bool) @@ -318,9 +328,7 @@ def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: or root_process_id <= 0 ): raise ValueError("invalid Linux root process identifier") - - parent_ids = _snapshot_linux_process_parent_ids() - if root_process_id not in parent_ids: + if root_process_id not in process_evidence: raise RuntimeError("Linux process snapshot did not contain the browser root PID") discovered = [root_process_id] @@ -328,7 +336,7 @@ def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: while True: children = sorted( process_id - for process_id, parent_process_id in parent_ids.items() + for process_id, (parent_process_id, _rss_bytes) in process_evidence.items() if parent_process_id in known and process_id not in known ) if not children: @@ -341,8 +349,11 @@ def _discover_linux_process_tree_ids(root_process_id: int) -> tuple[int, ...]: return tuple(discovered) -def _sample_linux_process_set_rss_bytes(process_ids: tuple[int, ...]) -> int: - """Sample and sum one exact bounded Linux process set without silent overflow.""" +def _sample_linux_process_set_rss_bytes( + process_ids: tuple[int, ...], + process_evidence: dict[int, tuple[int, int | None]], +) -> int: + """Sum positive sampled RSS for one exact bounded process set without overflow.""" if not process_ids or len(process_ids) > MAX_BROWSER_PROCESS_TREE_SIZE: raise ValueError("invalid Linux process set size") @@ -353,7 +364,11 @@ def _sample_linux_process_set_rss_bytes(process_ids: tuple[int, ...]) -> int: for process_id in process_ids: if isinstance(process_id, bool) or not isinstance(process_id, int) or process_id <= 0: raise ValueError("invalid Linux process identifier") - rss_bytes = _sample_linux_process_rss_bytes(process_id) + if process_id not in process_evidence: + raise ValueError("Linux process set was not present in the sampled evidence") + rss_bytes = process_evidence[process_id][1] + if isinstance(rss_bytes, bool) or not isinstance(rss_bytes, int) or rss_bytes <= 0: + raise ValueError("Linux process set contained unavailable sampled RSS") if rss_bytes > MAX_U64 - total_rss_bytes: raise OverflowError("Linux process-set RSS exceeds u64 byte range") total_rss_bytes += rss_bytes @@ -772,10 +787,15 @@ def _run_agent_task_browser_pass( if text != AGENT_TASK_INPUT_VALUE: raise RuntimeError("Agent Task result did not match the synthetic typed value") - chromium_process_ids = _discover_linux_process_tree_ids(browser_process_id) + process_evidence = _snapshot_linux_process_evidence() + chromium_process_ids = _discover_linux_process_tree_ids( + browser_process_id, + process_evidence, + ) browser_process_rss_bytes = _sample_linux_process_rss_bytes(browser_process_id) chromium_process_set_rss_bytes = _sample_linux_process_set_rss_bytes( - chromium_process_ids + chromium_process_ids, + process_evidence, ) chromium_process_count = len(chromium_process_ids) task_duration_ms = round((time.monotonic() - started) * 1000, 3) @@ -917,11 +937,12 @@ def main() -> int: trial_number, ) ) - except (OSError, ValueError, RuntimeError, json.JSONDecodeError): + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: trial_results.append( { "trial_number": trial_number, "passed": False, + "failure_type": type(exc).__name__, } ) @@ -959,11 +980,12 @@ def main() -> int: trial_number, ) ) - except (OSError, ValueError, RuntimeError, json.JSONDecodeError): + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: agent_task_trials.append( { "trial_number": trial_number, "passed": False, + "failure_type": type(exc).__name__, } ) From ec5a34ad7ef453879847182e357380dfb8e1312a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 13:22:19 +0900 Subject: [PATCH 07/14] docs(changelog): record sampled Chromium process-set evidence --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e4bd39c..15b5a2de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ All notable changes to OriginWeave are documented in this file. The format follo - Cumulative interactive-first RAM, VRAM, batch, local-model, admission, pause, and compositor-pressure mitigation plans, including active-consumer reduction at exact hard limits. - Universally value-redacted network evidence with explicit path, metadata, and provenance bounds; ambiguous path rejection; validated source URLs; lowercase SHA-256 identifiers; and verification state. - Rust 1.97.1 build contract, strict Clippy and rustdoc gates, and exact production function, line, region, and branch coverage enforcement. +- Pinned Chrome-for-Testing Agent Task evidence now captures a bounded sampled Chromium root-plus-descendant process count and RSS total from one `/proc` status sweep, with bounded failure-type diagnostics while preserving the root-only metric and making no trusted per-task attribution claim. - Hourly bounded OpenCode product-development workflow using `NVIDIA_NIM_API_KEY`, an unprivileged disposable workspace, loopback-only model broker, independently verified patches, and publication through a dedicated `OPENCODE_PR_TOKEN` that cannot review or merge. - Architecture, agent, security, contribution, research, database naming, roadmap, quality-gate, and TLS service-identity ADR documentation. - Authoritative product documentation graph spanning PRD, TRD, ADR lifecycle/index, product-wide UML, conceptual ERD, requirement/decision traceability, threat modeling, product-wide test strategy, operability, API/protocol, release/rollback, and current primary-source standards doctoring, with machine-checkable repository contracts that keep conversation-derived future work distinct from protected-main implementation claims. From 85df93827e187865a136facb6a2fe37e265d4df2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:02:02 +0900 Subject: [PATCH 08/14] test(browser): reproduce missing descendant RSS failure --- tests/test_agent_task_pinned_chrome_contract.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py index 9d24c1ac..7f8e91b7 100644 --- a/tests/test_agent_task_pinned_chrome_contract.py +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -129,6 +129,18 @@ def test_process_tree_and_rss_use_one_sampled_process_snapshot(self) -> None: with self.assertRaises(ValueError): sample((10, 50), evidence) + def test_process_set_tolerates_descendant_without_resident_rss(self) -> None: + """A sampled child with no resident RSS must not invalidate the whole tree.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_zero_rss_contract") + sample = namespace["_sample_linux_process_set_rss_bytes"] + evidence = { + 10: (1, 100), + 20: (10, None), + 30: (20, 300), + } + self.assertEqual(sample((10, 20, 30), evidence), 400) + def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: """Runner-side RSS evidence must not accept ambiguous proc status input.""" From cbf922fccc83782d3e114ed65afbeb6d84ef5ce6 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:08:05 +0900 Subject: [PATCH 09/14] fix(browser): tolerate nonresident Chromium descendants --- scripts/ci/run_mv3_compatibility.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 10c66af5..bf224c2e 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -353,7 +353,7 @@ def _sample_linux_process_set_rss_bytes( process_ids: tuple[int, ...], process_evidence: dict[int, tuple[int, int | None]], ) -> int: - """Sum positive sampled RSS for one exact bounded process set without overflow.""" + """Sum resident RSS for one exact bounded process set without overflow.""" if not process_ids or len(process_ids) > MAX_BROWSER_PROCESS_TREE_SIZE: raise ValueError("invalid Linux process set size") @@ -367,8 +367,10 @@ def _sample_linux_process_set_rss_bytes( if process_id not in process_evidence: raise ValueError("Linux process set was not present in the sampled evidence") rss_bytes = process_evidence[process_id][1] + if rss_bytes is None: + continue if isinstance(rss_bytes, bool) or not isinstance(rss_bytes, int) or rss_bytes <= 0: - raise ValueError("Linux process set contained unavailable sampled RSS") + raise ValueError("Linux process set contained invalid sampled RSS") if rss_bytes > MAX_U64 - total_rss_bytes: raise OverflowError("Linux process-set RSS exceeds u64 byte range") total_rss_bytes += rss_bytes From 015e4a5f79c0abee40c6807b481d3afce613c6c4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:35:32 +0900 Subject: [PATCH 10/14] test(browser): reject ambiguous sampled VmRSS evidence --- ...t_task_proc_snapshot_integrity_contract.py | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 tests/test_agent_task_proc_snapshot_integrity_contract.py diff --git a/tests/test_agent_task_proc_snapshot_integrity_contract.py b/tests/test_agent_task_proc_snapshot_integrity_contract.py new file mode 100644 index 00000000..4289296f --- /dev/null +++ b/tests/test_agent_task_proc_snapshot_integrity_contract.py @@ -0,0 +1,39 @@ +"""Integrity regressions for sampled Linux process evidence in the controlled browser fixture.""" + +from __future__ import annotations + +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class AgentTaskProcSnapshotIntegrityContractTests(unittest.TestCase): + """Keep nonresident processes distinct from malformed or ambiguous proc evidence.""" + + def test_optional_rss_parser_accepts_absent_or_zero_but_rejects_ambiguity(self) -> None: + """Only an unambiguous absent/zero VmRSS may mean no resident bytes.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_proc_integrity_contract") + parser = namespace["_parse_linux_proc_status_optional_rss_bytes"] + + self.assertIsNone(parser("Name:\tchrome\nPid:\t34\nPPid:\t12\n")) + self.assertIsNone(parser("Name:\tchrome\nVmRSS:\t0 kB\n")) + self.assertEqual(parser("Name:\tchrome\nVmRSS:\t123 kB\n"), 123 * 1024) + + for malformed in ( + "VmRSS:\t123 kB\nVmRSS:\t124 kB\n", + "VmRSS:\t0 kB\nVmRSS:\t124 kB\n", + "VmRSS:\t123 MB\n", + "VmRSS:\t123 kB extra\n", + "VmRSS:\tnot-a-number kB\n", + ): + with self.subTest(malformed=malformed): + with self.assertRaises(ValueError): + parser(malformed) + + +if __name__ == "__main__": + unittest.main() From ef6f23365f225b825505a58556d6917aeef505a2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 14:38:00 +0900 Subject: [PATCH 11/14] test(browser): restore green process-set evidence lane after RED probe --- ...t_task_proc_snapshot_integrity_contract.py | 39 ------------------- 1 file changed, 39 deletions(-) delete mode 100644 tests/test_agent_task_proc_snapshot_integrity_contract.py diff --git a/tests/test_agent_task_proc_snapshot_integrity_contract.py b/tests/test_agent_task_proc_snapshot_integrity_contract.py deleted file mode 100644 index 4289296f..00000000 --- a/tests/test_agent_task_proc_snapshot_integrity_contract.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Integrity regressions for sampled Linux process evidence in the controlled browser fixture.""" - -from __future__ import annotations - -import pathlib -import runpy -import unittest - -ROOT = pathlib.Path(__file__).resolve().parents[1] -RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" - - -class AgentTaskProcSnapshotIntegrityContractTests(unittest.TestCase): - """Keep nonresident processes distinct from malformed or ambiguous proc evidence.""" - - def test_optional_rss_parser_accepts_absent_or_zero_but_rejects_ambiguity(self) -> None: - """Only an unambiguous absent/zero VmRSS may mean no resident bytes.""" - - namespace = runpy.run_path(str(RUNNER), run_name="agent_task_proc_integrity_contract") - parser = namespace["_parse_linux_proc_status_optional_rss_bytes"] - - self.assertIsNone(parser("Name:\tchrome\nPid:\t34\nPPid:\t12\n")) - self.assertIsNone(parser("Name:\tchrome\nVmRSS:\t0 kB\n")) - self.assertEqual(parser("Name:\tchrome\nVmRSS:\t123 kB\n"), 123 * 1024) - - for malformed in ( - "VmRSS:\t123 kB\nVmRSS:\t124 kB\n", - "VmRSS:\t0 kB\nVmRSS:\t124 kB\n", - "VmRSS:\t123 MB\n", - "VmRSS:\t123 kB extra\n", - "VmRSS:\tnot-a-number kB\n", - ): - with self.subTest(malformed=malformed): - with self.assertRaises(ValueError): - parser(malformed) - - -if __name__ == "__main__": - unittest.main() From ceb1c72cf0f91ca8723bb5b3029044dae5d185b5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 15:08:04 +0900 Subject: [PATCH 12/14] test(browser): distinguish absent from ambiguous VmRSS --- .../test_agent_task_pinned_chrome_contract.py | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py index 7f8e91b7..df27f50b 100644 --- a/tests/test_agent_task_pinned_chrome_contract.py +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -95,6 +95,7 @@ def test_agent_task_records_bounded_chromium_process_tree_rss(self) -> None: "MAX_BROWSER_PROCESS_TREE_SIZE", "MAX_PROC_PROCESS_SCAN_SIZE", "_parse_linux_proc_status_process_identity", + "_parse_linux_proc_status_optional_rss_bytes", "_snapshot_linux_process_evidence", "_discover_linux_process_tree_ids", "_sample_linux_process_set_rss_bytes", @@ -141,6 +142,25 @@ def test_process_set_tolerates_descendant_without_resident_rss(self) -> None: } self.assertEqual(sample((10, 20, 30), evidence), 400) + def test_optional_linux_rss_parser_separates_absence_from_ambiguity(self) -> None: + """Snapshot parsing may tolerate absence, never malformed or duplicate VmRSS.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_optional_rss_contract") + parser = namespace["_parse_linux_proc_status_optional_rss_bytes"] + self.assertIsNone(parser("Name:\tchrome\n")) + self.assertIsNone(parser("Name:\tchrome\nVmRSS:\t0 kB\n")) + self.assertEqual(parser("Name:\tchrome\nVmRSS:\t123 kB\n"), 123 * 1024) + for malformed in ( + "VmRSS:\t123 MB\n", + "VmRSS:\t123 kB extra\n", + "VmRSS:\tnot-a-number kB\n", + "VmRSS:\t123 kB\nVmRSS:\t124 kB\n", + "VmRSS:\t18446744073709551616 kB\n", + ): + with self.subTest(malformed=malformed): + with self.assertRaises((ValueError, OverflowError)): + parser(malformed) + def test_linux_rss_parser_is_strict_and_overflow_safe(self) -> None: """Runner-side RSS evidence must not accept ambiguous proc status input.""" From e5fabfd57387ec7d2db692961eda93c95cf8d886 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 15:12:17 +0900 Subject: [PATCH 13/14] fix(browser): fail closed on ambiguous VmRSS evidence --- scripts/ci/run_mv3_compatibility.py | 33 ++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index bf224c2e..8240bf0c 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -78,7 +78,7 @@ def _path_token(value: str, label: str) -> str: def _webdriver_path(session_id: str, suffix: str) -> str: - """Build one bounded ChromeDriver path from a validated session identifier.""" + """Build a bounded ChromeDriver path from a validated session identifier.""" safe_session = _path_token(session_id, "session identifier") if suffix and not suffix.startswith("/"): @@ -230,6 +230,29 @@ def _parse_linux_proc_status_rss_bytes(status_text: str) -> int: return rss_values[0] +def _parse_linux_proc_status_optional_rss_bytes(status_text: str) -> int | None: + """Parse optional Linux ``VmRSS`` without normalizing malformed evidence.""" + + rss_lines = [line for line in status_text.splitlines() if line.startswith("VmRSS:")] + if not rss_lines: + return None + if len(rss_lines) != 1: + raise ValueError("Linux proc status must contain at most one VmRSS field") + + fields = rss_lines[0].split() + if len(fields) != 3 or fields[0] != "VmRSS:" or fields[2] != "kB": + raise ValueError("malformed Linux VmRSS field") + raw_kibibytes = fields[1] + if not raw_kibibytes.isascii() or not raw_kibibytes.isdigit(): + raise ValueError("malformed Linux VmRSS value") + kibibytes = int(raw_kibibytes, 10) + if kibibytes == 0: + return None + if kibibytes > MAX_U64 // 1024: + raise OverflowError("Linux VmRSS exceeds u64 byte range") + return kibibytes * 1024 + + def _parse_linux_proc_status_process_identity(status_text: str) -> tuple[int, int]: """Parse exactly one positive ``Pid`` and one non-negative ``PPid`` from status.""" @@ -305,13 +328,7 @@ def _snapshot_linux_process_evidence() -> dict[int, tuple[int, int | None]]: raise RuntimeError("Linux proc status identity did not match its directory") if process_id in process_evidence: raise RuntimeError("Linux proc process snapshot contained a duplicate PID") - try: - rss_bytes: int | None = _parse_linux_proc_status_rss_bytes(status_text) - except ValueError as exc: - if "VmRSS must be positive" in str(exc) or "exactly one VmRSS" in str(exc): - rss_bytes = None - else: - raise + rss_bytes = _parse_linux_proc_status_optional_rss_bytes(status_text) process_evidence[process_id] = (parent_process_id, rss_bytes) return process_evidence From 3ab127fef69096904df5db851f33824e7d868c0b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 23:50:27 +0900 Subject: [PATCH 14/14] chore(browser): sync hardened fixture contract --- tests/test_agent_task_fixture_contract.py | 66 +++++++++++++++++++---- 1 file changed, 57 insertions(+), 9 deletions(-) diff --git a/tests/test_agent_task_fixture_contract.py b/tests/test_agent_task_fixture_contract.py index 2a1465b2..2565a35c 100644 --- a/tests/test_agent_task_fixture_contract.py +++ b/tests/test_agent_task_fixture_contract.py @@ -10,6 +10,21 @@ FIXTURE = ROOT / "tests" / "fixtures" / "agent_task_basic" / "index.html" +def _is_credential_input(attributes: dict[str, str | None]) -> bool: + """Return whether parsed input attributes describe a credential surface.""" + + input_type = (attributes.get("type") or "").strip().lower() + if input_type == "password": + return True + + autocomplete = (attributes.get("autocomplete") or "").strip().lower() + autocomplete_tokens = autocomplete.split() + return any( + token == "one-time-code" or "password" in token + for token in autocomplete_tokens + ) + + class _FixtureParser(HTMLParser): """Collect the small semantic surface required by the deterministic fixture.""" @@ -18,6 +33,7 @@ def __init__(self) -> None: self.ids: set[str] = set() self.labels_for: set[str] = set() self.input_names: set[str] = set() + self.input_attributes: list[dict[str, str | None]] = [] self.button_types: set[str] = set() self.hidden_injection_markers = 0 @@ -30,12 +46,15 @@ def handle_starttag( self.ids.add(element_id) if tag == "label" and attributes.get("for"): self.labels_for.add(attributes["for"]) - if tag == "input" and attributes.get("name"): - self.input_names.add(attributes["name"]) + if tag == "input": + self.input_attributes.append(attributes) + if attributes.get("name"): + self.input_names.add(attributes["name"]) if tag == "button" and attributes.get("type"): self.button_types.add(attributes["type"]) if ( attributes.get("data-originweave-untrusted") == "prompt-injection" + and "hidden" in attributes and attributes.get("aria-hidden") == "true" ): self.hidden_injection_markers += 1 @@ -70,20 +89,49 @@ def test_fixture_contains_explicit_untrusted_hidden_prompt_injection(self) -> No self.assertIn("UNTRUSTED_PAGE_INSTRUCTION", self.html) self.assertIn("request new browser capabilities", self.html) + def test_hidden_injection_requires_the_actual_hidden_attribute(self) -> None: + """ARIA metadata alone must not satisfy the hidden-injection fixture contract.""" + + parser = _FixtureParser() + parser.feed( + "" + "" + ) + self.assertEqual(parser.hidden_injection_markers, 1) + def test_fixture_is_synthetic_and_has_no_credential_fields(self) -> None: """The controlled workflow must not require or imitate real secret collection.""" + for attributes in self.parser.input_attributes: + with self.subTest(attributes=attributes): + self.assertFalse(_is_credential_input(attributes)) + lowered = self.html.lower() - for forbidden in ( - 'type="password"', - 'autocomplete="current-password"', - 'autocomplete="one-time-code"', - "api_key", - "secret_key", - ): + for forbidden in ("api_key", "secret_key"): with self.subTest(forbidden=forbidden): self.assertNotIn(forbidden, lowered) + def test_credential_detection_is_quote_independent(self) -> None: + """Parsed credential semantics must reject single-quoted and tokenized forms.""" + + for html in ( + "", + "", + "", + "", + "", + ): + with self.subTest(html=html): + parser = _FixtureParser() + parser.feed(html) + self.assertEqual(len(parser.input_attributes), 1) + self.assertTrue(_is_credential_input(parser.input_attributes[0])) + + parser = _FixtureParser() + parser.feed("") + self.assertEqual(len(parser.input_attributes), 1) + self.assertFalse(_is_credential_input(parser.input_attributes[0])) + if __name__ == "__main__": unittest.main()