bitchat-android/tools/release_gate/test_release_gate.py

368 lines
13 KiB
Python

import json
import subprocess
import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest import mock
from tools.release_gate import android_lab
from tools.release_gate.release_gate import (
GateError,
append_trace_event,
canonical_json_digest,
create_bundle,
initialize_run,
load_json,
parse_evidence,
record_result,
validate_manifest,
validate_matrix,
validate_privacy,
validate_run,
)
COMMIT = "1" * 40
TOOL_DIRECTORY = Path(__file__).parent
def valid_matrix():
return {
"schema_version": 1,
"commit": COMMIT,
"clients": {
"android-current": {"version": "2.0.0-rc1", "commit": COMMIT},
"android-legacy": {"version": "1.9.0"},
"ios-current": {"version": "2.0.0"},
},
"lab_capabilities": ["local-relay", "tor"],
"devices": [
{
"alias": "android-low",
"platform": "android",
"model": "model-low",
"manufacturer_class": "vendor-a",
"api_level": 28,
"physical": True,
"capabilities": ["ble-central", "ble-peripheral"],
},
{
"alias": "android-current",
"platform": "android",
"model": "model-current",
"manufacturer_class": "vendor-b",
"api_level": 35,
"physical": True,
"capabilities": ["ble-central", "ble-peripheral", "wifi-aware"],
},
{
"alias": "android-relay",
"platform": "android",
"model": "model-relay",
"manufacturer_class": "vendor-c",
"api_level": 33,
"physical": True,
"capabilities": ["ble-central", "ble-peripheral"],
},
{
"alias": "android-aware",
"platform": "android",
"model": "model-aware",
"manufacturer_class": "vendor-b",
"api_level": 35,
"physical": True,
"capabilities": ["ble-central", "ble-peripheral", "wifi-aware"],
},
{
"alias": "android-legacy",
"platform": "android",
"model": "model-legacy",
"manufacturer_class": "vendor-a",
"api_level": 28,
"physical": True,
"capabilities": ["ble-central", "ble-peripheral"],
},
{
"alias": "ios-current",
"platform": "ios",
"model": "ios-model",
"manufacturer_class": "apple",
"physical": True,
"capabilities": ["ble-central", "ble-peripheral"],
},
],
}
class ReleaseGateTest(unittest.TestCase):
def setUp(self):
self.manifest = load_json(TOOL_DIRECTORY / "scenarios.json")
def test_manifest_covers_every_release_category(self):
scenarios = validate_manifest(self.manifest)
self.assertEqual(27, len(scenarios))
self.assertEqual(
{
"transport",
"android-platform",
"android-to-android",
"cross-client",
"endurance",
},
{scenario["category"] for scenario in scenarios.values()},
)
def test_documented_matrix_template_is_schema_valid_but_not_runnable(self):
template = load_json(TOOL_DIRECTORY / "device-matrix.example.json")
self.assertEqual(6, len(validate_matrix(template, allow_placeholders=True)))
with self.assertRaises(GateError):
validate_matrix(template)
def test_matrix_enforces_distinct_api_manufacturer_and_counterpart_clients(self):
matrix = valid_matrix()
self.assertEqual(6, len(validate_matrix(matrix, COMMIT)))
for device in matrix["devices"]:
if device["platform"] == "android":
device["manufacturer_class"] = "one-vendor"
with self.assertRaisesRegex(GateError, "manufacturer"):
validate_matrix(matrix, COMMIT)
def test_privacy_policy_rejects_identifiers_paths_addresses_and_long_ids(self):
rejected = (
{"serial": "device-selector"},
{"note": "/" + "home/operator/result"},
{"note": "operator@example.test"},
{"note": "192.0.2.1"},
{"note": "aa:bb:cc:dd:ee:ff"},
{"note": "0123456789abcdef"},
)
for value in rejected:
with self.subTest(value=value), self.assertRaises(GateError):
validate_privacy(value)
validate_privacy({"commit": COMMIT, "packet-correlation-count": 3})
def test_initialize_creates_disposable_fixtures_and_pending_results(self):
with tempfile.TemporaryDirectory() as temporary:
run = Path(temporary) / "rc-run"
results = initialize_run(
self.manifest,
valid_matrix(),
run,
COMMIT,
"rc-run",
started_at="2026-01-01T00:00:00+00:00",
)
self.assertTrue(all(
result["status"] == "pending"
for result in results["scenario_results"].values()
))
fixtures = load_json(run / "fixtures" / "manifest.json")["fixtures"]
sizes = {fixture["name"]: fixture["size_bytes"] for fixture in fixtures}
self.assertEqual(0, sizes["empty.bin"])
self.assertEqual(50 * 1024 * 1024, sizes["maximum.bin"])
self.assertEqual(50 * 1024 * 1024 + 1, sizes["oversized.bin"])
self.assertEqual(
results["scenario_manifest_sha256"],
canonical_json_digest(self.manifest),
)
summary = validate_run(run, allow_incomplete=True)
self.assertEqual(27, summary["pending"])
self.assertFalse(summary["complete"])
def test_record_requires_structured_evidence_and_rejects_pii(self):
with tempfile.TemporaryDirectory() as temporary:
run = Path(temporary) / "rc-run"
initialize_run(self.manifest, valid_matrix(), run, COMMIT, "rc-run")
with self.assertRaises(GateError):
record_result(run, "A2A-001", "pass", {}, None)
with self.assertRaises(GateError):
record_result(
run,
"A2A-001",
"pass",
{"connection-transitions": "operator@example.test"},
None,
)
with self.assertRaisesRegex(GateError, "reason code"):
record_result(run, "A2A-001", "fail", {}, None)
recorded = record_result(
run,
"A2A-001",
"blocked",
{},
"counterpart-unavailable",
updated_at="2026-01-01T00:00:00+00:00",
)
self.assertEqual(
["blocked"],
[
item["status"]
for item in recorded["scenario_results"]["A2A-001"]["history"]
],
)
def test_complete_run_requires_all_evidence_traces_and_endurance_bounds(self):
with tempfile.TemporaryDirectory() as temporary:
run = Path(temporary) / "rc-run"
initialize_run(self.manifest, valid_matrix(), run, COMMIT, "rc-run")
scenarios = validate_manifest(self.manifest)
for scenario_id, scenario in scenarios.items():
evidence = {key: 1 for key in scenario["evidence"]}
if "duration-minutes" in evidence:
evidence["duration-minutes"] = 240
if "cycle-count" in evidence:
evidence["cycle-count"] = 50
record_result(
run,
scenario_id,
"pass",
evidence,
None,
updated_at="2026-01-01T04:00:00+00:00",
)
append_trace_event(
run,
scenario_id,
scenario["participants"][0],
"scenario-terminal",
"pass",
None,
{"assertion-count": len(evidence)},
timestamp="2026-01-01T04:00:00+00:00",
)
summary = validate_run(run)
self.assertTrue(summary["complete"])
self.assertEqual(27, summary["pass"])
self.assertEqual(27, summary["trace_events"])
bundle = Path(temporary) / "release-gate.zip"
create_bundle(run, bundle)
with zipfile.ZipFile(bundle) as archive:
self.assertEqual(
{
"SHA256SUMS",
"fixtures/manifest.json",
"manifest.json",
"matrix.json",
"results.json",
"summary.md",
"trace.jsonl",
},
set(archive.namelist()),
)
with self.assertRaisesRegex(GateError, "overwrite"):
create_bundle(run, bundle)
def test_manifest_tampering_after_initialization_is_detected(self):
with tempfile.TemporaryDirectory() as temporary:
run = Path(temporary) / "rc-run"
initialize_run(self.manifest, valid_matrix(), run, COMMIT, "rc-run")
manifest = load_json(run / "manifest.json")
manifest["scenario_version"] = "tampered"
(run / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
with self.assertRaisesRegex(GateError, "changed"):
validate_run(run, allow_incomplete=True)
def test_trace_accepts_only_aggregate_metrics(self):
with tempfile.TemporaryDirectory() as temporary:
run = Path(temporary) / "rc-run"
initialize_run(self.manifest, valid_matrix(), run, COMMIT, "rc-run")
with self.assertRaisesRegex(GateError, "numeric"):
append_trace_event(
run,
"A2A-001",
"android-current",
"packet",
"observed",
None,
{"raw-packet": "payload"},
)
def test_evidence_parser_is_typed_and_privacy_checked(self):
self.assertEqual(
{"count": 3, "ratio": 0.5, "clean": True},
parse_evidence(["count=3", "ratio=0.5", "clean=true"]),
)
with self.assertRaises(GateError):
parse_evidence(["note=10.0.0.1"])
@mock.patch("tools.release_gate.android_lab.find_adb", return_value="adb")
def test_adb_device_count_never_returns_selectors(self, _find_adb):
completed = subprocess.CompletedProcess(
["adb", "devices"],
0,
"List of devices attached\nselector-one\tdevice\nselector-two\toffline\n",
"",
)
count = android_lab.count_connected_devices(runner=lambda *args, **kwargs: completed)
self.assertEqual(1, count)
@mock.patch("tools.release_gate.android_lab.run_adb")
def test_adb_probe_emits_only_logical_device_metadata(self, run_adb):
run_adb.side_effect = [
"35",
"feature:android.hardware.wifi.aware",
"Vendor",
"Model",
]
probe = android_lab.probe_device("ephemeral-selector", "android-current")
self.assertEqual("android-current", probe["alias"])
self.assertNotIn("serial", probe)
self.assertIn("wifi-aware", probe["capabilities"])
@mock.patch("tools.release_gate.android_lab.run_adb")
def test_disposable_cleanup_targets_the_real_application_id(self, run_adb):
run_adb.side_effect = ["", "Success"]
android_lab.prepare_disposable_device("ephemeral-selector", confirmed=True)
self.assertEqual(
[
mock.call(
"ephemeral-selector",
["shell", "am", "force-stop", "com.bitchat.droid"],
),
mock.call(
"ephemeral-selector",
["shell", "pm", "clear", "com.bitchat.droid"],
),
],
run_adb.call_args_list,
)
@mock.patch("tools.release_gate.android_lab.run_adb")
def test_resource_snapshot_returns_only_aggregate_metrics(self, run_adb):
run_adb.side_effect = [
"123",
"TOTAL 2048",
"7",
"11",
"WakeLock com.bitchat.droid\nWakeLock another.package",
"level: 73",
]
metrics = android_lab.collect_resource_snapshot("ephemeral-selector")
self.assertEqual(
{
"process-running": True,
"total-pss-kb": 2048,
"thread-count": 7,
"fd-count": 11,
"app-wakelock-count": 1,
"battery-level-percent": 73,
},
metrics,
)
self.assertEqual(
mock.call(
"ephemeral-selector",
["shell", "pidof", "com.bitchat.droid"],
),
run_adb.call_args_list[0],
)
if __name__ == "__main__":
unittest.main()