#!/usr/bin/env python3 """Laboratoire pédagogique : observations entièrement fictives, aucun navigateur lancé. Python 3, bibliothèque standard. Ce programme contrôle la comparabilité et calcule les écarts de deux parcours décrits dans les données. Il n'émet ni HTTP ni 103, ne simule pas le cache de Chrome et ne lit pas de HAR. Les quantités d'octets sont des tailles transférées fictives, pas une conversion automatique du LCP. """ from copy import deepcopy from itertools import permutations from math import isfinite from statistics import median import json ENVIRONMENT = { 'browser': 'Chrome-156-fixture', 'protocol': 'h2', 'trusted_https': True, 'cache_enabled': True, 'initial_cache': 'empty', 'service_worker': 'absent', 'edge_state': 'warm', 'build': 'catalog-v7', 'viewport': '390x844', 'network': '4g-fixed-fixture', 'cpu': '4x-fixture', 'lcp_element': 'catalog-hero', 'final_css': '/assets/catalog.v7.css', } CSS_BYTES = 45056 # 44 KiB, hypothèse fabriquée pour cet exercice. class InvalidObservation(ValueError): pass def observation(name, enabled=False, pair=1, offset=0): return { 'name': name, 'pair': pair, 'hints_enabled': enabled, 'environment': deepcopy(ENVIRONMENT), 'document_final_ms': 800, 'early_hint_ms': 120 if enabled else None, 'requests': [{'url': ENVIRONMENT['final_css'], 'start_ms': 150 if enabled else 820, 'end_ms': 370 if enabled else 1040, 'bytes': CSS_BYTES, 'initiator': 'early-hints' if enabled else 'parser'}], 'page_reused_hinted_css': enabled, 'lcp_ms': (1280 if enabled else 1500) + offset, } def number(value): return type(value) in (int, float) and isfinite(value) and value >= 0 def validate(row): if not isinstance(row, dict): raise InvalidObservation('observation absente') env = row.get('environment') if not isinstance(env, dict) or set(env) != set(ENVIRONMENT): raise InvalidObservation('conditions incomplètes') if any(env[k] is None or env[k] == '' for k in env): raise InvalidObservation('condition inconnue') if type(env['cache_enabled']) is not bool or type(env['trusted_https']) is not bool: raise InvalidObservation('booléen attendu') if env['initial_cache'] not in ('empty', 'warm'): raise InvalidObservation('état initial non qualifié') if type(row.get('hints_enabled')) is not bool or type(row.get('page_reused_hinted_css')) is not bool: raise InvalidObservation('preuve de traitement inconnue') if not row.get('name') or type(row.get('pair')) is not int or row['pair'] < 1: raise InvalidObservation('identité du passage manquante') final, lcp, hint = row.get('document_final_ms'), row.get('lcp_ms'), row.get('early_hint_ms') if not number(final) or not number(lcp) or lcp < final: raise InvalidObservation('horloge du document ou LCP incohérente') if hint is not None and (not number(hint) or hint >= final): raise InvalidObservation('103 non antérieur à la réponse finale') reqs = row.get('requests') if not isinstance(reqs, list) or not reqs: raise InvalidObservation('requêtes absentes') for request in reqs: if not isinstance(request, dict) or not request.get('url'): raise InvalidObservation('URL de ressource manquante') if not number(request.get('start_ms')) or not number(request.get('end_ms')) or request['start_ms'] > request['end_ms']: raise InvalidObservation('chronologie de ressource incohérente') if type(request.get('bytes')) is not int or request['bytes'] < 0: raise InvalidObservation('octets inconnus ou invalides') if request.get('initiator') not in ('early-hints', 'parser', 'cache'): raise InvalidObservation('initiateur hors périmètre') if request['initiator'] == 'early-hints' and (hint is None or request['start_ms'] < hint): raise InvalidObservation('requête anticipée sans signal antérieur') needed = [r for r in reqs if r['url'] == env['final_css']] if len(needed) != 1 or needed[0]['end_ms'] > lcp: raise InvalidObservation('CSS final ambigu ou postérieur au rendu') if row['page_reused_hinted_css'] and needed[0]['initiator'] != 'early-hints': raise InvalidObservation('réutilisation sans chargement correspondant') return needed[0] def compare(control, treatment): a, b = validate(control), validate(treatment) if control['pair'] != treatment['pair']: raise InvalidObservation('passages non appariés') if control['hints_enabled'] or not treatment['hints_enabled']: raise InvalidObservation('une paire témoin/activation est nécessaire') different = sorted(k for k in ENVIRONMENT if control['environment'][k] != treatment['environment'][k]) if different: return {'comparable': False, 'reason': different} env = treatment['environment'] # Périmètre choisi du laboratoire, pas une détection universelle du support. if not env['cache_enabled'] or not env['trusted_https'] or env['protocol'] not in ('h2', 'h3'): return {'comparable': False, 'reason': ['outside_pilot_conditions']} unused = sum(r['bytes'] for r in treatment['requests'] if r['initiator'] == 'early-hints' and r['url'] != env['final_css']) return { 'comparable': True, 'earlier_needed_request_ms': a['start_ms'] - b['start_ms'], 'earlier_needed_completion_ms': a['end_ms'] - b['end_ms'], 'lcp_gain_ms': control['lcp_ms'] - treatment['lcp_ms'], 'unused_hinted_bytes': unused, 'transfer_difference_bytes': sum(r['bytes'] for r in treatment['requests']) - sum(r['bytes'] for r in control['requests']), 'reused': treatment['page_reused_hinted_css'], } def summarize(rows): pairs = {} names = set() for row in rows: validate(row) if row['name'] in names: raise InvalidObservation('passage répété') names.add(row['name']) pair = pairs.setdefault(row['pair'], {}) side = row['hints_enabled'] if side in pair: raise InvalidObservation('deux passages du même côté') pair[side] = row gains = [] for pair in pairs.values(): if set(pair) != {False, True}: raise InvalidObservation('paire incomplète') result = compare(pair[False], pair[True]) if not result['comparable']: raise InvalidObservation('conditions différentes : pas de total global') gains.append(result['lcp_gain_ms']) if not gains: raise InvalidObservation('aucune paire') return {'pairs': len(gains), 'median_paired_gain_ms': median(gains), 'min_gain_ms': min(gains), 'max_gain_ms': max(gains)} def main(): checks = 0 def check(condition, message): nonlocal checks if not condition: raise AssertionError(message) checks += 1 def refused(fn, message): try: fn() except InvalidObservation: check(True, message) else: check(False, message) control = observation('A') treatment = observation('B', True) result = compare(control, treatment) check(result == {'comparable': True, 'earlier_needed_request_ms': 670, 'earlier_needed_completion_ms': 670, 'lcp_gain_ms': 220, 'unused_hinted_bytes': 0, 'transfer_difference_bytes': 0, 'reused': True}, 'corrigé indépendant') check(result['lcp_gain_ms'] != result['earlier_needed_request_ms'], 'avance réseau ne vaut pas gain LCP') check(round(100 * result['lcp_gain_ms'] / control['lcp_ms'], 2) == 14.67, 'pourcentage fictif') disabled = observation('C', True) disabled['environment']['cache_enabled'] = False disabled['requests'] = deepcopy(control['requests']) disabled['page_reused_hinted_css'] = False disabled['lcp_ms'] = 1500 check(compare(control, disabled) == {'comparable': False, 'reason': ['cache_enabled']}, 'cache désactivé') warm = observation('D', True) warm['environment']['initial_cache'] = 'warm' warm['requests'] = [{'url': ENVIRONMENT['final_css'], 'start_ms': 820, 'end_ms': 820, 'bytes': 0, 'initiator': 'cache'}] warm['page_reused_hinted_css'] = False warm['lcp_ms'] = 1100 check(compare(control, warm) == {'comparable': False, 'reason': ['initial_cache']}, 'visite chaude') check(control['lcp_ms'] - warm['lcp_ms'] == 400, 'faux gain séduisant mais non attribuable') stale = observation('E', True) stale['requests'][0]['url'] = '/assets/catalog.v6.css' stale['requests'].extend(deepcopy(control['requests'])) stale['page_reused_hinted_css'] = False stale['lcp_ms'] = 1500 stale_result = compare(control, stale) check(stale_result['comparable'] and stale_result['lcp_gain_ms'] == 0, 'mauvais hash conservé comme résultat négatif') check(stale_result['unused_hinted_bytes'] == 45056 and stale_result['transfer_difference_bytes'] == 45056, '44KiB gaspillage du scénario') check(compare(control, dict(stale, requests=list(reversed(stale['requests'])))) == stale_result, 'ordre requêtes sans effet') no_gain = deepcopy(treatment); no_gain['name'] = 'F'; no_gain['lcp_ms'] = 1500 check(compare(control, no_gain)['earlier_needed_request_ms'] == 670 and compare(control, no_gain)['lcp_gain_ms'] == 0, 'avance réelle dans la fixture sans gain rendu') regression = deepcopy(treatment); regression['name'] = 'G'; regression['lcp_ms'] = 1580 check(compare(control, regression)['lcp_gain_ms'] == -80, 'régression conservée') variants = {'browser':'other-version','protocol':'h3','trusted_https':False,'service_worker':'active', 'edge_state':'cold','build':'catalog-v8','viewport':'1280x900','network':'wifi','cpu':'1x','lcp_element':'title'} for field, value in variants.items(): changed = deepcopy(treatment); changed['environment'][field] = value check(compare(control, changed) == {'comparable': False, 'reason':[field]}, field) both_off = deepcopy(control); both_off['environment']['cache_enabled'] = False check(not compare(both_off, disabled)['comparable'], 'deux caches désactivés ne qualifient pas le pilote103') rows = [observation(f'{side}-{pair}', side, pair, offset) for pair, offset in [(1,0),(2,20),(3,-20)] for side in [False,True]] expected = {'pairs':3,'median_paired_gain_ms':220,'min_gain_ms':220,'max_gain_ms':220} check(summarize(rows) == expected, 'trois paires prédéfinies') for order in permutations([0,1,2]): shuffled = [rows[2*i+side] for i in order for side in [1,0]] check(summarize(shuffled) == expected, 'permutation des trois paires') mixed = deepcopy(rows); mixed[-1]['lcp_ms'] = 1560 # -80 pour la troisième paire. check(summarize(mixed) == {'pairs':3,'median_paired_gain_ms':220,'min_gain_ms':-80,'max_gain_ms':220}, 'résultat défavorable conservé dans la dispersion') refused(lambda: summarize(rows[:-1]), 'paire manquante') refused(lambda: summarize(rows+[rows[0]]), 'passage répété') refused(lambda: summarize([]), 'aucune observation') refused(lambda: compare(control, dict(treatment, pair=2)), 'mauvais appariement') refused(lambda: compare(control, control), 'activation absente') bad = deepcopy(rows); bad[-1]['environment']['initial_cache'] = 'warm' refused(lambda: summarize(bad), 'pas de mélange silencieux') mutations = [lambda x:x['environment'].pop('cpu'), lambda x:x['environment'].update(cache_enabled='false'), lambda x:x['environment'].update(initial_cache='unknown'), lambda x:x.update(lcp_ms=float('nan')), lambda x:x.update(lcp_ms=True), lambda x:x.update(early_hint_ms=900), lambda x:x['requests'][0].update(bytes=-1), lambda x:x['requests'][0].update(end_ms=100), lambda x:x['requests'][0].update(url='/other.css'), lambda x:x.update(page_reused_hinted_css=None)] for mutate in mutations: bad = deepcopy(treatment); mutate(bad) refused(lambda: compare(control,bad), 'observation incohérente') check(20 * 6 == 120, 'hypothèse de temps de recette en minutes') print(json.dumps({'status':'PASS','checks':checks,'normal':result,'wrong_hash':stale_result, 'paired':expected,'all_observations_fictitious':True, 'browser_executed':False,'http_sent':False,'real_user_measurement':False},ensure_ascii=False,indent=2)) if __name__ == '__main__': main()