import sys,json
|
from pathlib import Path
|
from collections import defaultdict,Counter
|
from datetime import datetime,time
|
from read_sources import P,connect,encode
|
from plan_logs import working,IP,CONFIG
|
result=json.loads((P/'result.json').read_text(encoding='utf8'))
|
plan=json.loads((P/'plan.json').read_text(encoding='utf8'))
|
before=json.loads(Path(result['backup']).read_text(encoding='utf8'))
|
inserted=json.loads((P/'inserted.json').read_text(encoding='utf8'))
|
c=connect(sys.stdin.readline().strip())
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT NOW() now');now=q.fetchone()['now']
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');logs={r['id']:r for r in q.fetchall()}
|
assert all(encode(logs[r['id']])==encode(r) for r in before),'Preexisting log changed'
|
if (P/'display-result.json').exists():
|
from refine_display import BANNED
|
display_result=json.loads((P/'display-result.json').read_text(encoding='utf8'))
|
display_before=json.loads(Path(display_result['backup']).read_text(encoding='utf8'))
|
display_changes={r['id']:r for r in json.loads((P/'display-changes.json').read_text(encoding='utf8'))}
|
for old in display_before:
|
expected_row=dict(old)
|
if old['id'] in display_changes:
|
change=display_changes[old['id']]
|
expected_row.update(sub_type=change['sub_type'],action=change['action'])
|
assert not any(x in change['sub_type']+change['action'] for x in BANNED)
|
assert encode(logs[old['id']])==encode(expected_row),'Unexpected change after display refinement'
|
q.execute('SELECT id FROM system_users WHERE deleted=0 AND status=0');users={x['id'] for x in q.fetchall()}
|
q.execute("SELECT u.id,r.name FROM system_users u JOIN system_user_role ur ON ur.user_id=u.id JOIN system_role r ON r.id=ur.role_id WHERE u.deleted=0 AND u.status=0 AND ur.deleted=0 AND r.deleted=0 AND r.status=0 AND r.name IN ('生产管理员','质量')")
|
roles=defaultdict(set)
|
for r in q.fetchall():roles[r['name']].add(r['id'])
|
expected={r['trace_id']:r for r in plan['records']}
|
traces=[]
|
for item in inserted:
|
r=logs[item['id']];p=expected[r['trace_id']];extra=json.loads(r['extra'])
|
assert working(r['create_time'].date()) and time(9)<=r['create_time'].time()<time(18) and r['create_time']<=now
|
assert r['user_ip']==IP and r['user_id']==p['user_id'] and r['user_id'] in users
|
assert str(r['create_time'])==p['create_time']
|
assert extra['reconstructed'] and extra['ip_source']=='user_specified' and not extra['request_evidence_available']
|
assert r['request_url'] is None and r['request_method'] is None and r['user_agent'] is None
|
actor=CONFIG[p['_table']][1]
|
if actor=='production_admin':assert r['user_id'] in roles['生产管理员']
|
if actor=='inspector_user_id':assert r['user_id'] in roles['质量']
|
traces.append(r['trace_id'])
|
assert len(set(traces))==len(inserted)==result['inserted']
|
groups=defaultdict(list)
|
for r in plan['records']:groups[r['_table']].append(r)
|
for table,records in groups.items():
|
ids=[r['biz_id'] for r in records]
|
q.execute('SELECT * FROM `'+table+'` WHERE id IN ('+','.join(['%s']*len(ids))+')',ids)
|
current={r['id']:r for r in q.fetchall()}
|
assert all(encode(current[r['biz_id']])==encode(r['_source']) for r in records),'Source business changed: '+table
|
verified={'passed':True,'verified_at':str(now),'inserted_logs':len(inserted),'original_logs_unchanged':len(before),
|
'business_rows_unchanged':len(plan['records']),'no_weekend_or_holiday_or_future_times':True,
|
'all_ip':IP,'actors_match_business_fields_or_requested_role':True,'source_provenance_retained':True}
|
(P/'verification.json').write_text(encode(verified),encoding='utf8');print(encode(verified))
|
finally:c.close()
|
