"""Tests for the DGR-001 performance contract metadata.""" from __future__ import annotations import json from unittest.mock import MagicMock, patch from meshnet_node.performance_contract import ( BENCHMARK_SCHEMA_VERSION, DEFAULT_CONTRACT, SCHEMA_VERSION, main, run_performance_benchmark, run_real_model_endpoint_benchmark, ) def test_default_contract_is_architecture_aligned_and_small(): """The baseline stays on DeepSeek2 and uses the smallest DeepSeek-family GGUF. Tags: performance, model, gguf """ payload = DEFAULT_CONTRACT.to_dict() assert payload["schema_version"] == SCHEMA_VERSION assert payload["story_id"] == "DGR-001" assert payload["model_target"] == { "name": "DeepSeek-V2-Lite-Chat", "architecture": "deepseek2", "safetensors_repo": "deepseek-ai/DeepSeek-V2-Lite-Chat", "safetensors_precision": "bfloat16", "gguf_repo": "second-state/DeepSeek-V2-Lite-Chat-GGUF", "gguf_quant": "Q2_K", "gguf_size_gb": 6.43, "comparison_policy": ( "same model/revision, closest practical low-footprint precision pair: " "BF16 safetensors versus Q2_K GGUF" ), "rationale": ( "Smallest DeepSeek-family benchmark anchor that still points toward " "DeepSeek-V4-Flash; keeps the runtime on the DeepSeek2 path instead " "of falling back to a tiny but architecture-mismatched smoke model." ), } assert payload["benchmark_lanes"] == [ { "id": "transformers-safetensors-cpu", "runtime": "transformers", "device": "cpu", "recipe": "current safetensors recipe", "concurrency_levels": [1, 4], }, { "id": "llama-cpp-gguf-cpu", "runtime": "llama.cpp", "device": "cpu", "recipe": "whole-model GGUF recipe", "concurrency_levels": [1, 4], }, { "id": "transformers-safetensors-gpu", "runtime": "transformers", "device": "gpu", "recipe": "current safetensors recipe", "concurrency_levels": [1, 4], }, { "id": "llama-cpp-gguf-gpu", "runtime": "llama.cpp", "device": "gpu", "recipe": "whole-model GGUF recipe", "concurrency_levels": [1, 4], }, ] assert "ttft_ms" in payload["metrics"] assert "output_drift" in payload["metrics"] assert "meaningful speed or fit benefit" in payload["stop_condition"] assert any("mounted drive" in note for note in payload["notes"]) def test_contract_cli_writes_json(tmp_path, capsys): """The contract can be emitted as a machine-readable artifact. Tags: performance, artifact """ output = tmp_path / "performance-contract.json" assert main(["--json-out", str(output)]) == 0 written = json.loads(output.read_text(encoding="utf-8")) assert written == DEFAULT_CONTRACT.to_dict() assert str(output) in capsys.readouterr().out def test_stub_benchmark_covers_every_lane_concurrency_and_metric(): """The runner exercises all four CPU/GPU lanes with the full metric set. Tags: performance, benchmark, gguf """ report = run_performance_benchmark() assert report["schema_version"] == BENCHMARK_SCHEMA_VERSION assert report["story_id"] == "DGR-001" assert report["source"] == "stub-backend" assert report["model_target"] == DEFAULT_CONTRACT.model_target.to_dict() assert [lane["id"] for lane in report["lanes"]] == [ lane.id for lane in DEFAULT_CONTRACT.benchmark_lanes ] for lane in report["lanes"]: assert [result["concurrency"] for result in lane["results"]] == [1, 4] for result in lane["results"]: assert set(result["metrics"]) == set(DEFAULT_CONTRACT.metrics) assert result["metrics"]["failure_count"] == 0 assert result["metrics"]["decode_tok_per_sec"] > 0 def test_stub_benchmark_is_deterministic(): """Two runs produce byte-identical reports; no clocks or randomness leak in. Tags: performance, benchmark, deterministic """ first = run_performance_benchmark() second = run_performance_benchmark() assert first == second assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) def test_stub_benchmark_compares_gguf_against_safetensors_per_device(): """Each device gets a GGUF-vs-safetensors comparison and a stop-condition verdict. Tags: performance, benchmark, gguf """ report = run_performance_benchmark() assert set(report["comparisons"]) == {"cpu", "gpu"} cpu, gpu = report["comparisons"]["cpu"], report["comparisons"]["gpu"] assert cpu["safetensors_lane"] == "transformers-safetensors-cpu" assert cpu["gguf_lane"] == "llama-cpp-gguf-cpu" assert cpu["memory_metric"] == "rss_bytes" assert gpu["safetensors_lane"] == "transformers-safetensors-gpu" assert gpu["gguf_lane"] == "llama-cpp-gguf-gpu" assert gpu["memory_metric"] == "vram_bytes" for comparison in (cpu, gpu): assert comparison["decode_speedup"] > 1.0 assert comparison["artifact_bytes_ratio"] < 0.5 assert comparison["memory_bytes_ratio"] < 1.0 assert comparison["output_drift"] == 0.0 assert comparison["gguf_benefit"] is True assert report["stop_condition"]["gguf_benefit"] is True assert report["stop_condition"]["triggered"] is False assert report["stop_condition"]["text"] == DEFAULT_CONTRACT.stop_condition def test_contract_cli_writes_benchmark_report(tmp_path, capsys): """--benchmark-out emits the stub benchmark report next to the contract. Tags: performance, benchmark, artifact """ contract_out = tmp_path / "performance-contract.json" benchmark_out = tmp_path / "artifacts" / "stub-benchmark-report.json" assert main(["--json-out", str(contract_out), "--benchmark-out", str(benchmark_out)]) == 0 report = json.loads(benchmark_out.read_text(encoding="utf-8")) assert report == run_performance_benchmark() output = capsys.readouterr().out assert str(contract_out) in output assert str(benchmark_out) in output def test_real_model_endpoint_benchmark_uses_lane_specific_endpoints_and_shared_schema(): """The live client path fans out to one endpoint per CPU/GPU lane. Tags: performance, benchmark, live """ response = MagicMock() response.read.return_value = json.dumps({"choices": [{"message": {"content": "mesh activation"}}]}).encode() response.headers.get.return_value = "lane-session" response.__enter__.return_value = response endpoints = { "transformers-safetensors-cpu": "http://cpu-safetensors", "llama-cpp-gguf-cpu": "http://cpu-gguf", "transformers-safetensors-gpu": "http://gpu-safetensors", "llama-cpp-gguf-gpu": "http://gpu-gguf", } with patch("meshnet_node.performance_contract.urllib.request.urlopen", return_value=response) as urlopen: report = run_real_model_endpoint_benchmark(endpoints=endpoints, model="deepseek-ai/DeepSeek-V2-Lite-Chat") assert report["source"] == "real-model-endpoints" assert report["model_target"] == DEFAULT_CONTRACT.model_target.to_dict() assert set(report["comparisons"]) == {"cpu", "gpu"} assert urlopen.call_count == len(endpoints) called_urls = [call.args[0].full_url for call in urlopen.call_args_list] assert called_urls == [f"{url}/v1/chat/completions" for url in endpoints.values()] for lane in report["lanes"]: assert lane["results"][0]["metrics"]["decode_tok_per_sec"] > 0 assert lane["results"][0]["metrics"]["ttft_ms"] > 0 assert lane["output_tokens"] == ["mesh", "activation"] assert report["comparisons"]["cpu"]["gguf_lane"] == "llama-cpp-gguf-cpu" assert report["comparisons"]["gpu"]["gguf_lane"] == "llama-cpp-gguf-gpu"