mqqt_washingMachine/internal/sensor/sampler_test.go

package sensor

import (
	"context"
	"reflect"
	"sync"
	"testing"
	"testing/synctest"
	"time"

	"mqtt_washingmachine/internal/status"
)

func TestSamplerIntervals(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		var sampler Sampler
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()
		ticks := make(chan time.Time)
		done := make(chan struct{})
		go func() { sampler.Run(ctx, ticks); close(done) }()
		tick := func() { ticks <- time.Now(); synctest.Wait() }
		receive := func(payload string) {
			t.Helper()
			if err := sampler.Receive([]byte(payload)); err != nil {
				t.Fatal(err)
			}
		}
		if got := sampler.Status(); got.State != "waiting" || len(got.Readings) != 0 {
			t.Fatalf("startup: %+v", got)
		}
		receive(`{"ENERGY":{"Power":100}}`)
		receive(`{"ENERGY":{"Power":47.7}}`)
		if err := sampler.Receive([]byte(`{"ENERGY":{}}`)); err == nil {
			t.Fatal("invalid payload accepted")
		}
		if got := sampler.Status(); len(got.Readings) != 0 {
			t.Fatal("message recorded before tick")
		}
		tick()
		if got := sampler.Status(); !reflect.DeepEqual(got.Readings, []status.Sample{{Power: 47.7, Valid: true}}) {
			t.Fatalf("latest valid measurement lost: %+v", got)
		}
		if err := sampler.Receive([]byte(`broken`)); err == nil {
			t.Fatal("malformed payload accepted")
		}
		tick()
		receive(`{"ENERGY":{"Power":0}}`)
		tick()
		want := []status.Sample{{Power: 47.7, Valid: true}, {}, {Valid: true}}
		if got := sampler.Status(); got.State != "incomplete" || !reflect.DeepEqual(got.Readings, want) {
			t.Fatalf("missing interval: %+v", got)
		}
		for range 2 {
			receive(`{"ENERGY":{"Power":2.99}}`)
			tick()
		}
		if got := sampler.Status(); got.State != "done" {
			t.Fatalf("gap did not age out: %+v", got)
		}
		receive(`{"ENERGY":{"Power":3.00}}`)
		tick()
		if got := sampler.Status(); got.State != "running" {
			t.Fatalf("threshold: %+v", got)
		}
		cancel()
		synctest.Wait()
		select {
		case <-done:
		default:
			t.Fatal("Run did not stop on cancellation")
		}
	})
}

func TestSamplerClosedTicks(t *testing.T) {
	var sampler Sampler
	ticks := make(chan time.Time)
	close(ticks)
	sampler.Run(t.Context(), ticks)
	if len(sampler.Status().Readings) != 0 {
		t.Fatal("closed ticks created a sample")
	}
}

func TestSamplerOnReceive(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		var sampler Sampler
		calls := 0
		ticks := make(chan time.Time)
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()
		go sampler.Run(ctx, ticks)
		sampler.OnReceive = func() {
			calls++
			// Both operations acquire the sampler lock: the callback must run
			// after storage and outside that lock.
			ticks <- time.Now()
			synctest.Wait()
			got := sampler.Status().Readings
			if got[len(got)-1] != (status.Sample{Power: 47.7, Valid: true}) {
				t.Errorf("callback ran before storage: %+v", got)
			}
		}
		for range 2 {
			if err := sampler.Receive([]byte(`{"ENERGY":{"Power":47.7}}`)); err != nil {
				t.Fatal(err)
			}
		}
		for _, payload := range []string{`broken`, `{}`, `{"ENERGY":{}}`, `{"ENERGY":{"Power":"bad"}}`} {
			if err := sampler.Receive([]byte(payload)); err == nil {
				t.Fatalf("accepted %s", payload)
			}
		}
		if calls != 2 {
			t.Fatalf("callback calls = %d; want 2", calls)
		}
	})
}

func TestSamplerConcurrentAccess(t *testing.T) {
	var sampler Sampler
	ctx, cancel := context.WithCancel(t.Context())
	ticks := make(chan time.Time)
	done := make(chan struct{})
	go func() { sampler.Run(ctx, ticks); close(done) }()
	var workers sync.WaitGroup
	for range 10 {
		workers.Go(func() {
			for range 100 {
				if err := sampler.Receive([]byte(`{"ENERGY":{"Power":1}}`)); err != nil {
					t.Error(err)
				}
				snapshot := sampler.Status()
				if len(snapshot.Readings) > 0 {
					snapshot.Readings[0].Power = 999
				}
			}
		})
	}
	for range 100 {
		ticks <- time.Now()
	}
	workers.Wait()
	cancel()
	<-done
	for _, sample := range sampler.Status().Readings {
		if sample.Power == 999 {
			t.Fatal("snapshot changed shared state")
		}
	}
}