import math import pytest from internetpressure.config import ConfigError, build_config, load_file_config from internetpressure.engine import scoring as S def test_robust_z_basic_and_clipping(): assert S.robust_z(30, 20, 2) == 5.0 assert S.robust_z(200, 20, 2) == 8.0 # clipped high assert S.robust_z(0, 20, 2) == -3.0 # clipped low assert S.robust_z(None, 20, 2) is None assert S.robust_z(20, None, 2) is None def test_robust_z_mad_floor_prevents_explosion(): # ultra-tight baseline (MAD 0.01 ms on a 20 ms median): floor = max(1.0, 0.05*20=1.0) → z uses 1.0 assert S.robust_z(25, 20, 0.01) == 5.0 # relative floor on large medians assert S.robust_z(1100, 1000, 1) == pytest.approx(2.0) # floor 50 ms def test_mad_from_iqr(): assert S.mad_from_iqr(10, 20) == pytest.approx(7.413) assert S.mad_from_iqr(None, 20) is None def test_z_to_stress_ramp(): assert S.z_to_stress(None) == 0 assert S.z_to_stress(0.5) == 0 assert S.z_to_stress(1.0) == 0 assert S.z_to_stress(3.5) == pytest.approx(0.5) assert S.z_to_stress(6) == 1 assert S.z_to_stress(8) == 1 def test_rate_stress(): assert S.rate_stress(0.0, 0.0) == 0 assert S.rate_stress(0.25, 0.0) == 1.0 assert S.rate_stress(0.125, 0.0) == pytest.approx(0.5) # a noisy target with a 10 % baseline failure rate needs 30 % excess for full stress assert S.rate_stress(0.25, 0.10) == pytest.approx(0.5) assert S.rate_stress(0.05, 0.10) == 0 def test_ratio_stress(): assert S.ratio_stress(100, 100) == 0 assert S.ratio_stress(150, 100) == 0 assert S.ratio_stress(375, 100) == pytest.approx(0.5) assert S.ratio_stress(1000, 100) == 1 assert S.ratio_stress(10, 0) == 0 def test_saturate_bounds_and_shape(): assert S.saturate(0) == 0 assert S.saturate(1) == pytest.approx(100) mid = S.saturate(0.5) assert 55 < mid < 75 # concave: half stress is well above 50 assert S.saturate(0.1) < S.saturate(0.2) < S.saturate(0.5) assert S.saturate(0.5, k=0) == 50 def test_coverage_factor(): assert S.coverage_factor(0, 24) == 0 assert S.coverage_factor(2, 24) == 0 # below 20 % assert S.coverage_factor(24, 24) == 1 assert 0 < S.coverage_factor(12, 24) < 1 assert S.coverage_factor(5, 0) == 1 def test_weighted_stress_and_breadth(): pairs = [ S.PairStress("a", 1.0, 2.0), S.PairStress("b", 0.0, 1.0), S.PairStress("c", 0.5, 1.0), S.PairStress("d", 0.9, 0.0), # zero weight ignored ] stress, tw, n = S.weighted_stress(pairs) assert n == 3 and tw == 4.0 assert stress == pytest.approx((2.0 + 0 + 0.5) / 4.0) assert S.breadth(pairs, threshold=0.4) == pytest.approx(3.0 / 4.0) assert S.weighted_stress([]) == (0.0, 0.0, 0) def test_component_score_renormalises_missing_signals(): w = {"a": 0.5, "b": 0.3, "c": 0.2} assert S.component_score({}, w) == 0 full = S.component_score({"a": 1, "b": 1, "c": 1}, w) assert full == pytest.approx(100) # only "a" observed at full stress → still 100 (renormalised), not 50 assert S.component_score({"a": 1}, w) == pytest.approx(100) assert S.component_score({"a": 0, "b": 0, "c": 0}, w) == 0 def test_global_score_weights_from_config(): cfg = load_file_config() comps = {c: 50.0 for c in cfg.weights} score, contrib = S.global_score(comps, cfg.weights) assert score == pytest.approx(50.0) assert sum(contrib.values()) == pytest.approx(50.0) # missing routing → renormalised over the rest comps["routing"] = None score2, contrib2 = S.global_score(comps, cfg.weights) assert score2 == pytest.approx(50.0) assert "routing" not in contrib2 def test_global_score_example_from_spec(): weights = {"routing": .25, "latency": .20, "dns": .15, "availability": .15, "http_tls": .10, "path": .10, "corroboration": .05} comps = {"routing": 57, "latency": 44, "dns": 19, "availability": 37, "http_tls": 28, "path": 51, "corroboration": 0} score, _ = S.global_score(comps, weights) assert score == pytest.approx(57 * .25 + 44 * .2 + 19 * .15 + 37 * .15 + 28 * .1 + 51 * .1) def test_confidence_monotone(): low = S.confidence_score(probes=1, probe_regions=1, signals_agreeing=1, samples=2, min_samples=24) high = S.confidence_score(probes=8, probe_regions=4, signals_agreeing=3, samples=100, min_samples=24, duration_s=1200, bgp_corroborated=True, external_corroborated=True) assert 0 < low < high <= 1.0 def test_velocity_and_acceleration(): pts = [(t * 60.0, 20 + t * 0.5) for t in range(0, 30)] # +0.5/min = +30/h v, a = S.velocity(pts) assert v == pytest.approx(30.0, rel=1e-3) assert a == pytest.approx(0.0, abs=1e-6) pts2 = [(t * 60.0, 20 + 0.02 * t * t) for t in range(0, 30)] # accelerating v2, a2 = S.velocity(pts2) assert v2 > 0 and a2 > 0 assert S.velocity(pts[:3]) == (None, None) def test_volatility_and_trend(): assert S.volatility([1, 1, 1]) == 0 assert S.volatility([1, 2]) is None assert S.volatility([10, 20, 30]) == pytest.approx(10.0) assert S.trend_word(0.4) == "stable" assert S.trend_word(5) == "rising" assert S.trend_word(-5) == "falling" assert S.trend_word(None) == "stable" def test_config_validation_rejects_bad_weights(): cfg = load_file_config() raw = dict(cfg.raw) raw["pressure_weights"] = {**cfg.weights, "routing": 0.5} with pytest.raises(ConfigError): build_config(raw) raw["pressure_weights"] = {k: v for k, v in cfg.weights.items() if k != "path"} with pytest.raises(ConfigError): build_config(raw) def test_levels_from_config(): cfg = load_file_config() assert cfg.level_for(5)[0] == "calm" assert cfg.level_for(10)[0] == "calm" assert cfg.level_for(10.1)[0] == "normal" assert cfg.level_for(42.7)[0] == "stressed" assert cfg.level_for(99)[0] == "extreme" assert cfg.level_for(None)[0] == "unknown" assert not math.isnan(cfg.importance_weight(5)) assert cfg.importance_weight(5) > cfg.importance_weight(1)