chore: convert watch_test into an integration test (#2511)

Not sure if this makes the watch tests much more readable, but at least
they are more consistent with the rest of the integration tests.

Closes #1585

internal/buildengine/watch_integration_test.go

@@ -0,0 +1,205 @@
+//go:build integration
+
+package buildengine_test
+
+import (
+	"context" //nolint:depguard
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/alecthomas/assert/v2"
+	"github.com/alecthomas/types/pubsub"
+
+	. "github.com/TBD54566975/ftl/internal/buildengine"
+	in "github.com/TBD54566975/ftl/internal/integration"
+	"github.com/TBD54566975/ftl/internal/moduleconfig"
+)
+
+const pollFrequency = time.Millisecond * 500
+
+func TestWatch(t *testing.T) {
+	var events chan WatchEvent
+	var topic *pubsub.Topic[WatchEvent]
+	var one, two Module
+	w := NewWatcher()
+
+	in.Run(t,
+		func(tb testing.TB, ic in.TestContext) {
+			events, topic = startWatching(ic, t, w, ic.WorkingDir())
+		},
+		// Add modules
+		in.FtlNew("go", "one"),
+		in.FtlNew("go", "two"),
+		func(tb testing.TB, ic in.TestContext) {
+			one = loadModule(t, ic.WorkingDir(), "one")
+			two = loadModule(t, ic.WorkingDir(), "two")
+		},
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(tb, events, []WatchEvent{
+				WatchEventModuleAdded{Module: one},
+				WatchEventModuleAdded{Module: two},
+			})
+		},
+
+		// Delete and modify a module
+		in.RemoveDir("two"),
+		updateModFile("one"),
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(tb, events, []WatchEvent{
+				WatchEventModuleChanged{Module: one},
+				WatchEventModuleRemoved{Module: two},
+			})
+		},
+
+		// Cleanup
+		func(tb testing.TB, ic in.TestContext) {
+			topic.Close()
+		},
+	)
+}
+
+func TestWatchWithBuildModifyingFiles(t *testing.T) {
+	var events chan WatchEvent
+	var topic *pubsub.Topic[WatchEvent]
+	var transaction ModifyFilesTransaction
+	w := NewWatcher()
+
+	in.Run(t,
+		func(tb testing.TB, ic in.TestContext) {
+			events, topic = startWatching(ic, t, w, ic.WorkingDir())
+		},
+
+		in.FtlNew("go", "one"),
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(tb, events, []WatchEvent{
+				WatchEventModuleAdded{Module: loadModule(t, ic.WorkingDir(), "one")},
+			})
+		},
+		func(tb testing.TB, ic in.TestContext) {
+			transaction = w.GetTransaction(filepath.Join(ic.WorkingDir(), "one"))
+			err := transaction.Begin()
+			assert.NoError(t, err)
+		},
+		updateModFile("one"),
+		func(tb testing.TB, ic in.TestContext) {
+			err := transaction.ModifiedFiles(filepath.Join(ic.WorkingDir(), "one", "go.mod"))
+			assert.NoError(t, err)
+		},
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(t, events, []WatchEvent{})
+			topic.Close()
+		},
+	)
+}
+
+func TestWatchWithBuildAndUserModifyingFiles(t *testing.T) {
+	var events chan WatchEvent
+	var topic *pubsub.Topic[WatchEvent]
+	var transaction ModifyFilesTransaction
+	w := NewWatcher()
+
+	in.Run(t,
+		func(tb testing.TB, ic in.TestContext) {
+			events, topic = startWatching(ic, t, w, ic.WorkingDir())
+		},
+
+		in.FtlNew("go", "one"),
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(tb, events, []WatchEvent{
+				WatchEventModuleAdded{Module: loadModule(t, ic.WorkingDir(), "one")},
+			})
+		},
+		// Change a file in a module, within a transaction
+		func(tb testing.TB, ic in.TestContext) {
+			transaction = w.GetTransaction(filepath.Join(ic.WorkingDir(), "one"))
+			err := transaction.Begin()
+			assert.NoError(t, err)
+		},
+		updateModFile("one"),
+		// Change a file in a module, without a transaction (user change)
+		in.MoveFile("one", "one.go", "one_.go"),
+		func(tb testing.TB, ic in.TestContext) {
+			err := transaction.End()
+			assert.NoError(t, err)
+		},
+		func(tb testing.TB, ic in.TestContext) {
+			waitForEvents(t, events, []WatchEvent{
+				WatchEventModuleChanged{Module: loadModule(t, ic.WorkingDir(), "one")},
+			})
+			topic.Close()
+		},
+	)
+}
+
+func loadModule(t *testing.T, dir, name string) Module {
+	t.Helper()
+	config, err := moduleconfig.LoadModuleConfig(filepath.Join(dir, name))
+	assert.NoError(t, err)
+	return Module{
+		Config: config,
+	}
+}
+
+func startWatching(ctx context.Context, t testing.TB, w *Watcher, dir string) (chan WatchEvent, *pubsub.Topic[WatchEvent]) {
+	t.Helper()
+	events := make(chan WatchEvent, 128)
+	topic, err := w.Watch(ctx, pollFrequency, []string{dir})
+	assert.NoError(t, err)
+	topic.Subscribe(events)
+
+	return events, topic
+}
+
+// waitForEvents waits for the expected events to be received on the events channel.
+//
+// It always waits for longer than just the expected events to confirm that no other events are received.
+// The expected events are matched by keyForEvent.
+func waitForEvents(t testing.TB, events chan WatchEvent, expected []WatchEvent) {
+	t.Helper()
+	visited := map[string]bool{}
+	expectedKeys := []string{}
+	for _, event := range expected {
+		key := keyForEvent(event)
+		visited[key] = false
+		expectedKeys = append(expectedKeys, key)
+	}
+	eventCount := 0
+	for {
+		select {
+		case actual := <-events:
+			key := keyForEvent(actual)
+			hasVisited, isExpected := visited[key]
+			assert.True(t, isExpected, "unexpected event %v instead of %v", key, expectedKeys)
+			assert.False(t, hasVisited, "duplicate event %v", key)
+			visited[key] = true
+
+			eventCount++
+		case <-time.After(pollFrequency * 5):
+			if eventCount == len(expected) {
+				return
+			}
+			t.Fatalf("timed out waiting for events: %v", visited)
+		}
+	}
+}
+
+func keyForEvent(event WatchEvent) string {
+	switch event := event.(type) {
+	case WatchEventModuleAdded:
+		return "added:" + event.Module.Config.Module
+	case WatchEventModuleRemoved:
+		return "removed:" + event.Module.Config.Module
+	case WatchEventModuleChanged:
+		return "updated:" + event.Module.Config.Module
+	default:
+		panic("unknown event type")
+	}
+}
+
+func updateModFile(module string) in.Action {
+	return in.EditFile(module, func(b []byte) []byte {
+		return []byte(strings.Replace(string(b), "github.com/TBD54566975/ftl", "../..", 1))
+	}, "go.mod")
+}