busya
bcc69f4bbe
Add versioned metric graph authority, owned browser evidence, paginated and all-tab traversal, deterministic sampling policies, and analyst-facing run inspection. Provision the DEV/PREPROD/PROD Superset, Gitea and million-row ClickHouse lab; retain reproducible lifecycle evidence and explicit incomplete-traversal limits.
100 lines
5.4 KiB
Python
100 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
# #region Verification.FinanceMillion.TabsAudit [C:4] [TYPE Module] [SEMANTICS tabs,evidence,ownership,coverage]
|
|
# @BRIEF Verify all three actual finance tabs using independently retained same-run page bytes.
|
|
# @PRE Input comes from the read-only actual journal extractor; no expected tab list is accepted from callers.
|
|
# @INVARIANT Page count alone cannot establish visited identity, chart settlement or artifact ownership.
|
|
import argparse
|
|
import base64
|
|
from hashlib import sha256
|
|
import json
|
|
|
|
EXPECTED = [('TAB-finance-receivables',[4,5,6]),('TAB-finance-cash',[7,8,9]),('TAB-finance-reserves',[10,11,12])]
|
|
|
|
|
|
# #region Verification.FinanceMillion.TabsAudit.Canonical [C:1] [TYPE Function]
|
|
def canonical(value):
|
|
return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=False,allow_nan=False).encode()
|
|
# #endregion Verification.FinanceMillion.TabsAudit.Canonical
|
|
|
|
|
|
# #region Verification.FinanceMillion.TabsAudit.Page [C:3] [TYPE Function]
|
|
def page_proof(record,args,ordinal,chain,source):
|
|
receipt,artifact=record['receipt'],record['artifact']
|
|
data=base64.b64decode(record['bytes_base64'],validate=True)
|
|
assert artifact['owner_type']=='scenario_run' and artifact['owner_id']==args.run_id
|
|
assert artifact['logical_step_id']==args.step_id and artifact['attempt']==args.attempt and artifact['is_active']
|
|
assert artifact['content_type']=='application/json'
|
|
assert artifact['id']==receipt['artifact_id'] and artifact['sha256']==receipt['sha256']==sha256(data).hexdigest()
|
|
assert artifact['content_ref']==receipt['content_ref']==f'draft:{args.run_id}:{receipt["sha256"]}'
|
|
assert artifact['byte_length']==receipt['byte_length']==len(data)
|
|
assert record['ordinal']==receipt['ordinal']==ordinal
|
|
assert receipt['row_count']==1 and receipt['next_available']==(ordinal<3)
|
|
chain=sha256(chain.encode()+canonical({k:v for k,v in receipt.items() if k!='chain_digest'})).hexdigest()
|
|
assert receipt['chain_digest']==chain
|
|
observed=json.loads(data)
|
|
expected_id,expected_charts=EXPECTED[ordinal-1]
|
|
assert observed['tab']==source['tabs'][ordinal-1]
|
|
assert observed['tab']=={'id':expected_id,'parent_tabs_id':'TABS-finance','ancestors':[],'chart_ids':expected_charts}
|
|
assert observed['active'] is True and observed['panel_id']
|
|
assert observed['charts']==[{'chart_id':i,'visible':True,'settled':True} for i in expected_charts]
|
|
return chain,len(data),observed['panel_id']
|
|
# #endregion Verification.FinanceMillion.TabsAudit.Page
|
|
|
|
|
|
# #region Verification.FinanceMillion.TabsAudit.Stream [C:4] [TYPE Function]
|
|
def audit(lines,args):
|
|
header=None
|
|
count=byte_count=0
|
|
chain=''
|
|
panels=[]
|
|
ended=False
|
|
for line in lines:
|
|
record=json.loads(line)
|
|
if record['record_type']=='header':
|
|
assert header is None and count==0
|
|
header=record
|
|
assert header['run']['id']==args.run_id and header['run']['status'] in ('passed','inconclusive')
|
|
journal=header['journal']
|
|
assert journal['run_id']==args.run_id and journal['logical_step_id']==args.step_id and journal['attempt']==args.attempt
|
|
assert journal['action']=='navigate_tabs' and journal['status']=='passed'
|
|
plan=header['run']['runner_plan']
|
|
assert journal['plan_hash']==plan['plan_hash']==sha256(canonical({k:v for k,v in plan.items() if k!='plan_hash'})).hexdigest()
|
|
assert journal['state']['source']['source_total']==3
|
|
elif record['record_type']=='page':
|
|
assert header is not None and not ended
|
|
count+=1
|
|
chain,length,panel=page_proof(record,args,count,chain,header['journal']['state']['source'])
|
|
byte_count+=length
|
|
panels.append(panel)
|
|
else:
|
|
assert record['record_type']=='end' and not ended and record['run_id']==args.run_id and record['logical_step_id']==args.step_id and record['attempt']==args.attempt
|
|
ended=True
|
|
assert header is not None and ended and count==3 and len(set(panels))==3
|
|
state=header['journal']['state']
|
|
assert state['terminal'] is True and state['next_ordinal']==4 and state['row_count']==3
|
|
assert state['chain_digest']==chain and state['byte_count']==byte_count
|
|
return {'status':'PASS','run_id':args.run_id,'logical_step_id':args.step_id,'run_status':header['run']['status'],'scope':'navigate_tabs_step_only','owned_tabs':3,'settled_chart_ids':list(range(4,13)),'chain_digest':chain,'limitation':'Retained DOM observations prove this run; original server-layout response bytes are not separately retained'}
|
|
# #endregion Verification.FinanceMillion.TabsAudit.Stream
|
|
|
|
|
|
# #region Verification.FinanceMillion.TabsAudit.Main [C:2] [TYPE Function]
|
|
def main():
|
|
parser=argparse.ArgumentParser()
|
|
parser.add_argument('input')
|
|
parser.add_argument('--run-id',required=True)
|
|
parser.add_argument('--step-id',required=True)
|
|
parser.add_argument('--attempt',type=int,default=1)
|
|
args=parser.parse_args()
|
|
try:
|
|
with open(args.input) as lines:
|
|
result=audit(lines,args)
|
|
except (AssertionError,ValueError,KeyError,TypeError,IndexError) as exc:
|
|
result={'status':'FAIL','reason':type(exc).__name__,'scope':'owned-all-finance-tabs'}
|
|
print(json.dumps(result,ensure_ascii=False,indent=2))
|
|
return int(result['status']!='PASS')
|
|
# #endregion Verification.FinanceMillion.TabsAudit.Main
|
|
|
|
if __name__=='__main__':
|
|
raise SystemExit(main())
|
|
# #endregion Verification.FinanceMillion.TabsAudit
|