You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
117 lines
2.9 KiB
117 lines
2.9 KiB
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 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")
|
|
}
|
|
}
|
|
}
|
|
|