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. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
95 lines
2.9 KiB
Go
95 lines
2.9 KiB
Go
package devicecontrolclient
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"google.golang.org/grpc"
|
|
|
|
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
|
|
)
|
|
|
|
type fakeGRPCClient struct {
|
|
lastTurnOn *devicecontrol.TurnOnRequest
|
|
lastTurnOff *devicecontrol.TurnOffRequest
|
|
lastSetLevel *devicecontrol.SetLevelRequest
|
|
result *devicecontrol.CommandResult
|
|
err error
|
|
}
|
|
|
|
func (f *fakeGRPCClient) TurnOn(_ context.Context, in *devicecontrol.TurnOnRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
|
|
f.lastTurnOn = in
|
|
return f.result, f.err
|
|
}
|
|
|
|
func (f *fakeGRPCClient) TurnOff(_ context.Context, in *devicecontrol.TurnOffRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
|
|
f.lastTurnOff = in
|
|
return f.result, f.err
|
|
}
|
|
|
|
func (f *fakeGRPCClient) SetLevel(_ context.Context, in *devicecontrol.SetLevelRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
|
|
f.lastSetLevel = in
|
|
return f.result, f.err
|
|
}
|
|
|
|
func TestDispatch_TurnOn(t *testing.T) {
|
|
fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: true}}
|
|
c := newWithClient(fake)
|
|
|
|
res, err := c.Dispatch(context.Background(), "fan-1", "turn_on", nil)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
if !res.Success {
|
|
t.Fatalf("got %+v, want success", res)
|
|
}
|
|
if fake.lastTurnOn == nil || fake.lastTurnOn.DeviceId != "fan-1" {
|
|
t.Fatalf("got TurnOn request %+v, want device_id=fan-1", fake.lastTurnOn)
|
|
}
|
|
}
|
|
|
|
func TestDispatch_SetLevel(t *testing.T) {
|
|
fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: true}}
|
|
c := newWithClient(fake)
|
|
|
|
res, err := c.Dispatch(context.Background(), "light-1", "set_level", map[string]any{"level": 42.5})
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
if !res.Success {
|
|
t.Fatalf("got %+v, want success", res)
|
|
}
|
|
if fake.lastSetLevel == nil || fake.lastSetLevel.Level != 42.5 {
|
|
t.Fatalf("got SetLevel request %+v, want level=42.5", fake.lastSetLevel)
|
|
}
|
|
}
|
|
|
|
func TestDispatch_SetLevel_MissingParam(t *testing.T) {
|
|
c := newWithClient(&fakeGRPCClient{})
|
|
|
|
if _, err := c.Dispatch(context.Background(), "light-1", "set_level", nil); err == nil {
|
|
t.Fatal("expected error for missing level param")
|
|
}
|
|
}
|
|
|
|
func TestDispatch_UnknownActionType(t *testing.T) {
|
|
c := newWithClient(&fakeGRPCClient{})
|
|
|
|
if _, err := c.Dispatch(context.Background(), "d1", "dance", nil); err == nil {
|
|
t.Fatal("expected error for unknown action_type")
|
|
}
|
|
}
|
|
|
|
func TestDispatch_BusinessFailurePassesThrough(t *testing.T) {
|
|
fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: false, Error: "device offline"}}
|
|
c := newWithClient(fake)
|
|
|
|
res, err := c.Dispatch(context.Background(), "fan-1", "turn_off", nil)
|
|
if err != nil {
|
|
t.Fatalf("unexpected transport error: %v", err)
|
|
}
|
|
if res.Success || res.Error != "device offline" {
|
|
t.Fatalf("got %+v, want business failure passed through", res)
|
|
}
|
|
}
|