`verify_cli.py`

85 lines
"""Integration proof: real CLI subprocesses with retained input/output evidence."""
import copy
import datetime
import platform
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent
CASES = Path(tempfile.mkdtemp(prefix='cli-', dir=ROOT / 'verification'))
HOOK = "import sys,runpy; sys.addaudithook(lambda e,a: (_ for _ in ()).throw(RuntimeError('SIDE_EFFECT_BLOCKED')) if e.startswith('socket.') or e in ('subprocess.Popen','os.system') else None); sys.argv=sys.argv[1:]; runpy.run_path(sys.argv[0],run_name='__main__')"
results = []


def run(name, bundle, pin=None, expected=0, fmt='json'):
    cmd = [sys.executable, '-c', HOOK, str(ROOT/'review.py'), '--bundle', str(bundle), '--format', fmt]
    if pin:
        cmd += ['--manifest-sha256', pin]
    result = subprocess.run(cmd, text=True, capture_output=True)
    (CASES/(name+'.stdout')).write_text(result.stdout)
    (CASES/(name+'.stderr')).write_text(result.stderr)
    assert result.returncode == expected, (name, result.returncode, result.stderr)
    if expected:
        assert not result.stdout and 'REVIEW ONLY' in result.stderr
    results.append({'case': name, 'exit': result.returncode, 'verdict': 'PASS'})
    return json.loads(result.stdout) if expected == 0 and fmt == 'json' else result.stdout


def fixture(name, mutate=None, malformed=False, duplicate_key=False):
    b = CASES/name; b.mkdir()
    scope = [{'gid':'gid://shopify/Product/1','status':'DRAFT','old_title':'Digital Silver Mylar | Bespoke','old_body':'silver mylar and mylarite'}]
    plan = [{'gid':scope[0]['gid'],'status':'DRAFT','changes':{'title':{'from':scope[0]['old_title'],'to':'Digital Silver Metallic | Bespoke'}}}]
    if mutate:
        mutate(scope, plan)
    for n, value in [('scope.json', scope), ('plan.json', plan)]:
        raw = json.dumps(value).encode()
        if n == 'scope.json' and malformed: raw = b'{'
        if n == 'scope.json' and duplicate_key: raw = raw.replace(b'"status": "DRAFT"', b'"status":"DRAFT","status":"ACTIVE"')
        (b/n).write_bytes(raw)
    m={'kind':'HISTORICAL_SNAPSHOT','expected_rows':len(scope),'files':{n:{'sha256':hashlib.sha256((b/n).read_bytes()).hexdigest()} for n in ['scope.json','plan.json']}}
    raw=json.dumps(m).encode();(b/'manifest.json').write_bytes(raw)
    return b, hashlib.sha256(raw).hexdigest()

actual=CASES/'actual-copy';shutil.copytree(ROOT/'snapshots',actual)
r=run('historical',actual)
assert r['snapshot_completeness']['reviewed']==91 and len({x['gid'] for x in r['rows']})==91
assert r['counts']=={'PROVISIONAL_CANDIDATE':5,'HELD_AMBIGUOUS':5,'EXCLUDED_NO_TITLE_SIGNALS':81}
assert all(not x['apply_eligible'] and x['vendor_identity']=='UNVERIFIED' for x in r['rows'])
assert all(x['historical_status']=='ARCHIVED' for x in r['rows'] if x['disposition']=='PROVISIONAL_CANDIDATE')
md=run('markdown',actual,fmt='markdown');assert md.count('## gid://shopify/Product/')==91
assert run('repeat',actual)==r
b,p=fixture('positive');rr=run('positive',b,p)
assert rr['rows'][0]['review_comparison']['descriptionHtml']['after_provisional']=='silver metallic and mylarite'
for name, mutate in [
 ('duplicate_id',lambda s,p:(s.append(copy.deepcopy(s[0])),p.append(copy.deepcopy(p[0])))),
 ('id_mismatch',lambda s,p:p[0].update(gid='gid://shopify/Product/2')),
 ('status_mismatch',lambda s,p:p[0].update(status='ACTIVE')),
 ('before_mismatch',lambda s,p:p[0]['changes']['title'].update({'from':'wrong'})),
 ('unexpected_field',lambda s,p:p[0]['changes'].update({'vendor':{'from':'x','to':'y'}})),
 ('invalid_text',lambda s,p:s[0].update(old_body=None)),
 ('count_mismatch',lambda s,p:p.clear())]:
 b,p=fixture(name,mutate);run(name,b,p,2)
b,p=fixture('malformed',malformed=True);run('malformed',b,p,2)
b,p=fixture('duplicate_key',duplicate_key=True);run('duplicate_key',b,p,2)
b,p=fixture('tamper');(b/'scope.json').write_text('[]');run('tamper',b,p,2)
run('bad_manifest_pin',actual,'0'*64,2)
for name,title,body,expect in [
 ('body_only','Silver Mylar | Bespoke','digital printing','HELD_AMBIGUOUS'),
 ('digital_only','Digital Mylar | Other','bespoke','HELD_AMBIGUOUS'),
 ('substring','Digitallymade Mylar | Bespoken','digital bespoke','EXCLUDED_NO_TITLE_SIGNALS'),
 ('digitally','Digitally Print Mylar | Bespoke','mylar','PROVISIONAL_CANDIDATE')]:
 def mutate(s,p):
  s[0].update(old_title=title,old_body=body);p[0]['changes']['title']['from']=title
 b,p=fixture(name,mutate);assert run(name,b,p)['rows'][0]['disposition']==expect
control=subprocess.run([sys.executable,'-c',"import sys,socket;sys.addaudithook(lambda e,a: (_ for _ in ()).throw(RuntimeError('SIDE_EFFECT_BLOCKED')) if e.startswith('socket.') else None);socket.socket()"],text=True,capture_output=True)
assert control.returncode!=0 and 'SIDE_EFFECT_BLOCKED' in control.stderr
results.append({'case':'network_tripwire_positive_control','verdict':'PASS'})
proof={'run_timestamp':datetime.datetime.now(datetime.timezone.utc).isoformat(),'environment':{'python':sys.version,'platform':platform.platform(),'cwd':str(ROOT)},'intent':'Offline historical snapshot review CLI','risk_tier':'R1','status':'partial','source_ticket':'TK-11785','task_id':'yf0045-build','input_identity':json.loads((actual/'manifest.json').read_text()),'command':'python3 verify_cli.py','cases':results,'case_evidence':str(CASES),'assertions':{'91_unique_snapshot_rows':'PASS','5_candidates_all_historically_archived':'PASS','all_vendor_unverified_and_no_apply_eligible':'PASS','strict_input_failure_controls':'PASS','real_cli_json_markdown_and_repeat':'PASS','network_and_child_process_tripwire':'PASS'},'skipped':{'live_current_inventory':'SKIP: authorized offline scope; fresh authoritative export still required','vendor_and_manufacturing_identity':'SKIP: absent in snapshot; operational dry-run remains blocked','parent_acceptance':'SKIP: parent owns independent review'},'cleanup':'Retained local test bundles and outputs; source snapshots unchanged. No live side effects.','build_identity':{'review_py_sha256':hashlib.sha256((ROOT/'review.py').read_bytes()).hexdigest(),'verify_cli_py_sha256':hashlib.sha256((ROOT/'verify_cli.py').read_bytes()).hexdigest()}}
(ROOT/'verification/e2e-proof.json').write_text(json.dumps(proof,indent=2)+'\n')
print(json.dumps({'cases_passed':len(results),'counts':r['counts'],'proof':'verification/e2e-proof.json'}))