git.giacomolli.uk - onehouse/blob - driver/m31/fake_helper_test.go

   1 package m31
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "slices"
   8         "sync"
   9         "testing"
  10         "time"
  11
  12         "github.com/stretchr/testify/require"
  13
  14         "onehouse/driver"
  15         "onehouse/protocol"
  16 )
  17
  18 const testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
  19
  20 type fakeClient struct {
  21         registerReadErr error
  22         wrongRegister   uint16
  23         failAfterWrites int
  24         mu              sync.Mutex
  25         inputs          []bool
  26         coils           []bool
  27         analog          []uint16
  28         registers       map[uint16]uint16
  29         writes          []coilWrite
  30         registerWrites  []uint16
  31         attempts        []time.Time
  32         closes          int
  33         coilReads       int
  34         connectErrors   []error
  35         inputErr        error
  36         readErr         error
  37         writeErr        error
  38         registerErr     error
  39         closeErr        error
  40         wrongConfig     bool
  41         shortRead       bool
  42         mismatch        int
  43         writeDelay      time.Duration
  44         inputDelay      time.Duration
  45 }
  46
  47 func (c *fakeClient) Connect(ctx context.Context, _ string, _ byte) error {
  48         c.mu.Lock()
  49         defer c.mu.Unlock()
  50
  51         c.attempts = append(c.attempts, time.Now())
  52         if len(c.connectErrors) != 0 {
  53                 err := c.connectErrors[0]
  54                 c.connectErrors = c.connectErrors[1:]
  55
  56                 return err
  57         }
  58
  59         return ctx.Err()
  60 }
  61
  62 func (c *fakeClient) Close() error {
  63         c.mu.Lock()
  64         defer c.mu.Unlock()
  65
  66         c.closes++
  67
  68         return c.closeErr
  69 }
  70
  71 func (c *fakeClient) ReadInputs(ctx context.Context, _ uint16) ([]bool, error) {
  72         c.mu.Lock()
  73         defer c.mu.Unlock()
  74
  75         time.Sleep(c.inputDelay)
  76
  77         if c.inputErr != nil || ctx.Err() != nil {
  78                 return nil, errors.Join(c.inputErr, ctx.Err())
  79         }
  80
  81         return slices.Clone(c.inputs), nil
  82 }
  83
  84 func (c *fakeClient) ReadCoils(ctx context.Context, _ uint16) ([]bool, error) {
  85         c.mu.Lock()
  86         defer c.mu.Unlock()
  87
  88         c.coilReads++
  89         if c.readErr != nil || ctx.Err() != nil {
  90                 return nil, errors.Join(c.readErr, ctx.Err())
  91         }
  92
  93         if c.shortRead {
  94                 return nil, nil
  95         }
  96
  97         return slices.Clone(c.coils), nil
  98 }
  99
 100 func (c *fakeClient) ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error) {
 101         c.mu.Lock()
 102         defer c.mu.Unlock()
 103
 104         if c.registerErr != nil || c.registerReadErr != nil || ctx.Err() != nil {
 105                 return nil, errors.Join(c.registerErr, c.registerReadErr, ctx.Err())
 106         }
 107
 108         if address >= pulseWidthBase {
 109                 value := c.registers[address]
 110                 if c.wrongConfig || address == c.wrongRegister {
 111                         value++
 112                 }
 113
 114                 return []uint16{value}, nil
 115         }
 116
 117         return slices.Clone(c.analog[:min(int(count), len(c.analog))]), nil
 118 }
 119
 120 func (c *fakeClient) WriteCoil(ctx context.Context, address uint16, value bool) error {
 121         return c.write(ctx, address, []bool{value}, false)
 122 }
 123
 124 func (c *fakeClient) WriteCoils(ctx context.Context, address uint16, values []bool) error {
 125         return c.write(ctx, address, values, true)
 126 }
 127
 128 func (c *fakeClient) write(ctx context.Context, address uint16, values []bool, multiple bool) error {
 129         c.mu.Lock()
 130         defer c.mu.Unlock()
 131
 132         time.Sleep(c.writeDelay)
 133
 134         if c.writeErr != nil && len(c.writes) >= c.failAfterWrites || ctx.Err() != nil {
 135                 return errors.Join(c.writeErr, ctx.Err())
 136         }
 137
 138         c.writes = append(c.writes, coilWrite{
 139                 address:  address,
 140                 values:   slices.Clone(values),
 141                 multiple: multiple,
 142         })
 143
 144         for offset, value := range values {
 145                 if int(address)+offset != c.mismatch {
 146                         c.coils[int(address)+offset] = value
 147                 }
 148         }
 149
 150         return nil
 151 }
 152
 153 func (c *fakeClient) WriteRegister(ctx context.Context, address uint16, value uint16) error {
 154         c.mu.Lock()
 155         defer c.mu.Unlock()
 156
 157         if c.registerErr != nil || ctx.Err() != nil {
 158                 return errors.Join(c.registerErr, ctx.Err())
 159         }
 160
 161         c.registerWrites = append(c.registerWrites, address)
 162         if address >= pulseWidthBase {
 163                 c.registers[address] = value
 164         } else if int(address) != c.mismatch {
 165                 c.analog[address] = value
 166         }
 167
 168         return nil
 169 }
 170
 171 func (c *fakeClient) withLock(fn func(*fakeClient)) {
 172         c.mu.Lock()
 173         defer c.mu.Unlock()
 174
 175         fn(c)
 176 }
 177
 178 type coilWrite struct {
 179         address  uint16
 180         values   []bool
 181         multiple bool
 182 }
 183
 184 type fakeTransport struct {
 185         mu       sync.Mutex
 186         messages []protocol.Message[any]
 187         failKind protocol.Kind
 188         failOnce bool
 189         err      error
 190 }
 191
 192 func (f *fakeTransport) Publish(ctx context.Context, publication protocol.Publication) error {
 193         f.mu.Lock()
 194         defer f.mu.Unlock()
 195
 196         message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
 197         if err != nil {
 198                 return fmt.Errorf("parse fake publication: %w", err)
 199         }
 200
 201         if message.Address.Kind == f.failKind && f.err != nil {
 202                 err = f.err
 203                 if f.failOnce {
 204                         f.err = nil
 205                 }
 206
 207                 return err
 208         }
 209
 210         if ctx.Err() != nil {
 211                 return ctx.Err()
 212         }
 213
 214         f.messages = append(f.messages, message)
 215
 216         return nil
 217 }
 218
 219 func (f *fakeTransport) ofKind(kind protocol.Kind) []protocol.Message[any] {
 220         f.mu.Lock()
 221         defer f.mu.Unlock()
 222
 223         var messages []protocol.Message[any]
 224
 225         for _, message := range f.messages {
 226                 if message.Address.Kind == kind {
 227                         messages = append(messages, message)
 228                 }
 229         }
 230
 231         return messages
 232 }
 233
 234 func (f *fakeTransport) all() []protocol.Message[any] {
 235         f.mu.Lock()
 236         defer f.mu.Unlock()
 237
 238         return slices.Clone(f.messages)
 239 }
 240
 241 func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport, *driver.Router) {
 242         t.Helper()
 243
 244         transport := &fakeTransport{}
 245         publisher := driver.NewPublisher(transport)
 246         router, err := driver.NewRouter([]string{testDeviceID})
 247         require.NoError(t, err)
 248
 249         client := &fakeClient{
 250                 inputs:    make([]bool, 16),
 251                 coils:     make([]bool, 32),
 252                 analog:    make([]uint16, 8),
 253                 registers: make(map[uint16]uint16),
 254                 mismatch:  -1,
 255         }
 256         settings := Settings{
 257                 DeviceID:     testDeviceID,
 258                 Address:      "example.com:502",
 259                 Unit:         1,
 260                 PollInterval: 50 * time.Millisecond,
 261                 Modules: []Module{
 262                         {Model: HostModel, PulseWidths: map[int]time.Duration{2: 200 * time.Millisecond}},
 263                         {Model: OutputModel},
 264                         {Model: AnalogModel},
 265                         {Model: MixedModel},
 266                 },
 267         }
 268         d, err := New(settings, publisher, router, client)
 269         require.NoError(t, err)
 270
 271         return d, client, transport, router
 272 }
 273
 274 func testCommand(t *testing.T, capability string, value string) driver.Command {
 275         t.Helper()
 276
 277         id, err := protocol.NewID(time.Now())
 278         require.NoError(t, err)
 279
 280         expires := time.Now().UTC().Add(time.Minute)
 281
 282         return driver.Command{
 283                 DeviceID:   testDeviceID,
 284                 Capability: capability,
 285                 Metadata: protocol.Metadata{
 286                         MessageID: id,
 287                         ExpiresAt: &expires,
 288                 },
 289                 Payload: protocol.ValuePayload{Value: []byte(value)},
 290         }
 291 }
 292
 293 func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
 294         t.Helper()
 295
 296         d, client, transport, _ := testAssembly(t)
 297         connection := &connection{owner: d}
 298         require.NoError(t, connection.start(t.Context()))
 299
 300         transport.messages = nil
 301         client.writes = nil
 302         client.coilReads = 0
 303
 304         return connection, client, transport
 305 }
 306
 307 func queueCommands(t *testing.T, router *driver.Router, commands []driver.Command) {
 308         t.Helper()
 309
 310         var (
 311                 publication protocol.Publication
 312                 err         error
 313         )
 314
 315         if len(commands) == 1 {
 316                 command := commands[0]
 317                 publication, err = protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
 318         } else {
 319                 members := make([]protocol.BatchCommand, len(commands))
 320
 321                 for index, command := range commands {
 322                         members[index] = protocol.BatchCommand{
 323                                 Capability: command.Capability,
 324                                 Metadata:   command.Metadata,
 325                                 Payload:    command.Payload,
 326                         }
 327                 }
 328
 329                 publication, err = protocol.BuildCommandBatch(testDeviceID, commands[0].Metadata, protocol.BatchPayload{Commands: members})
 330         }
 331
 332         require.NoError(t, err)
 333         require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
 334 }