City-Lifeline-Cockpit/.scratch/cockpit-real-data-wayfinding/implementation/t29/real-probe.py

from pathlib import Path
import json
import urllib.request
import urllib.error
import urllib.parse
from datetime import datetime, timezone

root = Path(__file__).parent
base = 'http://192.168.110.96:8301/api/cockpit/v1/'
keys = ['ALARM','WARNING','WORK_ORDER','NORMAL','OFFLINE','EMPTY','UNKNOWN','UNASSOCIATED']
rows = []
checks = []

def get(endpoint, params):
    url = base + endpoint + '?' + urllib.parse.urlencode(params, doseq=True)
    try:
        with urllib.request.urlopen(url, timeout=15) as response:
            status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:
        status, raw = error.code, error.read()
    envelope = json.loads(raw)
    data = envelope.get('data') or {}
    row = {'endpoint':endpoint,'params':params,'http':status,'business':envelope.get('code'),
           'category':data.get('category'),'total':data.get('total', (data.get('pagination') or {}).get('total')),
           'mapFilter':data.get('mapFilter'),'distribution':data.get('pointStatusDistribution'),
           'snapshotAt':data.get('snapshotAt'),'monitoringPoints':data.get('monitoringPoints'),
           'unassociatedEquipment':data.get('unassociatedEquipment')}
    rows.append(row)
    return row

def check(name, passed):
    checks.append({'name':name,'passed':bool(passed)})

baselines = {}
for endpoint in ['map-snapshot','monitoring/distribution']:
    first = get(endpoint, [('pageSize',2)])
    baselines[endpoint] = first
    dist = first['distribution'] or []
    check(endpoint+'-eight-buckets', [r['status'] for r in dist] == keys and all(isinstance(r['count'],int) and r['count'] >= 0 for r in dist))
    check(endpoint+'-total-partition', sum(r['count'] for r in dist)==first['total'])
    for raw in ['ALARM,WARNING','UNKNOWN','UNASSOCIATED','UNASSOCIATED,ALARM','EMPTY','NORMAL']:
        selected = get(endpoint, [('pageSize',2),('pointStatus',raw)])
        check(endpoint+'-'+raw+'-accepted', selected['http']==200 and selected['business']==200)
        check(endpoint+'-'+raw+'-global-unchanged', selected['distribution']==dist)
        if 'UNASSOCIATED' in raw:
            check(endpoint+'-'+raw+'-only-devices', not selected['monitoringPoints'])
    csv = get(endpoint,[('pointStatus','WARNING,ALARM')])
    repeat = get(endpoint,[('pointStatus','ALARM'),('pointStatus','WARNING')])
    check(endpoint+'-repeat-csv-equivalent', csv['mapFilter']==repeat['mapFilter'] and csv['total']==repeat['total'])
    check(endpoint+'-effective-status-filter', csv['total'] < first['total'])
    for raw in ['BOGUS','ALARM,BOGUS']:
        rejected = get(endpoint,[('pointStatus',raw)])
        check(endpoint+'-'+raw+'-invalid-400', rejected['http']==400 and rejected['category']=='INVALID_REQUEST')
    scoped = get(endpoint,[('district','0'),('pointStatus','UNASSOCIATED')])
    check(endpoint+'-district-global-unchanged', scoped['distribution']==dist)

first = baselines['map-snapshot']
second = get('map-snapshot',[('page',2),('pageSize',2),('snapshotAt',first['snapshotAt'])])
check('map-page2-token-retained', second['http']==200 and second['snapshotAt']==first['snapshotAt'] and second['distribution']==first['distribution'])
check('cross-endpoint-global-identical', baselines['map-snapshot']['distribution']==baselines['monitoring/distribution']['distribution'])
check('cross-endpoint-mapFilter-identical', get('map-snapshot',[('pointStatus','UNASSOCIATED,ALARM')])['mapFilter']==get('monitoring/distribution',[('pointStatus','UNASSOCIATED,ALARM')])['mapFilter'])
result={'checkedAt':datetime.now(timezone.utc).isoformat(),'baseUrl':base,'checks':checks,'requests':rows}
(root/'real-probe-new-runtime.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
for item in checks:
    print(('PASS ' if item['passed'] else 'FAIL ')+item['name'])
print(f'{sum(c["passed"] for c in checks)}/{len(checks)} passed')
raise SystemExit(0 if all(c['passed'] for c in checks) else 1)