tricu

An interpreted language for exploring Tree Calculus
Log | Files | Refs | README | LICENSE

fixture_sources.py (6228B)


      1 #!/usr/bin/env python3
      2 """Authenticate reference sources, or explicitly check working-tree freshness.
      3 
      4 No compilation and no metadata rewriting. Capture accepts only bytes matching
      5 existing fixture metadata; it cannot bless an edited source as an old artifact.
      6 """
      7 import argparse
      8 import hashlib
      9 from pathlib import Path, PurePosixPath
     10 
     11 ROOT = Path(__file__).resolve().parent.parent
     12 COHORTS = ('compiler-library', 'support-library')
     13 
     14 
     15 def digest(domain, payload):
     16     return hashlib.sha256(domain.encode() + b'\0' + payload).hexdigest()
     17 
     18 
     19 def source_rows(fixtures):
     20     rows, paths, aliases = [], set(), set()
     21     for cohort in COHORTS:
     22         lines = (fixtures / (cohort + '.v1')).read_text().splitlines()
     23         versions = {'tricu.' + cohort + '-fixture.v1'}
     24         if cohort == 'support-library':
     25             versions.update(('tricu.support-library-fixture.v2',
     26                              'tricu.support-library-fixture.v3'))
     27         if not lines or lines[0] not in versions:
     28             raise ValueError('unsupported fixture metadata: ' + cohort)
     29         bundle = (fixtures / (cohort + '.arboricx')).read_bytes()
     30         if 'bundle\t' + digest('tricu.' + cohort + '-fixture.v1', bundle) not in lines:
     31             raise ValueError('fixture bundle digest mismatch: ' + cohort)
     32         count = 0
     33         for line in lines[1:]:
     34             if not line.startswith('module\t'):
     35                 continue
     36             fields = line.split('\t')
     37             if len(fields) != 5:
     38                 raise ValueError('invalid fixture module row')
     39             _, legacy, alias, path, pin = fields
     40             parts = PurePosixPath(path)
     41             if (parts.is_absolute() or '..' in parts.parts or str(parts) != path
     42                     or not path or len(pin) != 64 or any(c not in '0123456789abcdef' for c in pin)):
     43                 raise ValueError('invalid fixture source path or digest')
     44             if path in paths or alias in aliases:
     45                 raise ValueError('duplicate fixture source or alias')
     46             paths.add(path)
     47             aliases.add(alias)
     48             rows.append((cohort, legacy, alias, path, pin))
     49             count += 1
     50         if not count:
     51             raise ValueError('empty fixture source cohort')
     52     return rows
     53 
     54 
     55 def snapshot_bytes(fixtures, pin):
     56     payload = (fixtures / 'reference-source' / (pin + '.tri')).read_bytes()
     57     if digest('arboricx.source.v1', payload) != pin:
     58         raise ValueError('reference source digest mismatch: ' + pin)
     59     return payload
     60 
     61 
     62 def verify(fixtures):
     63     rows = source_rows(fixtures)
     64     for *_, pin in rows:
     65         snapshot_bytes(fixtures, pin)
     66     return rows
     67 
     68 
     69 def capture(fixtures, source_root):
     70     # Validate the entire selected cohort before writing anything. No fallback to
     71     # another checkout; do not rewrite a corrupt existing snapshot to hide it.
     72     rows = source_rows(fixtures)
     73     payloads = {}
     74     for _, _, _, path, pin in rows:
     75         payload = (source_root / path).read_bytes()
     76         if digest('arboricx.source.v1', payload) != pin:
     77             raise ValueError('source does not match fixture: ' + path)
     78         destination = fixtures / 'reference-source' / (pin + '.tri')
     79         if destination.exists() and destination.read_bytes() != payload:
     80             raise ValueError('existing reference source differs: ' + pin)
     81         payloads[destination] = payload
     82     for destination, payload in payloads.items():
     83         destination.parent.mkdir(parents=True, exist_ok=True)
     84         if not destination.exists():
     85             with destination.open('xb') as output:
     86                 output.write(payload)
     87     return rows
     88 
     89 
     90 def current_mismatches(fixtures, source_root):
     91     rows = verify(fixtures)
     92     changed = []
     93     for _, _, _, path, pin in rows:
     94         current = source_root / path
     95         if not current.is_file() or digest('arboricx.source.v1', current.read_bytes()) != pin:
     96             changed.append(path)
     97     # Source membership/order is part of the compiler input, too. Dependencies
     98     # are checked by the generation gate and by each source's native header.
     99     map_path = source_root / 'bootstrap/compiler-source-map.v1'
    100     actual = []
    101     if map_path.is_file():
    102         lines = map_path.read_text().splitlines()
    103         if lines[:1] == ['arboricx.source-map.v1']:
    104             for line in lines[1:]:
    105                 fields = line.split('\t')
    106                 if len(fields) != 5 or fields[0] != 'native':
    107                     break
    108                 _, alias, legacy, path, _ = fields
    109                 actual.append((legacy, alias, path))
    110             else:
    111                 if actual == [(legacy, alias, path) for cohort, legacy, alias, path, _ in rows
    112                               if cohort == 'compiler-library']:
    113                     return changed
    114     return changed + ['bootstrap/compiler-source-map.v1']
    115 
    116 
    117 def main():
    118     parser = argparse.ArgumentParser(description=__doc__)
    119     parser.add_argument('mode', choices=('verify', 'capture', 'current'))
    120     parser.add_argument('--fixtures', type=Path, default=ROOT / 'test/fixtures')
    121     parser.add_argument('--source-root', type=Path,
    122                         help='required for capture; current defaults to the checkout')
    123     args = parser.parse_args()
    124     try:
    125         if args.mode == 'capture':
    126             if args.source_root is None:
    127                 parser.error('capture requires an explicit --source-root')
    128             rows = capture(args.fixtures, args.source_root)
    129             print(f'Captured {len(rows)} authenticated reference sources; artifacts unchanged.')
    130         elif args.mode == 'current':
    131             changed = current_mismatches(args.fixtures, args.source_root or ROOT)
    132             if changed:
    133                 parser.exit(1, 'Fixture freshness pending (reference tests do not cover these working inputs):\n'
    134                             + '\n'.join(changed) + '\n')
    135             print('Fixture sources and compiler map match the working inputs; not generation acceptance.')
    136         else:
    137             rows = verify(args.fixtures)
    138             print(f'Authenticated {len(rows)} reference sources; working-tree freshness NOT checked.')
    139     except (ValueError, OSError) as error:
    140         parser.exit(1, str(error) + '\n')
    141 
    142 
    143 if __name__ == '__main__':
    144     main()