File size: 7,137 Bytes
4b8c78e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
"""Small regression checks for release utilities, without assets or GPUs."""
from __future__ import annotations
import copy
import csv
import hashlib
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'scripts'))
from _common import write_solid_png, write_npy_f4


class ReleaseToolsTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.work = Path(self.temp.name)
        self.rows = [json.loads(s) for s in (ROOT / 'examples/metadata/candidates.jsonl').read_text().splitlines() if s]
        self.candidates = self.work / 'candidates.jsonl'
        self.selection = self.work / 'selection.json'
        self.save_candidates(self.rows)
        self.selection.write_text(json.dumps({'scene_id': 'tiny_blend_scene', 'selected_viewpoints': [{'candidate_id': 'tiny_000', 'rank': 0, 'position': [0, 1.6, 0]}]}))

    def save_candidates(self, rows):
        self.candidates.write_text(''.join(json.dumps(r) + '\n' for r in rows))

    def run_script(self, relative, *args, ok=True, cwd=None):
        result = subprocess.run([sys.executable, str(ROOT / relative), *map(str, args)], cwd=cwd or ROOT, capture_output=True, text=True)
        if ok:
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        else:
            self.assertNotEqual(result.returncode, 0, 'Invalid input was accepted')
        return result

    def select(self, *extra, ok=True):
        return self.run_script('scripts/select_views.py', '--candidates', self.candidates, '--output-dir', self.work / 'run', *extra, ok=ok)

    def test_selector_rejects_duplicate_ids(self):
        self.save_candidates(self.rows + [self.rows[0]])
        self.assertIn('Duplicate candidate_id', self.select(ok=False).stderr)

    def test_selector_rejects_mixed_scenes(self):
        self.rows[1]['scene_id'] = 'another_scene'
        self.save_candidates(self.rows)
        self.assertIn('one source and scene', self.select(ok=False).stderr)

    def test_selector_rejects_invalid_boolean_and_nonfinite_priors(self):
        for field, value in [('valid', 'false'), ('conflict_prior', float('nan'))]:
            with self.subTest(field=field):
                rows = copy.deepcopy(self.rows)
                rows[0][field] = value
                self.save_candidates(rows)
                self.select(ok=False)

    def test_selector_rejects_invalid_parameters(self):
        for args in [('--budget', '-1'), ('--lambda-conflict', 'nan'), ('--min-gain', '1.5')]:
            with self.subTest(args=args):
                self.select(*args, ok=False)

    def test_coverage_rejects_unknown_invalid_and_duplicate_selections(self):
        for ids in [['missing'], ['tiny_005'], ['tiny_000', 'tiny_000']]:
            with self.subTest(ids=ids):
                self.selection.write_text(json.dumps({'scene_id': 'tiny_blend_scene', 'selected_viewpoints': [{'candidate_id': x} for x in ids]}))
                out = self.work / 'coverage.csv'
                self.run_script('scripts/evaluate_coverage.py', '--candidates', self.candidates, '--selected', self.selection, '--output', out, ok=False)
                self.assertFalse(out.exists())

    def make_renders(self):
        d = self.work / 'renders'
        d.mkdir()
        write_solid_png(d / '000_tiny_000_rgb.png', 8, 4, (20, 40, 60))
        write_npy_f4(d / '000_tiny_000_depth.npy', 4, 8, 1.0)
        (d / '000_tiny_000_pose.json').write_text(json.dumps({'candidate_id': 'tiny_000', 'position': [0, 1.6, 0]}))
        return d

    def audit(self, d, ok=True):
        out = self.work / 'audit.csv'
        self.run_script('scripts/audit_quality.py', '--render-dir', d, '--metadata', self.selection, '--output', out, ok=ok)
        with out.open() as stream:
            return list(csv.DictReader(stream))[0]

    def test_audit_accepts_valid_example(self):
        self.assertEqual(self.audit(self.make_renders())['passed'], 'True')

    def test_audit_rejects_corrupt_file_even_if_it_exists(self):
        d = self.make_renders()
        (d / '000_tiny_000_rgb.png').write_bytes(b'not a PNG')
        row = self.audit(d, ok=False)
        self.assertEqual(row['rgb_exists'], 'True')
        self.assertEqual(row['rgb_png_magic'], 'False')
        self.assertEqual(row['passed'], 'False')

    def test_audit_rejects_mismatch_truncation_and_scalar_pose(self):
        d = self.make_renders()
        depth = d / '000_tiny_000_depth.npy'
        write_npy_f4(depth, 2, 8, 1.0)
        self.assertEqual(self.audit(d, ok=False)['rgb_depth_shape_match'], 'False')
        write_npy_f4(depth, 4, 8, 1.0)
        depth.write_bytes(depth.read_bytes()[:-4])
        self.assertEqual(self.audit(d, ok=False)['depth_array_valid'], 'False')
        write_npy_f4(depth, 4, 8, 1.0)
        (d / '000_tiny_000_pose.json').write_text('42')
        self.assertEqual(self.audit(d, ok=False)['pose_json_valid'], 'False')

    def test_manifest_relative_root_spaces_and_generated_outputs(self):
        root = self.work / 'release space'
        root.mkdir()
        (root / 'a file.txt').write_text('content')
        (root / 'outputs').mkdir()
        (root / 'outputs' / 'generated.txt').write_text('not distributed')
        tool = 'tools/release_integrity.py'
        self.run_script(tool, 'generate', 'release space', cwd=self.work)
        manifest = root / 'SHA256SUMS'
        self.assertEqual(len(manifest.read_text().splitlines()), 1)
        self.run_script(tool, 'verify', 'release space', cwd=self.work)
        (root / 'a file.txt').write_text('modified')
        self.assertIn('Hash mismatch', self.run_script(tool, 'verify', root, ok=False).stderr)
        (root / 'new.py').write_text('print(1)')
        self.assertIn('Unlisted file', self.run_script(tool, 'verify', root, ok=False).stderr)

    def test_hash_update_uses_actual_bytes_and_rejects_inventory_mode(self):
        doc = self.work / 'metadata.json'
        out = self.work / 'updated.json'
        resource = self.work / 'payload.csv'
        resource.write_text('real bytes\n')
        original = {'distribution': [{'@id': 'frame-manifest', '@type': 'cr:FileObject', 'contentUrl': 'https://example.org/payload.csv', 'sha256': '0' * 64}]}
        doc.write_text(json.dumps(original))
        tool = 'tools/update_croissant_with_real_hashes.py'
        self.run_script(tool, '--croissant', doc, '--out', out, '--resource', 'frame-manifest=' + str(resource))
        self.assertEqual(json.loads(out.read_text())['distribution'][0]['sha256'], hashlib.sha256(resource.read_bytes()).hexdigest())
        before = out.read_bytes()
        self.run_script(tool, '--croissant', doc, '--out', out, '--digests', resource, ok=False)
        self.run_script(tool, '--croissant', doc, '--out', out, '--resource', 'missing=' + str(resource), ok=False)
        self.assertEqual(out.read_bytes(), before)
        self.assertEqual(json.loads(doc.read_text()), original)


if __name__ == '__main__':
    unittest.main()