test(sensors): synchronise energy counter snapshots

Sleeping fixture writers can expose only some updated domains to the second sample, producing false zero watts in CI. Run collection under synctest, wait for its sampling timer, update counters synchronously and join completion before assertions or cleanup. Retain availability, routing and wraparound assertions, add exact delta checks, and apply the same ordering to AMD fallback and RAPL preference fixtures. Production collection and sampling semantics are unchanged. Change-source: pulse-maintainer
2026-09-09 18:15:50 +00:00 · 2026-09-09 15:32:07 +01:00
parent 7a1d78944b
commit 2e3ff7494c
2 changed files with 69 additions and 90 deletions
@@ -3,11 +3,11 @@ package sensors
 import (
 	"context"
 	"errors"
+	"math"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
-	"time"
 )

 func TestCollectPower_PrefersRAPLWhenAvailable(t *testing.T) {
@@ -42,26 +42,16 @@ func TestCollectPower_PrefersRAPLWhenAvailable(t *testing.T) {
 		t.Fatalf("write amd energy: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(pkg0, "energy_uj"), "2000000"); err != nil {
-			errCh <- err
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-
-	data, err := CollectPower(ctx)
+	data, err := collectWithEnergyUpdate(t, CollectPower, update)
 	if err != nil {
 		t.Fatalf("CollectPower error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update RAPL energy: %v", err)
-	default:
-	}
 	if data.Source != "rapl" {
 		t.Fatalf("expected rapl source, got %q", data.Source)
 	}
@@ -125,30 +115,19 @@ func TestCollectAMDEnergy_LabelRoutingAndWraparound(t *testing.T) {
 		t.Fatalf("write misc label: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(hwmon, "energy1_input"), "2000000"); err != nil {
-			errCh <- err
-			return
+			t.Fatalf("update energy counter: %v", err)
 		}
 		if err := writeEnergyFile(filepath.Join(hwmon, "energy2_input"), "100"); err != nil {
-			errCh <- err
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-
-	data, err := collectAMDEnergy(ctx)
+	data, err := collectWithEnergyUpdate(t, collectAMDEnergy, update)
 	if err != nil {
 		t.Fatalf("collectAMDEnergy error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update AMD energy inputs: %v", err)
-	default:
-	}

 	if !data.Available {
 		t.Fatalf("expected available power data")
@@ -165,6 +144,9 @@ func TestCollectAMDEnergy_LabelRoutingAndWraparound(t *testing.T) {
 	if data.PackageWatts >= 1 {
 		t.Fatalf("expected wraparound-derived package watts to stay small, got %f", data.PackageWatts)
 	}
+	if math.Abs(data.CoreWatts-10) > 1e-9 || math.Abs(data.PackageWatts-0.00106) > 1e-12 {
+		t.Fatalf("expected core watts 10 and 106uJ wraparound delta, got %+v", data)
+	}
 }

 func TestFindAMDEnergyHwmon_DeterministicBranches(t *testing.T) {
@@ -3,9 +3,11 @@ package sensors
 import (
 	"context"
 	"errors"
+	"math"
 	"os"
 	"path/filepath"
 	"testing"
+	"testing/synctest"
 	"time"
 )

@@ -54,35 +56,22 @@ func TestCollectRALP_MockSysfs(t *testing.T) {
 		t.Fatalf("write dram name: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(pkg0, "energy_uj"), "2000000"); err != nil {
-			errCh <- err
-			return
+			t.Fatalf("update energy counter: %v", err)
 		}
 		if err := writeEnergyFile(filepath.Join(core, "energy_uj"), "400000"); err != nil {
-			errCh <- err
-			return
+			t.Fatalf("update energy counter: %v", err)
 		}
 		if err := writeEnergyFile(filepath.Join(dram, "energy_uj"), "600000"); err != nil {
-			errCh <- err
-			return
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-
-	data, err := collectRAPL(ctx)
+	data, err := collectWithEnergyUpdate(t, collectRAPL, update)
 	if err != nil {
 		t.Fatalf("collectRAPL error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update energy files: %v", err)
-	default:
-	}
 	if !data.Available {
 		t.Fatalf("expected data.Available true")
 	}
@@ -92,6 +81,9 @@ func TestCollectRALP_MockSysfs(t *testing.T) {
 	if data.PackageWatts <= 0 || data.CoreWatts <= 0 || data.DRAMWatts <= 0 {
 		t.Fatalf("expected non-zero watts, got %+v", data)
 	}
+	if math.Abs(data.PackageWatts-10) > 1e-9 || math.Abs(data.CoreWatts-2) > 1e-9 || math.Abs(data.DRAMWatts-3) > 1e-9 {
+		t.Fatalf("expected package/core/DRAM watts 10/2/3, got %+v", data)
+	}
 }

 func TestCollectAMDEnergy_MockSysfs(t *testing.T) {
@@ -116,26 +108,16 @@ func TestCollectAMDEnergy_MockSysfs(t *testing.T) {
 		t.Fatalf("write energy label: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(hwmon, "energy1_input"), "2000000"); err != nil {
-			errCh <- err
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-
-	data, err := collectAMDEnergy(ctx)
+	data, err := collectWithEnergyUpdate(t, collectAMDEnergy, update)
 	if err != nil {
 		t.Fatalf("collectAMDEnergy error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update energy input: %v", err)
-	default:
-	}
 	if !data.Available {
 		t.Fatalf("expected data.Available true")
 	}
@@ -175,26 +157,16 @@ func TestCollectPower_FallbackToAMD(t *testing.T) {
 		t.Fatalf("write energy label: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(hwmon, "energy1_input"), "2000000"); err != nil {
-			errCh <- err
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-
-	data, err := CollectPower(ctx)
+	data, err := collectWithEnergyUpdate(t, CollectPower, update)
 	if err != nil {
 		t.Fatalf("CollectPower error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update energy input: %v", err)
-	default:
-	}
 	if data.Source != "amd_energy" {
 		t.Fatalf("expected amd_energy fallback, got %q", data.Source)
 	}
@@ -238,7 +210,10 @@ func TestCollectRALP_CanceledDuringSampleWait(t *testing.T) {
 	}
 }

-func TestCollectAMDEnergy_CanceledDuringSampleWait(t *testing.T) {
+func TestCollectPower_NilContextFallsBackToAMD(t *testing.T) {
+	originalRAPL := raplBasePath
+	raplBasePath = filepath.Join(t.TempDir(), "missing-rapl")
+	t.Cleanup(func() { raplBasePath = originalRAPL })
 	tmpDir := t.TempDir()
 	original := hwmonBasePath
 	hwmonBasePath = tmpDir
@@ -260,24 +235,46 @@ func TestCollectAMDEnergy_CanceledDuringSampleWait(t *testing.T) {
 		t.Fatalf("write energy label: %v", err)
 	}

-	errCh := make(chan error, 1)
-	go func() {
-		time.Sleep(20 * time.Millisecond)
+	update := func(t *testing.T) {
 		if err := writeEnergyFile(filepath.Join(hwmon, "energy1_input"), "2000000"); err != nil {
-			errCh <- err
+			t.Fatalf("update energy counter: %v", err)
 		}
-	}()
+	}

-	data, err := CollectPower(nil)
+	data, err := collectWithEnergyUpdate(t, func(context.Context) (*PowerData, error) { return CollectPower(nil) }, update)
 	if err != nil {
 		t.Fatalf("CollectPower error: %v", err)
 	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("update energy input: %v", err)
-	default:
-	}
 	if data.Source != "amd_energy" {
 		t.Fatalf("expected amd_energy fallback, got %q", data.Source)
 	}
 }
+
+// collectWithEnergyUpdate makes the fixture's two counter snapshots independent
+// of host scheduling. The collector must be waiting on its sample timer before
+// any writes begin; virtual time cannot advance while the updater is running.
+// Unlike a sleeping writer goroutine, this cannot expose a partial update or
+// leave writes running after the assertion/temporary-directory cleanup.
+func collectWithEnergyUpdate(t *testing.T, collect func(context.Context) (*PowerData, error), update func(*testing.T)) (*PowerData, error) {
+	t.Helper()
+	var data *PowerData
+	var err error
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		defer func() { cancel(); <-done }()
+		go func() {
+			defer close(done)
+			data, err = collect(ctx)
+		}()
+		synctest.Wait()
+		select {
+		case <-done:
+			t.Fatalf("collector returned before counter update: data=%+v err=%v", data, err)
+		default:
+		}
+		update(t)
+		<-done
+	})
+	return data, err
+}