182 lines
8.4 KiB
Python
182 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""STEP-119：基于真实 Style 索引的粒子独占 C 分段验证器测试。仅结构测试，非 GPU。"""
|
||
import importlib.util
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
MODULE = Path(__file__).with_name("verify_step113_fast_path.py")
|
||
spec = importlib.util.spec_from_file_location("step113", MODULE)
|
||
verifier = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(verifier)
|
||
|
||
|
||
def sample(ranges):
|
||
return {
|
||
"planValid": True, "basePreserved": False, "tideLayerExcluded": True,
|
||
"originalBaseLayerCount": 63, "effectiveBaseLayerCount": 62,
|
||
"excludedLayerId": "tide-spots", "excludedLayerSource": "tide",
|
||
"excludedLayerCount": 1, "actualLayers": 64,
|
||
"commands": {"emptyBuiltinAnnotationLayerRemoved": True,
|
||
"annotationSourcePreserved": True},
|
||
"sources": [{"sourceId": "navsea_delivery"},
|
||
{"sourceId": "rn-weather-fields"}],
|
||
"retained": [{"begin": start, "end": end} for start, end in ranges],
|
||
"segmentRanges": (
|
||
[{"strategy": "RETAINED", "begin": ranges[0][0], "end": ranges[0][1]},
|
||
{"strategy": "CONTINUOUS", "begin": ranges[0][1], "end": ranges[1][0]}]
|
||
+ [{"strategy": "RETAINED", "begin": a, "end": b}
|
||
for a, b in ranges[1:]]
|
||
if len(ranges) >= 2 else []
|
||
),
|
||
}
|
||
|
||
|
||
SHAPES = {
|
||
# Stable WeatherColor is in R[0,32); only Particle at [32,33) is C.
|
||
"RCRR_C_NARROW_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
|
||
"RCRR_FULL_CONTROL_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
|
||
"RCR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 64)],
|
||
"RCRRR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 38), (38, 63), (63, 64)],
|
||
# Intentionally unsupported boundaries: keep as independent negative scenes.
|
||
"RCRR_C_WIDE_NO_TIDE": [(0, 2), (38, 63), (63, 64)],
|
||
"RCRR_R_DYNAMIC_NO_TIDE": [(0, 33), (38, 63), (63, 64)],
|
||
}
|
||
|
||
|
||
class Step119RcrrVerifierTests(unittest.TestCase):
|
||
def test_all_expected_shapes_pass_exact_validation(self):
|
||
for name, ranges in SHAPES.items():
|
||
with self.subTest(name=name):
|
||
self.assertTrue(verifier.scene_parity_valid(sample(ranges), name))
|
||
|
||
def test_incorrect_ranges_do_not_pass_other_layouts(self):
|
||
for name, ranges in SHAPES.items():
|
||
for other in SHAPES:
|
||
if other == name or {name, other} == {
|
||
"RCRR_C_NARROW_NO_TIDE", "RCRR_FULL_CONTROL_NO_TIDE"}:
|
||
continue
|
||
with self.subTest(name=name, other=other):
|
||
self.assertFalse(verifier.scene_parity_valid(sample(ranges), other))
|
||
|
||
def test_missing_source_or_annotation_guard_fails_closed(self):
|
||
for name, ranges in SHAPES.items():
|
||
a = sample(ranges)
|
||
a["sources"].append({"sourceId": "tide"})
|
||
self.assertFalse(verifier.scene_parity_valid(a, name))
|
||
b = sample(ranges)
|
||
b["commands"]["emptyBuiltinAnnotationLayerRemoved"] = False
|
||
self.assertFalse(verifier.scene_parity_valid(b, name))
|
||
|
||
def test_invalid_last_extent_or_count(self):
|
||
for name, ranges in SHAPES.items():
|
||
bad = [*ranges]
|
||
bad[-1] = (bad[-1][0], 65)
|
||
self.assertFalse(verifier.scene_parity_valid(sample(bad), name))
|
||
self.assertFalse(verifier.scene_parity_valid(sample(ranges[:-1]), name))
|
||
|
||
def test_actual_c_segment_must_match_particle_only_boundary(self):
|
||
exact = sample(SHAPES["RCRR_C_NARROW_NO_TIDE"])
|
||
exact["segmentRanges"][1]["end"] = 38
|
||
self.assertFalse(verifier.scene_parity_valid(
|
||
exact, "RCRR_C_NARROW_NO_TIDE"))
|
||
missing = sample(SHAPES["RCRR_C_NARROW_NO_TIDE"])
|
||
missing.pop("segmentRanges")
|
||
self.assertFalse(verifier.scene_parity_valid(
|
||
missing, "RCRR_C_NARROW_NO_TIDE"))
|
||
|
||
def test_full_control_may_omit_cache_but_not_segment_plan(self):
|
||
control = sample(SHAPES["RCRR_FULL_CONTROL_NO_TIDE"])
|
||
control["retained"] = []
|
||
self.assertTrue(verifier.scene_parity_valid(
|
||
control, "RCRR_FULL_CONTROL_NO_TIDE"))
|
||
control["segmentRanges"][1]["begin"] = 31
|
||
self.assertFalse(verifier.scene_parity_valid(
|
||
control, "RCRR_FULL_CONTROL_NO_TIDE"))
|
||
|
||
def test_static_rc_no_tide_old_contract_unchanged(self):
|
||
old = sample([(0, 62)])
|
||
self.assertTrue(verifier.scene_parity_valid(old, "STATIC_RC_NO_TIDE"))
|
||
self.assertFalse(verifier.scene_parity_valid(old, "RCRR_C_NARROW_NO_TIDE"))
|
||
|
||
def test_legacy_original_63_layer_fails_if_changed(self):
|
||
original = {"planValid": True, "basePreserved": True,
|
||
"tideLayerExcluded": False}
|
||
self.assertTrue(verifier.scene_parity_valid(original, "PARTIAL_RCRR"))
|
||
original["tideLayerExcluded"] = True
|
||
self.assertFalse(verifier.scene_parity_valid(original, "PARTIAL_RCRR"))
|
||
|
||
|
||
class Step119ProcessCpuTests(unittest.TestCase):
|
||
def test_proc_stat_parses_cpu_and_stable_birth_with_parentheses(self):
|
||
spec_cpu = importlib.util.spec_from_file_location(
|
||
"benchmark", Path(__file__).with_name("benchmark_step119_cpu.py"))
|
||
# Direct loading tests the real no-third-party benchmark implementation.
|
||
import sys
|
||
parent = str(Path(__file__).parent)
|
||
if parent not in sys.path:
|
||
sys.path.insert(0, parent)
|
||
benchmark = importlib.util.module_from_spec(spec_cpu)
|
||
spec_cpu.loader.exec_module(benchmark)
|
||
parts = ["S"] + ["0"] * 19
|
||
parts[11] = "120"
|
||
parts[12] = "20"
|
||
parts[19] = "90000"
|
||
ticks, born = benchmark.parse_proc_stat(
|
||
"222 (android app (bench)) " + " ".join(parts))
|
||
self.assertEqual((ticks, born), (140, 90000))
|
||
|
||
def test_cpu_per_frame_is_normalized_by_actual_native_fps(self):
|
||
import sys
|
||
parent = str(Path(__file__).parent)
|
||
if parent not in sys.path:
|
||
sys.path.insert(0, parent)
|
||
import benchmark_step119_cpu as benchmark
|
||
a = {"fastRendered": 100, "fullRendered": 5, "sampleElapsedMs": 1000}
|
||
b = {"fastRendered": 200, "fullRendered": 5, "sampleElapsedMs": 61000}
|
||
result = benchmark.collect_result(
|
||
(a, b), {"ticks": 200}, {"ticks": 500}, 100, 60.0)
|
||
self.assertAlmostEqual(result["cpuSeconds"], 3.0)
|
||
self.assertAlmostEqual(result["cpuCoreEquivalent"], .05)
|
||
self.assertAlmostEqual(result["nativeObservedFps"], 100.0 / 60.0)
|
||
self.assertAlmostEqual(result["estimatedCpuMsPerNativeFrame"], 30.0)
|
||
|
||
|
||
class Step119CpuComparisonTests(unittest.TestCase):
|
||
@staticmethod
|
||
def report(scenario, status, ms, cores, fps):
|
||
native = {"fingerprint": "test-pbf", "actualLayers": 64,
|
||
"lifecycle": 7, "decisionMismatch": 0, "retained": []}
|
||
return {
|
||
"sourceCommit": "source-locked", "apkSha256": "sha-locked",
|
||
"device": "test-device", "scenario": scenario, "status": status,
|
||
"nativeStart": dict(native), "nativeEnd": dict(native),
|
||
"nativeCpuBoundaryOffsetMs": [100, 300],
|
||
"parityValid": True, "controlMatched": True,
|
||
"metrics": {"estimatedCpuMsPerNativeFrame": ms, "cpuCoreEquivalent": cores,
|
||
"nativeObservedFps": fps, "fastFraction": .98},
|
||
}
|
||
|
||
def test_valid_aba_cpu_saving_report(self):
|
||
import compare_step119_cpu as compare
|
||
a = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 40, .50, 15)
|
||
b = self.report("RCRR_C_NARROW_NO_TIDE", "FAST_DOMINANT", 20, .42, 28)
|
||
c = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 42, .51, 14)
|
||
result = compare.summarize(a, b, c, [])
|
||
self.assertAlmostEqual(result["fullControlCpuMsPerFrameEstimatedMedian"], 41)
|
||
self.assertAlmostEqual(result["comparisons"][0]["cpuMsPerFrameSavingPercentVsFull"],
|
||
100 * (1 - 20 / 41))
|
||
self.assertNotEqual(result["conclusion"], "CPU_DRIFT_REQUIRES_RETEST")
|
||
|
||
def test_aba_source_mismatch_is_rejected(self):
|
||
import compare_step119_cpu as compare
|
||
a = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 40, .5, 15)
|
||
b = self.report("RCRR_C_NARROW_NO_TIDE", "FAST_DOMINANT", 20, .4, 28)
|
||
c = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 41, .5, 15)
|
||
b["apkSha256"] = "different-apk"
|
||
with self.assertRaises(ValueError):
|
||
compare.summarize(a, b, c, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|