VerseVDI-Data-Plane/gateway/qualification_contract_test.go

package gateway

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/binary"
	"errors"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"
)

func TestQualificationCatalogMatchesSection7(t *testing.T) {
	media := qualificationMediaProfiles()
	if len(media) != 3 {
		t.Fatalf("media profile count = %d, want 3", len(media))
	}
	wantMedia := []qualificationMediaProfile{
		{Name: "1080p60-h264", Codec: "h264", BitrateKbps: 20000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "1440p120-hevc", Codec: "hevc", BitrateKbps: 50000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "4k60-hevc", Codec: "hevc", BitrateKbps: 80000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
	}
	if !reflect.DeepEqual(media, wantMedia) {
		t.Fatalf("media profiles = %#v, want %#v", media, wantMedia)
	}

	impairments := qualificationImpairmentProfiles()
	wantImpairments := []qualificationImpairmentProfile{
		{Name: "baseline", RTT: 20 * time.Millisecond},
		{Name: "latency", RTT: 150 * time.Millisecond},
		{Name: "jitter", RTT: 50 * time.Millisecond, Jitter: 30 * time.Millisecond},
		{Name: "loss", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 5},
		{Name: "reorder", RTT: 100 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 1, Reorder: true},
		{Name: "constrained", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 2, Reorder: true, CapacitySteps: []int{25, 50}},
	}
	if !reflect.DeepEqual(impairments, wantImpairments) {
		t.Fatalf("impairment profiles = %#v, want %#v", impairments, wantImpairments)
	}
}

func TestQualificationProtocolVersionIsExplicitAndImmutable(t *testing.T) {
	const version = "v1.0.0-phase3c-gateway-rc.8"
	t.Setenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION", version)
	got, err := qualificationProtocolVersion()
	if err != nil || got != version {
		t.Fatalf("qualificationProtocolVersion() = %q, want %q", got, version)
	}
	for _, invalid := range []string{"", "unknown", "v1", " v1.0.0", "v1.0.0+mutable"} {
		t.Setenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION", invalid)
		if _, err := qualificationProtocolVersion(); err == nil {
			t.Fatalf("qualificationProtocolVersion() accepted %q", invalid)
		}
	}
}

func TestQualificationRecordsLinkedToolVersions(t *testing.T) {
	versions, err := qualificationToolVersions()
	if err != nil || versions["qualification"] != qualificationToolVersion ||
		versions["go"] == "" || versions["quic-go"] == "" {
		t.Fatalf("qualification tool versions = %#v, %v", versions, err)
	}
}

func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) {
	if err := validateQualificationOutputDir("relative/evidence"); err == nil {
		t.Fatal("relative evidence directory was accepted")
	}
	if err := validateQualificationOutputDir(filepath.Join(t.TempDir(), "evidence")); err != nil {
		t.Fatalf("absolute evidence directory rejected: %v", err)
	}

	summary, err := summarizeQualificationSamples([]time.Duration{
		time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond,
		4 * time.Millisecond, 5 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	if summary.Count != 5 || summary.Min != time.Millisecond || summary.Median != 3*time.Millisecond ||
		summary.P90 != 5*time.Millisecond || summary.P95 != 5*time.Millisecond ||
		summary.P99 != 5*time.Millisecond || summary.Max != 5*time.Millisecond ||
		summary.Mean != 3*time.Millisecond || summary.Histogram["le_5ms"] != 5 {
		t.Fatalf("summary = %#v", summary)
	}
	if err := enforceQualificationProcessingGate(summary); err != nil {
		t.Fatalf("5 ms p95 rejected: %v", err)
	}
	summary.P95++
	if err := enforceQualificationProcessingGate(summary); err == nil {
		t.Fatal("p95 above 5 ms was accepted")
	}
}

func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) {
	profile := qualificationMediaProfile{
		Name: "smoke", Codec: "h264", BitrateKbps: 20000,
		Duration: time.Second, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	rawPath := filepath.Join(t.TempDir(), "processing.csv.gz")
	summary, err := runQualificationProcessing(t, profile, rawPath)
	if err != nil {
		t.Fatal(err)
	}
	if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 ||
		summary.ResourceSamples < 2 || summary.RawResourcesSHA256 == "" || summary.RawResourcesBytes < 1 ||
		summary.CPUScope != qualificationGatewayCPUScope || summary.ClockOverhead <= 0 || summary.ClockMethod == "" {
		t.Fatalf("processing summary = %#v", summary)
	}
	file, err := os.Open(rawPath)
	if err != nil {
		t.Fatal(err)
	}
	defer file.Close()
	reader, err := gzip.NewReader(file)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := io.ReadAll(reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := reader.Close(); err != nil {
		t.Fatal(err)
	}
	if !strings.HasPrefix(string(raw), "elapsed_ns,queue_ns,processing_ns,pacing_ns\n") ||
		strings.Count(string(raw), "\n") != int(summary.Count)+1 {
		t.Fatalf("raw sample rows do not match summary count: %q", raw)
	}
}

func TestQualificationShortProcessingSubprocessCoversFixedProfiles(t *testing.T) {
	for _, profile := range qualificationMediaProfiles() {
		profile.Duration = time.Second
		profile.Warmup = 10 * time.Millisecond
		summary, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if summary.Count < 1 || summary.CPUScope != qualificationGatewayCPUScope ||
			summary.ClockOverhead <= 0 || summary.ClockMethod != qualificationClockOverheadMethod ||
			summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 ||
			summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 {
			t.Fatalf("%s subprocess summary = %#v", profile.Name, summary)
		}
	}
}

func TestQualificationProcessingResourcesExcludeParentDriverCPU(t *testing.T) {
	profile := qualificationMediaProfile{
		Name: "resource-isolation", Codec: "h264", BitrateKbps: 1000,
		Duration: 300 * time.Millisecond, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	baseline, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "baseline.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	stop := make(chan struct{})
	done := make(chan struct{})
	go func() {
		defer close(done)
		var value uint64 = 1
		for {
			select {
			case <-stop:
				if value == 0 {
					panic("bounded parent work was optimized away")
				}
				return
			default:
				value = value*2862933555777941757 + 3037000493
			}
		}
	}()
	busy, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "busy.csv.gz"))
	close(stop)
	<-done
	if err != nil {
		t.Fatal(err)
	}
	if busy.CPUSeconds > baseline.CPUSeconds+50*time.Millisecond.Seconds() {
		t.Fatalf("parent CPU leaked into gateway sample: baseline=%.6fs busy=%.6fs", baseline.CPUSeconds, busy.CPUSeconds)
	}
}

func TestQualificationProcessingPreservesPayload(t *testing.T) {
	profile := qualificationMediaProfiles()[0]
	payload := qualificationPayload(profile)
	trace, elapsed, err := newQualificationPath(t, profile, profile.BitrateKbps).traverse(t, payload)
	if err != nil {
		t.Fatal(err)
	}
	if !trace.PayloadPreserved || !trace.ApolloRecovered || !trace.VerseQUIC {
		t.Fatalf("production path trace = %#v", trace)
	}
	if elapsed <= 0 {
		t.Fatalf("processing duration = %s", elapsed)
	}
}

func TestQualificationRepeatedTraversalTracksEveryProductionStage(t *testing.T) {
	profile := qualificationMediaProfiles()[1]
	pacerKbps := (profile.BitrateKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
	path := newQualificationPath(t, profile, pacerKbps)
	payload := qualificationPayload(profile)
	if err := runQualificationWarmup(t, path, profile, payload); err != nil {
		t.Fatal(err)
	}
	for index := 0; index < 25_000; index++ {
		current := append([]byte(nil), payload...)
		binary.BigEndian.PutUint32(current[len(current)-4:], uint32(index))
		trace, _, err := path.traverse(t, current)
		if err != nil {
			t.Fatal(err)
		}
		if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
			!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
			!trace.PublicClientDecode || !trace.PayloadPreserved {
			t.Fatalf("traversal %d missed a production stage: %#v", index, trace)
		}
	}
}

func TestQualificationZeroLossBaselinesAreAttributedAtNormativeScale(t *testing.T) {
	for _, media := range qualificationMediaProfiles() {
		observation, err := runQualificationImpairment(t, qualificationImpairmentProfiles()[0], media,
			qualificationImpairmentPacketCount, filepath.Join(t.TempDir(), media.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", media.Name, err)
		}
		if observation.InjectedDropped != 0 || observation.Dropped != 0 ||
			observation.ProviderFECDropped != 0 || observation.ProviderEnqueueDropped != 0 ||
			observation.ProviderQueueReplaced != 0 || observation.GatewayDropped != 0 ||
			observation.QUICSendDropped != 0 || observation.ClientDeliveryDropped != 0 ||
			observation.UnexplainedDropped != 0 {
			t.Fatalf("%s clean-path loss attribution = %#v", media.Name, observation)
		}
		if observation.SourceEmitted != qualificationImpairmentPacketCount ||
			observation.ProviderRecovered != qualificationImpairmentPacketCount ||
			observation.ProviderEnqueued != qualificationImpairmentPacketCount ||
			observation.GatewayForwarded != qualificationImpairmentPacketCount ||
			observation.QUICSent != qualificationImpairmentPacketCount ||
			observation.Delivered != qualificationImpairmentPacketCount {
			t.Fatalf("%s stage counts = %#v", media.Name, observation)
		}
	}
}

func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) {
	profile := qualificationImpairmentProfiles()[3]
	first, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "first.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	second, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "second.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	if first.Dropped != second.Dropped || first.InjectedReordered != second.InjectedReordered {
		t.Fatalf("deterministic impairment selection differs:\n%#v\n%#v", first, second)
	}
	if first.Sent != 1000 || first.Delivered+first.Dropped != first.Sent ||
		first.ObservedLossPercent < 3.5 || first.ObservedLossPercent > 6.5 ||
		first.MaxQueuePackets > qualificationImpairmentQueuePackets || first.RawSamplesSHA256 == "" {
		t.Fatalf("impairment observation = %#v", first)
	}
	if _, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1, filepath.Join(t.TempDir(), "invalid.csv.gz")); err == nil {
		t.Fatal("unbounded impairment packet count was accepted")
	}
	unknown := profile
	unknown.Name = "private-simulator"
	if _, err := runQualificationImpairment(t, unknown, qualificationMediaProfiles()[0], 1, filepath.Join(t.TempDir(), "unknown.csv.gz")); err == nil {
		t.Fatal("unregistered impairment profile was accepted")
	}
}

func TestQualificationRTTIsNotSyntheticDoubleOneWayCompletion(t *testing.T) {
	rawPath := filepath.Join(t.TempDir(), "latency.csv.gz")
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[1],
		qualificationMediaProfiles()[0],
		40,
		rawPath,
	)
	if err != nil {
		t.Fatal(err)
	}
	meanLatency := qualificationRawMeanLatency(t, rawPath)
	delta := observation.ObservedRTT - 2*meanLatency
	if delta < 0 {
		delta = -delta
	}
	if delta < 5*time.Millisecond {
		t.Fatalf("RTT %s was synthesized as twice one-way completion %s", observation.ObservedRTT, meanLatency)
	}
}

func TestQualificationFixedSeedJitterIsObservableOnTraversedTraffic(t *testing.T) {
	profile := qualificationImpairmentProfiles()[2]
	observation, err := runQualificationImpairment(
		t,
		profile,
		qualificationMediaProfiles()[0],
		200,
		filepath.Join(t.TempDir(), "jitter.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.RTTSource != "apollo_enet_acknowledge" || observation.ObservedRTT <= 0 {
		t.Fatalf("RTT observation is not transport-acknowledged: %#v", observation)
	}
	if observation.ObservedLatency < 5*time.Millisecond || observation.ObservedLatency > 100*time.Millisecond {
		t.Fatalf("observed one-way latency %s does not reflect configured traversal", observation.ObservedLatency)
	}
	if observation.AppliedJitter < 10*time.Millisecond || observation.AppliedJitter > 25*time.Millisecond {
		t.Fatalf("applied fixed-seed jitter %s is outside the uniform-delay tolerance", observation.AppliedJitter)
	}
	if observation.ObservedJitter <= 0 || observation.ObservedJitter > observation.AppliedJitter {
		t.Fatalf("ordered traversal jitter %s is not bounded by applied jitter %s", observation.ObservedJitter, observation.AppliedJitter)
	}
	if observation.InjectedReordered != 0 || observation.ObservedOutOfOrder != 0 {
		t.Fatalf("reorder-off jitter changed source order: %#v", observation)
	}
}

func TestQualificationLossOnlyDoesNotImplicitlyReorder(t *testing.T) {
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[3],
		qualificationMediaProfiles()[0],
		400,
		filepath.Join(t.TempDir(), "loss.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.InjectedReordered != 0 || observation.ObservedOutOfOrder != 0 {
		t.Fatalf("loss-only profile changed source order: %#v", observation)
	}
}

func TestQualificationExplicitReorderIsBoundedAndAttributed(t *testing.T) {
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[4],
		qualificationMediaProfiles()[0],
		400,
		filepath.Join(t.TempDir(), "reorder.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.InjectedReordered == 0 || observation.ObservedOutOfOrder != observation.InjectedReordered {
		t.Fatalf("explicit reorder attribution = %#v", observation)
	}
}

func TestQualificationGatewaySubprocessResourcesResetAndTrackWork(t *testing.T) {
	profile := qualificationMediaProfile{
		Name: "resource-process", Codec: "h264", BitrateKbps: 100000,
		Duration: time.Second, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	path := newQualificationProcessingPath(t, profile, qualificationMediaPacerKbps(profile, 0))
	defer path.Close()
	output := t.TempDir()
	record := func(name string, work func() error) qualificationProcessRecordResult {
		t.Helper()
		if err := path.process.startRecording(
			filepath.Join(output, name+".csv.gz"),
			filepath.Join(output, name+"-resources.csv.gz"),
		); err != nil {
			t.Fatal(err)
		}
		if err := work(); err != nil {
			t.Fatal(err)
		}
		result, err := path.process.stopRecording()
		if err != nil {
			t.Fatal(err)
		}
		return result
	}
	idle := record("idle", func() error {
		time.Sleep(150 * time.Millisecond)
		return nil
	})
	payload := qualificationPayload(profile)
	work := record("work", func() error {
		for index := 0; index < 500; index++ {
			if _, err := path.emit(t, payload); err != nil {
				return err
			}
			recovered, err := path.receivePayload(context.Background())
			if err != nil || !bytes.Equal(recovered, payload) {
				return errors.New("gateway subprocess work payload mismatch")
			}
		}
		return nil
	})
	secondIdle := record("idle-again", func() error {
		time.Sleep(150 * time.Millisecond)
		return nil
	})
	if idle.CPUSeconds > 50*time.Millisecond.Seconds() || secondIdle.CPUSeconds > 50*time.Millisecond.Seconds() {
		t.Fatalf("idle gateway CPU was reported as consumed work: first=%.6fs second=%.6fs", idle.CPUSeconds, secondIdle.CPUSeconds)
	}
	if work.CPUSeconds <= idle.CPUSeconds || work.Count != 500 || work.Mallocs == 0 ||
		work.AllocatedBytes == 0 || work.PeakHeapBytes == 0 || work.PeakGoroutines == 0 {
		t.Fatalf("gateway work resource sample = %#v idle=%#v", work, idle)
	}
	if secondIdle.Mallocs >= work.Mallocs || secondIdle.AllocatedBytes >= work.AllocatedBytes {
		t.Fatalf("successive recording inherited counters: work=%#v second=%#v", work, secondIdle)
	}
	if work.ClockOverhead <= 0 || work.ClockMethod != qualificationClockOverheadMethod {
		t.Fatalf("clock overhead evidence = %#v", work)
	}
}

func qualificationRawMeanLatency(t *testing.T, path string) time.Duration {
	t.Helper()
	file, err := os.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	defer file.Close()
	compressed, err := gzip.NewReader(file)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := io.ReadAll(compressed)
	if err != nil {
		t.Fatal(err)
	}
	if err := compressed.Close(); err != nil {
		t.Fatal(err)
	}
	var total time.Duration
	var count int
	for _, line := range strings.Split(string(raw), "\n")[1:] {
		fields := strings.Split(line, ",")
		if len(fields) < 5 || fields[4] != "delivered" {
			continue
		}
		sent, sentErr := strconv.ParseInt(fields[1], 10, 64)
		delivered, deliveredErr := strconv.ParseInt(fields[2], 10, 64)
		if sentErr != nil || deliveredErr != nil || delivered < sent {
			t.Fatalf("invalid raw latency row %q", line)
		}
		total += time.Duration(delivered - sent)
		count++
	}
	if count == 0 {
		t.Fatal("no delivered raw latency rows")
	}
	return total / time.Duration(count)
}

func TestQualificationSixImpairmentProfilesTraverseProductionPath(t *testing.T) {
	profiles := qualificationImpairmentProfiles()
	if len(profiles) != 6 {
		t.Fatalf("impairment profile count = %d, want exactly 6", len(profiles))
	}
	for _, profile := range profiles {
		observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 40,
			filepath.Join(t.TempDir(), profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if observation.Profile != profile.Name || observation.Delivered+observation.Dropped != 40 ||
			observation.RawSamplesSHA256 == "" || observation.MaxQueuePackets > qualificationImpairmentQueuePackets {
			t.Fatalf("%s observation = %#v", profile.Name, observation)
		}
	}
}

func TestQualificationLossAndSteppedThroughputBounds(t *testing.T) {
	profiles := qualificationImpairmentProfiles()
	for _, profile := range profiles[3:] {
		observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0],
			qualificationImpairmentPacketCount, filepath.Join(t.TempDir(), profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if observation.ObservedThroughputKbps <= 0 ||
			profile.Name == "constrained" && len(observation.CapacityStepObservations) != 2 {
			t.Fatalf("%s observation = %#v", profile.Name, observation)
		}
	}
}

func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) {
	qualificationTraverseProfiles(t, qualificationMediaProfiles())
	evidence, err := qualificationPacerEvidence(t, filepath.Join(t.TempDir(), "fairness.csv.gz"), 2*time.Second, 4*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 ||
		len(evidence.Series) != 10 || evidence.RawSamplesSHA256 == "" || evidence.JainIndex < 0.99 {
		t.Fatalf("pacer evidence = %#v", evidence)
	}
	for _, step := range evidence.CapacitySteps {
		if step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100 {
			t.Fatalf("capacity step = %#v", step)
		}
	}
}

func TestQualificationSmokeTraversesNativeApolloRecoveryQueuePacerAndQUIC(t *testing.T) {
	trace := qualificationProductionPathSmoke(t, qualificationMediaProfiles()[0])
	if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer ||
		!trace.VerseQUIC || !trace.PayloadPreserved {
		t.Fatalf("qualification production-path trace = %#v", trace)
	}
}