cacto
abc83faf05
Слушает telemetry.new_reading (ручной ack/nack, реквизишн только при транспортных ошибках gRPC), кэширует активные automation_rules в памяти с периодическим обновлением из PostgreSQL (JOIN с devices — резолвит внутренние ID в external_id, которым оперируют MQTT/Redis/gRPC). Условия правил (>,<,>=,<=,=,!=) оцениваются обобщённо, без привязки к конкретным типам устройств. При срабатывании вызывает device-control-service по gRPC и публикует automation.rule_triggered в RabbitMQ. Проверено сквозным тестом через docker compose на полном пайплайне: mosquitto_pub → ingest-service (ClickHouse + telemetry.new_reading) → rule-engine-service (совпадение правила) → device-control-service (gRPC → Redis desired_state + MQTT-команда) → automation.rule_triggered. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
111 lines
2.9 KiB
Go
111 lines
2.9 KiB
Go
package rulecache
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type fakeFetcher struct {
|
|
mu sync.Mutex
|
|
rows []Row
|
|
calls chan struct{}
|
|
}
|
|
|
|
func (f *fakeFetcher) setRows(rows []Row) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.rows = rows
|
|
}
|
|
|
|
func (f *fakeFetcher) FetchActiveRules(context.Context) ([]Row, error) {
|
|
f.mu.Lock()
|
|
rows := append([]Row(nil), f.rows...)
|
|
f.mu.Unlock()
|
|
if f.calls != nil {
|
|
f.calls <- struct{}{}
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
func discardLogger() *slog.Logger {
|
|
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
|
}
|
|
|
|
func TestCache_StartLoadsSynchronously(t *testing.T) {
|
|
fetcher := &fakeFetcher{rows: []Row{
|
|
{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ConditionOperator: ">", ConditionValue: 28, TargetExternalID: "fan-1", ActionType: "turn_on"},
|
|
}}
|
|
c := New(fetcher, time.Hour, discardLogger())
|
|
|
|
if err := c.Start(context.Background()); err != nil {
|
|
t.Fatalf("start: %v", err)
|
|
}
|
|
defer c.Stop()
|
|
|
|
got := c.MatchingRules("sensor-1", "temperature")
|
|
if len(got) != 1 || got[0].TargetDeviceID != "fan-1" {
|
|
t.Fatalf("got %+v, want one rule targeting fan-1", got)
|
|
}
|
|
|
|
// Different sensor_type on the same device must not match.
|
|
if got := c.MatchingRules("sensor-1", "humidity"); len(got) != 0 {
|
|
t.Fatalf("got %+v, want no rules for humidity", got)
|
|
}
|
|
}
|
|
|
|
func TestCache_SkipsRuleWithInvalidActionParams(t *testing.T) {
|
|
fetcher := &fakeFetcher{rows: []Row{
|
|
{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ActionParams: []byte(`not json`)},
|
|
{RuleID: 2, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ActionParams: []byte(`{"level":50}`)},
|
|
}}
|
|
c := New(fetcher, time.Hour, discardLogger())
|
|
if err := c.Start(context.Background()); err != nil {
|
|
t.Fatalf("start: %v", err)
|
|
}
|
|
defer c.Stop()
|
|
|
|
got := c.MatchingRules("sensor-1", "temperature")
|
|
if len(got) != 1 || got[0].ID != 2 {
|
|
t.Fatalf("got %+v, want only rule 2 to survive", got)
|
|
}
|
|
}
|
|
|
|
func TestCache_RefreshesOnInterval(t *testing.T) {
|
|
calls := make(chan struct{}, 4)
|
|
fetcher := &fakeFetcher{calls: calls}
|
|
c := New(fetcher, 20*time.Millisecond, discardLogger())
|
|
if err := c.Start(context.Background()); err != nil {
|
|
t.Fatalf("start: %v", err)
|
|
}
|
|
defer c.Stop()
|
|
|
|
<-calls // consume the initial synchronous load
|
|
|
|
fetcher.setRows([]Row{
|
|
{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", TargetExternalID: "fan-1", ActionType: "turn_on"},
|
|
})
|
|
|
|
select {
|
|
case <-calls:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("timed out waiting for periodic refresh")
|
|
}
|
|
|
|
got := c.MatchingRules("sensor-1", "temperature")
|
|
if len(got) != 1 {
|
|
t.Fatalf("got %+v after refresh, want the newly added rule", got)
|
|
}
|
|
}
|
|
|
|
func TestCache_StopWaitsForLoopExit(t *testing.T) {
|
|
c := New(&fakeFetcher{}, time.Hour, discardLogger())
|
|
if err := c.Start(context.Background()); err != nil {
|
|
t.Fatalf("start: %v", err)
|
|
}
|
|
c.Stop() // must return without hanging
|
|
}
|