#!/usr/bin/env python3 """Three real user/auth flows; two config-only routes do not inflate DB coverage.""" import copy import hashlib import json from pathlib import Path import integrate_20261010 as tool from integrate_20261011_batches46_48 import item HERE = Path(__file__).resolve().parent P = Path('/Users/wiscourper/Documents/workspace/local/dm8-batch116-20261012.S1TtvG') BASE = 'work_ai_api/2026-10-12-dm8-hk-course-batch/' REPORT = BASE+'NEXT5_ROOT_REVIEW.md' def local(path): return P/Path(path).name if path.startswith('/tmp/cdx-ai-hk-course-c.') else Path(path) def normalized(body, actor=None): body = copy.deepcopy(body); body.pop('trace_id', None) if actor is not None and isinstance(body.get('data'), dict) and 'userId' in body['data']: assert type(body['data']['userId']) is int and body['data']['userId'] == actor assert body['data']['token'] == '[REDACTED]' body['data']['userId'] = 0 return body def verify(): raw = (tool.ROOT/BASE/'NEXT5_FINAL_MANIFEST.json').read_bytes() assert hashlib.sha256(raw).hexdigest() == '3eb7c9b6e7ab7dc5987ece958749639ba12519fcd74a9f0be3bf0e168627a317' specs = json.loads(raw)['keys']; assert len(specs) == 5 for path,pin in { 'app/api/controller/User.php':'91c85805222273382f59639171b5c4ccd537c0775f76c96c107d7814400d158b', 'app/api/model/User.php':'cd2855df8b0ab7869451e5189c6c28b9d10d1ce568664bd0427f2bfea2c30699', 'app/api/controller/Passport.php':'06d6e4b4891776d5a9ca347bbbe46d766bb309123a2be80f5b969a8069aec1b8', 'app/api/service/passport/Login.php':'90cca88e572003eeacf74aa2d28c75bf1cca25d6dbb3482b124cefa86bbbb597', }.items(): assert hashlib.sha256((tool.source_root('ai_api')/path).read_bytes()).hexdigest() == pin for index,spec in enumerate(specs): responses=[] for lane in ['B','C']: rawkey = lane+('_raw_redacted' if lane+'_raw_redacted' in spec else '_raw') for key in [lane+'_receipt',rawkey]: assert hashlib.sha256(local(spec[key]).read_bytes()).hexdigest() == spec[key+'_sha256'] events = json.loads(local(spec[lane+'_receipt']).read_text())['events'] wires = [json.loads(x) for x in local(spec[rawkey]).read_text().splitlines() if x] before, final = events[0],events[-1] assert final['token_absent'] assert [e['at_epoch'] for e in events] == sorted(e['at_epoch'] for e in events) if index == 0: after = events[1] assert after['http'] == after['business_status'] == 200 and after['basic_pass'] and after['sidecar_pass'] assert after['store_rows_unchanged'] and before['original_rows']['store'] == after['original_rows']['store'] assert after['today_weight_record_count'] == 1 and after['today_weight_matches_input'] assert after['response_id_type'] == 'integer' assert len(wires) == 1 and wires[0]['path'] == spec['path'] assert hashlib.sha256(wires[0]['response_raw'].encode()).hexdigest() == after['response_raw_sha256'] body=json.loads(wires[0]['response_raw']); assert type(body['data']['id']) is int assert body['data']['id'] == before['own_ai_user_id'] body['data']['id']=0; responses.append(normalized(body)) elif index in [1,2]: wrong = events[1] assert wrong['http'] == 200 and wrong['business_status'] == 500 and wrong['pass'] for field in ['ai_user_row_sha256','store_user_row_sha256']: assert before[field] == wrong[field] assert wrong['captcha_still_present'] and final['captcha_absent'] if index == 1: good=events[2] assert good['http'] == good['business_status'] == 200 and good['pass'] assert good['store_user_unchanged'] and good['store_user_row_sha256']==before['store_user_row_sha256'] assert good['private_captcha_consumed'] and good['private_token_present'] assert good['external_stub_delta_bytes']==0 and len(wires)==2 else: assert [e['phase'] for e in events] == ['before_http','after_wrong_sms','after_correct_http_raw','after_correct_sms_reset','after_new_password_login','final_private_cache'] good, login = events[3],events[4] assert not before['new_password_matches_old_hash'] assert good['http']==good['business_status']==200 and good['pass'] assert good['store_pwd_matches_input_in_memory'] and good['store_pwd_sha256'] != before['store_pwd_sha256'] assert good['other_store_fields_unchanged'] and good['ai_user_unchanged'] and good['captcha_consumed'] assert login['http']==login['business_status']==200 and login['pass'] and login['store_pwd_still_matches_input'] assert login['private_token_present'] and login['external_stub_delta_bytes']==0 and len(wires)==3 bodies=[] for n,wire in enumerate(wires): assert wire['http']==200 and wire['method']=='POST' assert wire['response_redacted']['status']==(500 if n==0 else 200) bodies.append(normalized(wire['response_redacted'],before['own_ai_user_id'])) responses.append(bodies) else: for event in events[1:-1]: assert event['http']==event['business_status']==200 and event['pass'] and event['own_rows_unchanged'] assert event['external_stub_delta_bytes']==0 assert len(wires)==2 responses.append([normalized(json.loads(w['response_raw'])) for w in wires]) assert responses[0] == responses[1],spec['path'] if __name__ == '__main__': audit=HERE/'integration-20261012-batch120.json'; assert not audit.exists() targets=[item('ai_api','POST',path,BASE,evidence=REPORT, success_case=scope, readback='tested-pass-contemporaneous-own-AI-and-Store-rows-private-token-captcha-readback', parity='partial-evidence-complete-nonsensitive-response-equal-own-integer-ID-mapped-token-redacted', cleanup='own-synthetic-users-retained-token-captcha-absent-no-real-SMS-or-third-party') for path,scope in [('/api/User/sync','partial-evidence-existing-owned-user-staff-weight-update-not-create'), ('/api/Passport/login','partial-evidence-private-SMS-wrong-and-correct-code'), ('/api/Passport/resetPassword','partial-evidence-new-independent-wrong-correct-SMS-and-new-password-real-login')]] targets.append(item('ai_api','POST','/api/Passport/loginAccount',BASE,evidence=REPORT,evidence_only=True)) tool.main(targets,audit,verify)