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         err      error
 189 }
 190
 191 func (f *fakeTransport) Publish(ctx context.Context, publication protocol.Publication) error {
 192         f.mu.Lock()
 193         defer f.mu.Unlock()
 194
 195         message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
 196         if err != nil {
 197                 return fmt.Errorf("parse fake publication: %w", err)
 198         }
 199
 200         if message.Address.Kind == f.failKind && f.err != nil {
 201                 return f.err
 202         }
 203
 204         if ctx.Err() != nil {
 205                 return ctx.Err()
 206         }
 207
 208         f.messages = append(f.messages, message)
 209
 210         return nil
 211 }
 212
 213 func (f *fakeTransport) ofKind(kind protocol.Kind) []protocol.Message[any] {
 214         f.mu.Lock()
 215         defer f.mu.Unlock()
 216
 217         var messages []protocol.Message[any]
 218
 219         for _, message := range f.messages {
 220                 if message.Address.Kind == kind {
 221                         messages = append(messages, message)
 222                 }
 223         }
 224
 225         return messages
 226 }
 227
 228 func (f *fakeTransport) all() []protocol.Message[any] {
 229         f.mu.Lock()
 230         defer f.mu.Unlock()
 231
 232         return slices.Clone(f.messages)
 233 }
 234
 235 func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport, *driver.Router) {
 236         t.Helper()
 237
 238         transport := &fakeTransport{}
 239         publisher := driver.NewPublisher(transport)
 240         router, err := driver.NewRouter([]string{testDeviceID})
 241         require.NoError(t, err)
 242
 243         client := &fakeClient{
 244                 inputs:    make([]bool, 16),
 245                 coils:     make([]bool, 32),
 246                 analog:    make([]uint16, 8),
 247                 registers: make(map[uint16]uint16),
 248                 mismatch:  -1,
 249         }
 250         settings := Settings{
 251                 DeviceID:     testDeviceID,
 252                 Address:      "example.com:502",
 253                 Unit:         1,
 254                 PollInterval: 50 * time.Millisecond,
 255                 Modules: []Module{
 256                         {Model: HostModel, PulseWidths: map[int]time.Duration{2: 200 * time.Millisecond}},
 257                         {Model: OutputModel},
 258                         {Model: AnalogModel},
 259                         {Model: MixedModel},
 260                 },
 261         }
 262         d, err := New(settings, publisher, router, client)
 263         require.NoError(t, err)
 264
 265         return d, client, transport, router
 266 }
 267
 268 func testCommand(t *testing.T, capability string, value string) driver.Command {
 269         t.Helper()
 270
 271         id, err := protocol.NewID(time.Now())
 272         require.NoError(t, err)
 273
 274         expires := time.Now().UTC().Add(time.Minute)
 275
 276         return driver.Command{
 277                 DeviceID:   testDeviceID,
 278                 Capability: capability,
 279                 Metadata: protocol.Metadata{
 280                         MessageID: id,
 281                         ExpiresAt: &expires,
 282                 },
 283                 Payload: protocol.ValuePayload{Value: []byte(value)},
 284         }
 285 }
 286
 287 func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
 288         t.Helper()
 289
 290         d, client, transport, _ := testAssembly(t)
 291         connection := &connection{owner: d}
 292         require.NoError(t, connection.start(t.Context()))
 293
 294         transport.messages = nil
 295         client.writes = nil
 296         client.coilReads = 0
 297
 298         return connection, client, transport
 299 }
 300
 301 func queueCommands(t *testing.T, router *driver.Router, commands []driver.Command) {
 302         t.Helper()
 303
 304         var (
 305                 publication protocol.Publication
 306                 err         error
 307         )
 308
 309         if len(commands) == 1 {
 310                 command := commands[0]
 311                 publication, err = protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
 312         } else {
 313                 members := make([]protocol.BatchCommand, len(commands))
 314
 315                 for index, command := range commands {
 316                         members[index] = protocol.BatchCommand{
 317                                 Capability: command.Capability,
 318                                 Metadata:   command.Metadata,
 319                                 Payload:    command.Payload,
 320                         }
 321                 }
 322
 323                 publication, err = protocol.BuildCommandBatch(testDeviceID, commands[0].Metadata, protocol.BatchPayload{Commands: members})
 324         }
 325
 326         require.NoError(t, err)
 327         require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
 328 }