Apply M31 commands through the shared ledger

[onehouse] / driver / m31 / driver_test.go

diff --git a/driver/m31/driver_test.go b/driver/m31/driver_test.go
index 1ba61909a48c5cf8b5b8f7b54ea4750565f04158..7081f929e9077bfcf1e1c93fa171fca8be9f1223 100644 (file)
--- a/driver/m31/driver_test.go
+++ b/driver/m31/driver_test.go
@@ -9,6 +9,7 @@ import (
 
        "github.com/stretchr/testify/require"
 
+       "onehouse/driver"
        "onehouse/protocol"
 )
 
@@ -54,7 +55,7 @@ func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
-                       d, client, _ := testAssembly(t)
+                       d, client, _, router := testAssembly(t)
                        settings := Settings{
                                DeviceID:     testDeviceID,
                                Address:      "example.com:502",
@@ -64,15 +65,28 @@ func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
                        }
                        tc.change(&settings)
 
-                       _, err := New(settings, d.publisher, client)
+                       _, err := New(settings, d.publisher, router, client)
                        require.Error(t, err)
                })
        }
+
+       d, client, _, _ := testAssembly(t)
+       router, err := driver.NewRouter(nil)
+       require.NoError(t, err)
+
+       _, err = New(Settings{
+               DeviceID:     testDeviceID,
+               Address:      "example.com:502",
+               Unit:         1,
+               PollInterval: 50 * time.Millisecond,
+               Modules:      []Module{{Model: HostModel}},
+       }, d.publisher, router, client)
+       require.ErrorContains(t, err, "command queue")
 }
 
 func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
-               d, client, transport := testAssembly(t)
+               d, client, transport, router := testAssembly(t)
                client.connectErrors = []error{errors.New("unavailable")}
                client.inputs[1] = true
                ctx, cancel := context.WithCancel(t.Context())
@@ -91,6 +105,16 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
                })
                require.Empty(t, transport.ofKind(protocol.Event))
 
+               command := testCommand(t, "m0.do0", "true")
+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+
+               record, exists := d.ledger.Lookup(command.Metadata.MessageID)
+               require.True(t, exists)
+               require.Equal(t, "failed", record.Payload.Status)
+               client.withLock(func(client *fakeClient) {
+                       require.Empty(t, client.writes)
+               })
                time.Sleep(time.Second)
                synctest.Wait()
                client.withLock(func(client *fakeClient) {
@@ -132,6 +156,18 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
                        require.Len(t, client.registerWrites, 66)
                })
                require.Len(t, transport.ofKind(protocol.Event), 1)
+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Empty(t, client.writes)
+               })
+
+               command = testCommand(t, "m1.do15", "true")
+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.writes, 1)
+               })
                cancel()
                require.ErrorIs(t, <-done, context.Canceled)
                client.withLock(func(client *fakeClient) {
@@ -142,7 +178,7 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
 
 func TestDriverBoundsReconnectBackoff(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
-               d, client, _ := testAssembly(t)
+               d, client, _, _ := testAssembly(t)
                for range 8 {
                        client.withLock(func(client *fakeClient) {
                                client.connectErrors = append(client.connectErrors, errors.New("offline"))
@@ -188,7 +224,7 @@ func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
-                       d, client, transport := testAssembly(t)
+                       d, client, transport, _ := testAssembly(t)
                        failure := errors.New("independent failure")
                        if tc.closeFailure {
                                client.closeErr = failure
@@ -205,7 +241,7 @@ func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
 
 func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
-               d, client, transport := testAssembly(t)
+               d, client, transport, _ := testAssembly(t)
                ctx, cancel := context.WithCancel(t.Context())
                defer cancel()