Digital Drosophila Archive
135 lines
6.6 KiB
Python
135 lines
6.6 KiB
Python
"""Validate staged integration against both original trees without science runs."""
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
def require(value, message):
|
|
if not value:
|
|
raise ValueError(message)
|
|
|
|
|
|
def git(*args):
|
|
return subprocess.check_output(['git', *args])
|
|
|
|
|
|
def tree(ref):
|
|
result = {}
|
|
for row in git('ls-tree', '-r', '-z', ref).split(b'\0'):
|
|
if row:
|
|
meta, path = row.split(b'\t', 1)
|
|
mode, kind, oid = meta.decode().split()
|
|
require(kind == 'blob', 'Unexpected tree object')
|
|
result[path.decode()] = (mode, oid)
|
|
return result
|
|
|
|
|
|
def main(output, check_lfs):
|
|
root = Path.cwd()
|
|
require(root == Path('/home/qishizhe/digital-drosophila-main-integration-20261001'), 'Independent integration clone required')
|
|
plan = json.loads((root / 'registry/branch-consolidation-20261001/plan.json').read_text())
|
|
left, right = plan['merge_parents']
|
|
base = git('merge-base', left, right).decode().strip()
|
|
a, b, old = tree(left), tree(right), tree(base)
|
|
staged = {}
|
|
for row in git('ls-files', '--stage', '-z').split(b'\0'):
|
|
if row:
|
|
meta, path = row.split(b'\t', 1)
|
|
mode, oid, stage = meta.decode().split()
|
|
require(stage == '0', 'Unresolved merge conflict')
|
|
staged[path.decode()] = (mode, oid)
|
|
allowed = set(plan['allowed_existing_file_changes'])
|
|
checked = 0
|
|
for name in sorted(set(a) | set(b)):
|
|
require(name in staged, 'Original file removed: ' + name)
|
|
if name in allowed:
|
|
continue
|
|
av, bv, ov = a.get(name), b.get(name), old.get(name)
|
|
if av == bv:
|
|
expected = av
|
|
elif av == ov:
|
|
expected = bv
|
|
elif bv == ov:
|
|
expected = av
|
|
elif av is None:
|
|
expected = bv
|
|
elif bv is None:
|
|
expected = av
|
|
else:
|
|
raise ValueError('Unapproved conflicting original file: ' + name)
|
|
require(staged[name] == expected, 'Original blob or mode changed: ' + name)
|
|
checked += 1
|
|
new_paths = sorted(set(staged) - set(a) - set(b))
|
|
for name in new_paths:
|
|
require(any(name.startswith(p) if p.endswith('/') else name == p for p in plan['new_paths']), 'Unapproved new path')
|
|
artifacts = {p: v for p, v in staged.items() if p.startswith('artifacts/')}
|
|
require(len(artifacts) == plan['expected_artifact_paths'], 'Artifact path inventory differs')
|
|
pointers = {}
|
|
pattern = re.compile(rb'version https://git-lfs.github.com/spec/v1\noid sha256:([0-9a-f]{64})\nsize ([0-9]+)\n?')
|
|
for name, (mode, oid) in artifacts.items():
|
|
data = git('cat-file', 'blob', oid)
|
|
match = pattern.fullmatch(data)
|
|
if match:
|
|
pointers[name] = dict(oid=match[1].decode(), bytes=int(match[2]), git_blob=oid)
|
|
require(len(pointers) == plan['expected_lfs_pointer_paths'], 'LFS pointer path inventory differs')
|
|
attributes = git('check-attr', '--cached', 'filter', '--', *pointers).decode().splitlines()
|
|
require(len(attributes) == len(pointers) and all(x.endswith(': filter: lfs') for x in attributes), 'Merged LFS rules do not cover every original object')
|
|
lfs_report = None
|
|
if check_lfs:
|
|
source = root / 'tools/lfs-publication/lan_lfs_transfer.py'
|
|
require(hashlib.sha256(source.read_bytes()).digest() == hashlib.sha256(git('show', left + ':tools/lfs-publication/lan_lfs_transfer.py')).digest(), 'Transfer adapter changed')
|
|
spec = importlib.util.spec_from_file_location('merge_lan', source)
|
|
lan = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(lan)
|
|
agent = lan.Agent(dict(event='init', operation='download', remote='origin', concurrent=True, concurrenttransfers=2), lambda _: None)
|
|
href, headers = agent.authenticate()
|
|
unique = {v['oid']: v['bytes'] for v in pointers.values()}
|
|
requested = [{'oid': oid, 'size': size} for oid, size in sorted(unique.items())]
|
|
batches = 0
|
|
for begin in range(0, len(requested), 100):
|
|
items = requested[begin:begin + 100]
|
|
response = lan.decode_json(agent.curl('POST', href + '/objects/batch', headers,
|
|
body=dict(operation='download', transfers=['basic'], objects=items)))
|
|
require(response.get('transfer', 'basic') == 'basic', 'Unexpected LFS transfer mode')
|
|
received = response.get('objects', [])
|
|
require(len(received) == len(items), 'LFS metadata inventory incomplete')
|
|
seen = set()
|
|
for item in received:
|
|
oid = item.get('oid')
|
|
require(oid in unique and oid not in seen and item.get('size') == unique[oid] and 'error' not in item, 'LFS object unavailable')
|
|
action = item.get('actions', {}).get('download')
|
|
require(isinstance(action, dict), 'LFS object has no download action')
|
|
lan.safe_url(action['href'])
|
|
lan.safe_headers(action.get('header', {}))
|
|
seen.add(oid)
|
|
require(seen == {item['oid'] for item in items}, 'LFS response object set differs')
|
|
batches += 1
|
|
lfs_report = dict(unique_objects=len(unique), unique_bytes=sum(unique.values()), batches=batches,
|
|
all_download_actions_available=True, payloads_redownloaded=False,
|
|
scope='Authenticated read-only LFS availability; full byte validation remains the unchanged prior fresh-download receipts')
|
|
report = dict(schema='main-integration-audit-v1', verified_utc=datetime.now(timezone.utc).isoformat(),
|
|
pass_all=True, parents=[left, right], merge_base=base, staged_original_blobs_checked=checked,
|
|
allowed_existing_changes=sorted(allowed), new_paths=new_paths, artifact_paths=len(artifacts),
|
|
lfs_pointer_paths=len(pointers), pointers=pointers, lfs_availability=lfs_report,
|
|
model_tool_registry_artifact_bytes_unchanged=True, scientific_execution=False,
|
|
source_worktree_write_operations_by_this_tool=False,
|
|
source_worktree_integrity_requires_separate_audit=True,
|
|
staged_tree=git('write-tree').decode().strip())
|
|
with output.open('x') as stream:
|
|
json.dump(report, stream, indent=2, sort_keys=True)
|
|
stream.write('\n')
|
|
print(json.dumps({k: v for k, v in report.items() if k not in ('pointers', 'new_paths')}, indent=2))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
parser.add_argument('--check-lfs', action='store_true')
|
|
args = parser.parse_args()
|
|
main(args.output, args.check_lfs)
|