P23-03: 生活の観察

2026-10-16 14:41:54 +09:00
parent 447f77a13b
commit 7f1920b329
2 changed files with 262 additions and 19 deletions
@@ -183,6 +183,12 @@ class Repair28CompositionTests(unittest.TestCase):
            return self.files[revision][path]
        raise AssertionError(args)

+    def batch(self, oids):
+        self.calls.append(('cat-file', '--batch'))
+        blobs = {self.blob(v): v for files in self.files.values() for v in files.values()}
+        return io.BytesIO(b''.join(oid.encode()+b' blob '+str(len(blobs[oid])).encode()+b'\n'
+                                  +blobs[oid]+b'\n' for oid in oids)), 0
+
    def verify(self):
        original = Path.read_bytes
        def read(path):
@@ -190,7 +196,8 @@ class Repair28CompositionTests(unittest.TestCase):
                return self.files[self.t][str(path.relative_to(check.ROOT))]
            return original(path)
        with mock.patch.object(Path, 'read_bytes', read):
-            return check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,self.git)
+            return check.verify_composition(check.ROOT,self.roles,self.source,self.inputs,self.git,
+                                            read_batch=self.batch)

    def test_valid_route_b_keeps_unchanged_allowed_path_from_main(self):
        self.assertEqual(self.verify()['tested_tree'], self.t)
@@ -252,6 +259,145 @@ class Repair28CompositionTests(unittest.TestCase):
            self.roles, self.source = original_roles, original_source


+class Repair31StreamTests(unittest.TestCase):
+    def test_owned_child_cleanup_on_parse_failure_and_early_close(self):
+        oid = check.blob_oid(b'x')
+        for malformed in (True, False):
+            child = mock.Mock()
+            child.stdin = io.BytesIO()
+            child.stdout = io.BytesIO(b'bad\n' if malformed else oid.encode()+b' blob 1\nx\n')
+            child.poll.return_value = None
+            child.wait.return_value = -9
+            with mock.patch.object(subprocess, 'Popen', return_value=child):
+                stream = check.git_blob_stream(check.ROOT, [oid])
+                if malformed:
+                    with self.assertRaises(ValueError): next(stream)
+                else:
+                    self.assertEqual(next(stream), (oid, b'x'))
+                    stream.close()
+            child.kill.assert_called_once_with()
+            child.wait.assert_called_once_with()
+            self.assertTrue(child.stdin.closed)
+            self.assertTrue(child.stdout.closed)
+
+    def test_strict_stream_negatives_and_repeated_oids(self):
+        content = b'abc'
+        oid = check.blob_oid(content)
+        other = check.blob_oid(b'def')
+        record = oid.encode()+b' blob 3\nabc\n'
+        def read(data, oids=(oid,), status=0):
+            return list(check.git_blob_stream(check.ROOT, oids,
+                        lambda requested: (io.BytesIO(data), status)))
+        self.assertEqual(read(record*2, (oid, oid)), [(oid, content)]*2)
+        invalid = [record[:-2], record[:-1], record.replace(b'blob 3', b'blob 2'),
+                   record.replace(b'blob 3', b'blob 4'), record.replace(b'blob', b'tree'),
+                   oid.encode()+b' missing\n', oid.encode()+b' ambiguous\n',
+                   record+record, record.replace(b'abc', b'def'), b'',
+                   record.replace(oid.encode(), other.encode())]
+        for data in invalid:
+            with self.subTest(data=data), self.assertRaises(ValueError):
+                read(data)
+        with self.assertRaises(ValueError):
+            read(record, status=1)
+        with self.assertRaises(ValueError):
+            read(record*2, (oid, other))
+
+    def test_checkout_paths_and_modes(self):
+        oid = check.blob_oid(b'x')
+        roles = dict(tested_tree='1'*40)
+        for path, mode, kind, duplicate in [('a', '100644', 'blob', True),
+                ('a','120000','blob',False), ('a','160000','commit',False),
+                ('/a','100644','blob',False), ('a/../b','100644','blob',False)]:
+            record = f'{mode} {kind} {oid}\t{path}\0'.encode()
+            with self.subTest(path=path, mode=mode), self.assertRaises(ValueError):
+                with check.tested_checkout(check.ROOT, roles,
+                        lambda *args: record*(2 if duplicate else 1),
+                        lambda oids: (io.BytesIO(oid.encode()+b' blob 1\nx\n'),0)):
+                    self.fail('unsafe checkout yielded')
+
+    def test_view_authority_and_binding_recheck(self):
+        fixture = Repair28CompositionTests()
+        fixture.setUp()
+        def verify(view):
+            return check.verify_composition(check.ROOT, fixture.roles, fixture.source,
+                fixture.inputs, fixture.git, view=view, read_batch=fixture.batch)
+        with tempfile.TemporaryDirectory() as temporary:
+            with self.assertRaisesRegex(ValueError, 'unregistered'):
+                verify(Path(temporary))
+        with check.tested_checkout(check.ROOT, fixture.roles, fixture.git, fixture.batch) as view:
+            registered = check._VERIFIED_VIEWS[view.resolve()]
+            check._VERIFIED_VIEWS[view.resolve()] = ('0'*40, registered[1])
+            with self.assertRaisesRegex(ValueError, 'tree mismatch'): verify(view)
+            check._VERIFIED_VIEWS[view.resolve()] = (registered[0], {})
+            with self.assertRaisesRegex(ValueError, 'manifest mismatch'): verify(view)
+            check._VERIFIED_VIEWS[view.resolve()] = registered
+            # A normative document is outside source_refs; its bytes must still be checked.
+            (view/check.PLAN).write_bytes(b'tampered')
+            with mock.patch.object(Path, 'read_bytes', autospec=True) as reader:
+                # Preserve real view reads and synthesize only the source checkout comparison.
+                original = io.open
+                def read(path):
+                    if path.is_relative_to(view):
+                        with original(path, 'rb') as stream: return stream.read()
+                    return fixture.files[fixture.t][str(path.relative_to(check.ROOT))]
+                reader.side_effect = read
+                with self.assertRaisesRegex(ValueError, 'SHA-1 mismatch'): verify(view)
+        self.assertNotIn(view.resolve(), check._VERIFIED_VIEWS)
+
+    def test_constant_calls_full_composition_with_5000_paths(self):
+        fixture = Repair28CompositionTests()
+        fixture.setUp()
+        additions = {f'assets/bulk/{i:05}.txt': b'repeated small blob' for i in range(5000)}
+        for files in fixture.files.values(): files.update(additions)
+        fixture.source.update({p: hashlib.sha256(v).hexdigest() for p,v in additions.items()})
+        fixture.roles['blob_bindings'].update({p: check.blob_oid(v) for p,v in additions.items()})
+        original = Path.read_bytes
+        verified = set()
+        def read(path):
+            if path.is_relative_to(check.ROOT):
+                return fixture.files[fixture.t][str(path.relative_to(check.ROOT))]
+            data = original(path)
+            if 'bulk' in path.parts:
+                self.assertEqual(data, b'repeated small blob')
+                verified.add(path.name)
+            return data
+        with mock.patch.object(Path, 'read_bytes', read):
+            with check.tested_checkout(check.ROOT, fixture.roles, fixture.git, fixture.batch) as view:
+                check.verify_composition(check.ROOT, fixture.roles, fixture.source, fixture.inputs,
+                    fixture.git, view=view, read_batch=fixture.batch)
+        self.assertEqual(len(verified), 5000)
+        self.assertEqual(fixture.calls.count(('cat-file','--batch')), 1)
+        self.assertEqual(len([c for c in fixture.calls if c[0] == 'show']), 4)
+
+    def test_real_pipe_overflow_requests_and_repeated_paths(self):
+        runtime = os.environ.get('TMPDIR')
+        self.assertTrue(runtime, 'assigned TMPDIR required')
+        with tempfile.TemporaryDirectory(dir=runtime, prefix='p23-repair31-') as temporary:
+            repo = Path(temporary)
+            def git(*args, data=None):
+                return subprocess.run(['git','-C',str(repo),*args], input=data,
+                    capture_output=True, check=True, timeout=30).stdout
+            git('init','-q')
+            content = b'pipe pacing verified\n'
+            oid = git('hash-object','-w','--stdin', data=content).decode().strip()
+            count = 5000
+            tree = git('mktree', data=b''.join(
+                f'100644 blob {oid}\tpath-{i:05}\n'.encode() for i in range(count))).decode().strip()
+            # A watchdog bounds a deadlock regression; elapsed time is not a pass metric.
+            program = '''import sys
+from pathlib import Path
+sys.path.insert(0, sys.argv[1])
+import check_p23_life_observation_evidence as c
+repo, tree = Path(sys.argv[2]), sys.argv[3]
+with c.tested_checkout(repo, {'tested_tree': tree}) as view:
+    paths = list(view.iterdir())
+    assert len(paths) == 5000
+    assert all(p.read_bytes() == b'pipe pacing verified\\n' for p in paths)
+'''
+            subprocess.run([sys.executable, '-B', '-c', program, str(check.ROOT/'tools'),
+                            str(repo), tree], check=True, timeout=60)
+
+
 class Repair30EntryTests(unittest.TestCase):
    """Synthetic labelled Git history; zero native or OS credit."""
    def setUp(self):
@@ -274,6 +420,8 @@ class Repair30EntryTests(unittest.TestCase):
                     'scenes/main/world_preview.gd', 'tests/business/test_required_kitchen.gd',
                     'tools/regression/suite-policy.json'):
            self.write(path, 'synthetic baseline\n')
+        for i in range(220):
+            self.write(f'assets/repair31/{i:04}.txt', f'verified path {i}\n')
        self.marker = dict(schema_version=1, rows=[dict(test='history-'+str(i), required=True) for i in range(6)])
        self.write_marker(self.marker)
        self.base = self.commit('synthetic base')
@@ -357,6 +505,25 @@ class Repair30EntryTests(unittest.TestCase):
                with self.subTest(head=head, verify=verify):
                    self.assert_entry(verify)

+    def test_whole_entry_has_constant_git_calls_and_verifies_every_path(self):
+        spawn = subprocess.Popen
+        calls, read_paths = [], set()
+        original = Path.read_bytes
+        def counted(*args, **kwargs):
+            calls.append(args[0])
+            return spawn(*args, **kwargs)
+        def read(path):
+            data = original(path)
+            if 'repair31' in path.parts:
+                self.assertEqual(data, f'verified path {int(path.stem)}\n'.encode())
+                read_paths.add(path.name)
+            return data
+        with mock.patch.object(subprocess, 'Popen', counted), mock.patch.object(Path, 'read_bytes', read):
+            self.assert_entry()
+        self.assertEqual(len(read_paths), 220)
+        self.assertEqual(sum(command[-2:] == ['cat-file','--batch'] for command in calls), 1)
+        self.assertEqual(sum('show' in command for command in calls), 4)
+
    def test_wrong_roles_and_missing_Q(self):
        for key, value in (('submitted_sha',self.base), ('main_context_sha',self.base),
                           ('tested_tree',self.git('rev-parse', self.base+'^{tree}')),