"""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)