tricu

A lazy functional language based on Tree Calculus
Log | Files | Refs | README | LICENSE

fixture_sources.py (8580B)


      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 import json
     10 from pathlib import Path, PurePosixPath
     11 
     12 ROOT = Path(__file__).resolve().parent.parent
     13 COHORTS = ('compiler-library', 'support-library')
     14 
     15 
     16 def digest(domain, payload):
     17     return hashlib.sha256(domain.encode() + b'\0' + payload).hexdigest()
     18 
     19 
     20 def source_rows(fixtures):
     21     rows, paths, aliases = [], set(), set()
     22     for cohort in COHORTS:
     23         lines = (fixtures / (cohort + '.v1')).read_text().splitlines()
     24         versions = {'tricu.' + cohort + '-fixture.v1'}
     25         if cohort == 'support-library':
     26             versions.update(('tricu.support-library-fixture.v2',
     27                              'tricu.support-library-fixture.v3', 'tricu.support-library-fixture.v4',
     28                              'tricu.support-library-fixture.v5', 'tricu.support-library-fixture.v6',
     29                              'tricu.support-library-fixture.v7'))
     30         if not lines or lines[0] not in versions:
     31             raise ValueError('unsupported fixture metadata: ' + cohort)
     32         bundle = (fixtures / (cohort + '.arboricx')).read_bytes()
     33         if 'bundle\t' + digest('tricu.' + cohort + '-fixture.v1', bundle) not in lines:
     34             raise ValueError('fixture bundle digest mismatch: ' + cohort)
     35         count = 0
     36         for line in lines[1:]:
     37             if not line.startswith('module\t'):
     38                 continue
     39             fields = line.split('\t')
     40             if len(fields) != 5:
     41                 raise ValueError('invalid fixture module row')
     42             _, legacy, alias, path, pin = fields
     43             parts = PurePosixPath(path)
     44             if (parts.is_absolute() or '..' in parts.parts or str(parts) != path
     45                     or not path or len(pin) != 64 or any(c not in '0123456789abcdef' for c in pin)):
     46                 raise ValueError('invalid fixture source path or digest')
     47             if path in paths or alias in aliases:
     48                 raise ValueError('duplicate fixture source or alias')
     49             paths.add(path)
     50             aliases.add(alias)
     51             rows.append((cohort, legacy, alias, path, pin))
     52             count += 1
     53         if not count:
     54             raise ValueError('empty fixture source cohort')
     55     return rows
     56 
     57 
     58 def snapshot_bytes(fixtures, pin):
     59     payload = (fixtures / 'reference-source' / (pin + '.tri')).read_bytes()
     60     if digest('arboricx.source.v1', payload) != pin:
     61         raise ValueError('reference source digest mismatch: ' + pin)
     62     return payload
     63 
     64 
     65 def verify(fixtures):
     66     rows = source_rows(fixtures)
     67     for *_, pin in rows:
     68         snapshot_bytes(fixtures, pin)
     69     return rows
     70 
     71 
     72 def capture(fixtures, source_root):
     73     # Validate the entire selected cohort before writing anything. No fallback to
     74     # another checkout; do not rewrite a corrupt existing snapshot to hide it.
     75     rows = source_rows(fixtures)
     76     payloads = {}
     77     for _, _, _, path, pin in rows:
     78         payload = (source_root / path).read_bytes()
     79         if digest('arboricx.source.v1', payload) != pin:
     80             raise ValueError('source does not match fixture: ' + path)
     81         destination = fixtures / 'reference-source' / (pin + '.tri')
     82         if destination.exists() and destination.read_bytes() != payload:
     83             raise ValueError('existing reference source differs: ' + pin)
     84         payloads[destination] = payload
     85     for destination, payload in payloads.items():
     86         destination.parent.mkdir(parents=True, exist_ok=True)
     87         if not destination.exists():
     88             with destination.open('xb') as output:
     89                 output.write(payload)
     90     return rows
     91 
     92 
     93 def materialize(fixtures, destination):
     94     """Create a reference-only source tree, never bless current working bytes.
     95 
     96     Module bytes come exclusively from authenticated fixture snapshots. The map
     97     is reconstructed from recorded membership and those sources' native headers;
     98     this is not a claim that the result is an original Git checkout or accepted
     99     compiler generation. Validate everything before creating the destination.
    100     """
    101     from native_header import native_header
    102     rows = verify(fixtures)
    103     payloads = {path: snapshot_bytes(fixtures, pin) for _, _, _, path, pin in rows}
    104     mapping = ['arboricx.source-map.v1']
    105     compiler_aliases = {alias for cohort, _, alias, _, _ in rows if cohort == 'compiler-library'}
    106     for cohort, legacy, alias, path, _ in rows:
    107         if cohort == 'compiler-library':
    108             dependencies, _ = native_header(payloads[path].decode('utf-8'))
    109             if not set(dependencies) <= compiler_aliases:
    110                 raise ValueError('reference compiler dependency outside cohort')
    111             mapping.append('\t'.join(['native', alias, legacy, path, ','.join(dependencies) or '-']))
    112     destination.mkdir(parents=True, exist_ok=False)
    113     for path, payload in payloads.items():
    114         target = destination / path
    115         target.parent.mkdir(parents=True, exist_ok=True)
    116         target.write_bytes(payload)
    117     (destination / 'bootstrap').mkdir(exist_ok=True)
    118     (destination / 'bootstrap/compiler-source-map.v1').write_text('\n'.join(mapping)+'\n')
    119     (destination / 'reference-sources.json').write_text(json.dumps(dict(
    120         schema='tricu.fixture-reference-sources.v1', scope='reference-only-not-working-tree',
    121         sources={path: pin for _, _, _, path, pin in rows}), indent=2)+'\n')
    122     return rows
    123 
    124 
    125 def current_mismatches(fixtures, source_root):
    126     rows = verify(fixtures)
    127     changed = []
    128     for _, _, _, path, pin in rows:
    129         current = source_root / path
    130         if not current.is_file() or digest('arboricx.source.v1', current.read_bytes()) != pin:
    131             changed.append(path)
    132     # Source membership/order is part of the compiler input, too. Dependencies
    133     # are checked by the generation gate and by each source's native header.
    134     map_path = source_root / 'bootstrap/compiler-source-map.v1'
    135     actual = []
    136     if map_path.is_file():
    137         lines = map_path.read_text().splitlines()
    138         if lines[:1] == ['arboricx.source-map.v1']:
    139             for line in lines[1:]:
    140                 fields = line.split('\t')
    141                 if len(fields) != 5 or fields[0] != 'native':
    142                     break
    143                 _, alias, legacy, path, _ = fields
    144                 actual.append((legacy, alias, path))
    145             else:
    146                 if actual == [(legacy, alias, path) for cohort, legacy, alias, path, _ in rows
    147                               if cohort == 'compiler-library']:
    148                     return changed
    149     return changed + ['bootstrap/compiler-source-map.v1']
    150 
    151 
    152 def main():
    153     parser = argparse.ArgumentParser(description=__doc__)
    154     parser.add_argument('mode', choices=('verify', 'capture', 'current', 'materialize'))
    155     parser.add_argument('--fixtures', type=Path, default=ROOT / 'test/fixtures')
    156     parser.add_argument('--source-root', type=Path,
    157                         help='required for capture; current defaults to the checkout')
    158     args = parser.parse_args()
    159     try:
    160         if args.mode == 'materialize':
    161             if args.source_root is None:
    162                 parser.error('materialize requires a new --source-root directory')
    163             rows = materialize(args.fixtures, args.source_root)
    164             print(f'Materialized {len(rows)} authenticated reference sources; working-tree freshness NOT checked.')
    165         elif args.mode == 'capture':
    166             if args.source_root is None:
    167                 parser.error('capture requires an explicit --source-root')
    168             rows = capture(args.fixtures, args.source_root)
    169             print(f'Captured {len(rows)} authenticated reference sources; artifacts unchanged.')
    170         elif args.mode == 'current':
    171             changed = current_mismatches(args.fixtures, args.source_root or ROOT)
    172             if changed:
    173                 parser.exit(1, 'Fixture freshness pending (reference tests do not cover these working inputs):\n'
    174                             + '\n'.join(changed) + '\n')
    175             print('Fixture sources and compiler map match the working inputs; not generation acceptance.')
    176         else:
    177             rows = verify(args.fixtures)
    178             print(f'Authenticated {len(rows)} reference sources; working-tree freshness NOT checked.')
    179     except (ValueError, OSError) as error:
    180         parser.exit(1, str(error) + '\n')
    181 
    182 
    183 if __name__ == '__main__':
    184     main()