City-Lifeline-Cockpit/.scratch/cockpit-real-data-wayfinding/implementation/t30/live-acceptance.py

"""Read-only anonymous GET acceptance; never launches or changes the backend."""
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError

URL = 'http://192.168.110.96:8301/api/cockpit/v1/infrastructure/geometry'
OUTPUT = Path(__file__).resolve().parent / 'live-acceptance.json'
checks = []

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

report = {'url': URL, 'checkedAt': datetime.now(timezone.utc).isoformat(),
          'authorizationHeaderSent': False, 'checks': checks}
try:
    with urlopen(Request(URL, headers={'Accept': 'application/json'}), timeout=30) as response:
        report['httpStatus'] = response.status
        envelope = json.load(response)
    check('HTTP 200 and business code 200', report['httpStatus'] == 200 and envelope.get('code') == 200)
    data = envelope.get('data', {})
    check('GeoJSON FeatureCollection', data.get('type') == 'FeatureCollection')
    features = data.get('features', [])
    coverage = data.get('coverage', [])
    report['featureCount'] = len(features)
    report['coverage'] = coverage
    report['industries'] = dict(Counter(f.get('properties', {}).get('industry') for f in features))
    report['pointTypes'] = dict(Counter(str(f.get('properties', {}).get('pointType')) for f in features
                                      if f.get('properties', {}).get('source') == 'gas_pipe_point'))
    report['validationStates'] = dict(Counter(f.get('properties', {}).get('coordinateValidation', {}).get('state')
                                            for f in features))
    report['validationReasons'] = dict(Counter(str(f.get('properties', {}).get('coordinateValidation', {}).get('reason'))
                                             for f in features))
    check('Feature IDs unique', len({f.get('id') for f in features}) == len(features))
    check('Feature types and namespaced source IDs', all(f.get('type') == 'Feature' and
          f.get('id') == f.get('properties', {}).get('source', '') + ':' + str(f.get('properties', {}).get('sourceId'))
          for f in features))
    valid_shapes = True
    valid_validation = True
    geometry_consistent = True
    raw_consistent = True
    expected_industries = {'pipe_network_base': {'GAS', 'WATER_SUPPLY', 'DRAINAGE', 'UNKNOWN'},
                           'gas_pipe_point': {'GAS'}, 'manhole_base': {'MANHOLE'}}
    check('Industry source domains preserved', all(f.get('properties', {}).get('industry') in
          expected_industries.get(f.get('properties', {}).get('source'), set()) for f in features))
    check('Point type and network relationship explicitly present', all('pointType' in f.get('properties', {}) and
          'networkId' in f.get('properties', {}) for f in features))
    for feature in features:
        p = feature.get('properties', {})
        validation = p.get('coordinateValidation', {})
        validations = p.get('coordinateValidations', [])
        raw = p.get('sourceCoordinates', [])
        geometry = feature.get('geometry')
        line = p.get('source') == 'pipe_network_base'
        expected = 2 if line else 1
        valid_validation &= all({'state', 'reason', 'renderable', 'crs'} <= set(v) and
                                v.get('renderable') is False and v.get('crs') == 'WGS84'
                                for v in [validation, *validations])
        valid_validation &= validation.get('state') in {'INVALID', 'UNVERIFIED'} and len(validations) == expected
        raw_consistent &= len(raw) == expected and all(isinstance(pair, list) and len(pair) == 2 for pair in raw)
        if validation.get('state') == 'INVALID':
            geometry_consistent &= 'geometry' in feature and geometry is None and bool(validation.get('reason'))
        else:
            geometry_consistent &= geometry is not None and validation.get('reason') == 'UNVERIFIED_COORDINATE'
            if geometry is None:
                continue
            valid_shapes &= geometry.get('type') == ('LineString' if line else 'Point')
            coordinates = geometry.get('coordinates')
            raw_consistent &= coordinates == (raw if line else raw[0])
            pairs = coordinates if line else [coordinates]
            valid_shapes &= len(pairs) == expected
            for pair in pairs:
                valid_shapes &= isinstance(pair, list) and len(pair) == 2
                if not isinstance(pair, list) or len(pair) != 2:
                    continue
                lon, lat = pair
                valid_shapes &= isinstance(lon, (int, float)) and isinstance(lat, (int, float))
                if isinstance(lon, (int, float)) and isinstance(lat, (int, float)):
                    valid_shapes &= -180 <= lon <= 180 and -90 <= lat <= 90 and (lon, lat) != (0, 0)
    check('Real two-endpoint lines and points use longitude latitude order', valid_shapes and raw_consistent)
    check('Coordinate validation same shape and no fabricated verified state', valid_validation)
    check('Invalid geometry is null and unverified geometry remains nonrenderable', geometry_consistent)
    expected_coverage = {('GAS', 'LineString'), ('WATER_SUPPLY', 'LineString'), ('DRAINAGE', 'LineString'),
                         ('GAS', 'Point'), ('WATER_SUPPLY', 'Point'), ('DRAINAGE', 'Point'),
                         ('MANHOLE', 'Point'), ('MANHOLE', 'LineString'), ('UNKNOWN', 'LineString')}
    check('All industry coverage entries present once', len(coverage) == 9 and
          {(c.get('industry'), c.get('geometryType')) for c in coverage} == expected_coverage)
    count_consistent = True
    for item in coverage:
        matching = [f for f in features if f.get('properties', {}).get('source') == item.get('source') and
                    f.get('properties', {}).get('industry') == item.get('industry')]
        count_consistent &= item.get('recordCount') == len(matching)
        count_consistent &= item.get('geometryCount') == sum(f.get('geometry') is not None for f in matching)
        count_consistent &= item.get('renderableCount') == 0
        if item.get('source'):
            count_consistent &= item.get('state') == ('AVAILABLE' if matching else 'EMPTY')
    check('Coverage counts recompute from collection', count_consistent)
    gaps = {(c.get('industry'), c.get('geometryType')): c for c in coverage}
    check('Water drainage manhole coverage gaps explicit', all(
          gaps.get(key, {}).get('state') == 'UNAVAILABLE' and gaps.get(key, {}).get('reason') == reason
          for key, reason in [(('WATER_SUPPLY', 'Point'), 'NO_COMPONENT_SOURCE'),
                              (('DRAINAGE', 'Point'), 'NO_TYPED_COMPONENT_SOURCE'),
                              (('MANHOLE', 'LineString'), 'NO_PIPELINE_SOURCE')]))
    report['observedInvalidCount'] = sum(f.get('properties', {}).get('coordinateValidation', {}).get('state') == 'INVALID'
                                         for f in features)
    report['coverageMissingSamples'] = [source for source in expected_industries if
                                         not any(f.get('properties', {}).get('source') == source for f in features)]
    check('Real pipeline sample exists', any(f.get('properties', {}).get('source') == 'pipe_network_base' for f in features))
except HTTPError as error:
    report['httpStatus'] = error.code
    report['error'] = error.read(4096).decode('utf-8', errors='replace')
    check('Endpoint available', False)
except (URLError, OSError, ValueError, TypeError, KeyError) as error:
    report['error'] = str(error)
    check('Probe completed', False)
report['passed'] = bool(checks) and all(c['passed'] for c in checks)
OUTPUT.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps(report, ensure_ascii=False, indent=2))
raise SystemExit(0 if report['passed'] else 1)