forked from GiteaTest2015/streamsql
rulego-team
94 lines
2.4 KiB
Go
94 lines
2.4 KiB
Go
package window
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/rulego/streamsql/types"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestCountingWindow(t *testing.T) {
|
|
_, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
// Test case 1: Normal operation
|
|
cw, _ := NewCountingWindow(types.WindowConfig{
|
|
Params: map[string]interface{}{
|
|
"count": 3,
|
|
"callback": func(results []interface{}) {
|
|
t.Logf("Received results: %v", results)
|
|
},
|
|
},
|
|
})
|
|
go cw.Start()
|
|
|
|
for i := 0; i < 3; i++ {
|
|
cw.Add(i)
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
|
|
// Trigger one more element to check threshold
|
|
cw.Add(3)
|
|
|
|
resultsChan := cw.OutputChan()
|
|
//results := make(chan []types.Row)
|
|
// go func() {
|
|
// for res := range cw.OutputChan() {
|
|
// results <- res
|
|
// }
|
|
// }()
|
|
|
|
select {
|
|
case res := <-resultsChan:
|
|
assert.Len(t, res, 3)
|
|
assert.Equal(t, 0, res[0].Data, "First element should be 0")
|
|
assert.Equal(t, 1, res[1].Data, "Second element should be 1")
|
|
assert.Equal(t, 2, res[2].Data, "Third element should be 2")
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("No results received within timeout")
|
|
}
|
|
// Verify window state: After adding 4th data, first window has triggered, remaining 1 data (value 3)
|
|
// Continue adding 2 more data, should trigger again
|
|
cw.Add(4) // Add 5th data
|
|
cw.Add(5) // Add 6th data, should trigger again (3,4,5)
|
|
|
|
// Wait for second trigger
|
|
select {
|
|
case res := <-resultsChan:
|
|
assert.Len(t, res, 3)
|
|
assert.Equal(t, 3, res[0].Data, "First element of second batch should be 3")
|
|
assert.Equal(t, 4, res[1].Data, "Second element of second batch should be 4")
|
|
assert.Equal(t, 5, res[2].Data, "Third element of second batch should be 5")
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("No second results received within timeout")
|
|
}
|
|
|
|
// Test case 2: Reset
|
|
cw.Reset()
|
|
// Add data after reset to verify reset was successful
|
|
cw.Add(100)
|
|
cw.Add(101)
|
|
cw.Add(102)
|
|
select {
|
|
case res := <-resultsChan:
|
|
assert.Len(t, res, 3)
|
|
assert.Equal(t, 100, res[0].Data, "First element after reset should be 100")
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("No results after reset received within timeout")
|
|
}
|
|
}
|
|
|
|
func TestCountingWindowBadThreshold(t *testing.T) {
|
|
_, err := CreateWindow(types.WindowConfig{
|
|
Type: "counting",
|
|
Params: map[string]interface{}{
|
|
"count": 0,
|
|
},
|
|
})
|
|
require.Error(t, err)
|
|
}
|