chore: use builtin run watcher for integration tests

internal/integration/daemon_helpers_test.go

@@ -207,19 +207,20 @@ func (s *testDaemon) getRun(t *testing.T, ctx context.Context, runID resource.ID
 	return run
 }
 
-func (s *testDaemon) waitRunStatus(t *testing.T, runID resource.ID, status run.Status) {
+func (s *testDaemon) waitRunStatus(t *testing.T, runID resource.ID, status run.Status) *run.Run {
 	t.Helper()
 
 	for event := range s.runEvents {
 		if event.Payload.ID == runID {
 			if event.Payload.Status == status {
-				break
+				return event.Payload
 			}
 			if event.Payload.Done() && event.Payload.Status != status {
 				t.Fatalf("expected run status %s but run finished with status %s", status, event.Payload.Status)
 			}
 		}
 	}
+	return nil
 }
 
 func (s *testDaemon) createVCSProvider(t *testing.T, ctx context.Context, org *organization.Organization) *vcsprovider.VCSProvider {

internal/integration/github_app_test.go

@@ -250,13 +250,10 @@ func TestIntegration_GithubApp_Event(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	sub, unsub := daemon.Runs.Watch(ctx)
-	defer unsub()
-
 	// send event
 	push := testutils.ReadFile(t, "./fixtures/github_app_push.json")
 	github.SendEventRequest(t, github.PushEvent, daemon.System.URL(github.AppEventsPath), "secret", push)
 
 	// wait for run to be created
-	<-sub
+	<-daemon.runEvents
 }

internal/integration/remote_state_sharing_test.go

@@ -39,24 +39,19 @@ func TestRemoteStateSharing(t *testing.T) {
 	producerCV := daemon.createConfigurationVersion(t, ctx, producer, nil)
 	err = daemon.Configs.UploadConfig(ctx, producerCV.ID, tarball)
 	require.NoError(t, err)
-	// listen to run events, and create run and apply
-	sub, unsub := daemon.Runs.Watch(ctx)
-	defer unsub()
-	_ = daemon.createRun(t, ctx, producer, producerCV, nil)
-applied:
-	for event := range sub {
-		switch event.Payload.Status {
-		case run.RunPlanned:
-			err := daemon.Runs.Apply(ctx, event.Payload.ID)
-			require.NoError(t, err)
-		case run.RunApplied:
-			break applied
-		case run.RunErrored:
-			t.Fatalf("run unexpectedly errored")
-		}
-	}
 
-	// consume state from a run in the consumer workspace
+	producerRun := daemon.createRun(t, ctx, producer, producerCV, nil)
+
+	// Wait for run to reach planned state before applying
+	planned := daemon.waitRunStatus(t, producerRun.ID, run.RunPlanned)
+	err = daemon.Runs.Apply(ctx, planned.ID)
+	require.NoError(t, err)
+
+	// Wait for run to be applied
+	daemon.waitRunStatus(t, producerRun.ID, run.RunApplied)
+
+	// consume state in a run in the consumer workspace from the producer
+	// workspace.
 	consumerRoot := t.TempDir()
 	consumerConfig := fmt.Sprintf(`
 data "terraform_remote_state" "producer" {
@@ -84,21 +79,14 @@ output "remote_foo" {
 	require.NoError(t, err)
 
 	// create run and apply
-	_ = daemon.createRun(t, ctx, consumer, consumerCV, nil)
-	for event := range sub {
-		switch event.Payload.Status {
-		case run.RunPlanned:
-			err := daemon.Runs.Apply(ctx, event.Payload.ID)
-			require.NoError(t, err)
-		case run.RunApplied:
-			return
-		case run.RunErrored:
-			t.Fatalf("run unexpectedly errored")
-		}
-	}
+	consumerRun := daemon.createRun(t, ctx, consumer, consumerCV, nil)
+	planned = daemon.waitRunStatus(t, consumerRun.ID, run.RunPlanned)
+	err = daemon.Runs.Apply(ctx, planned.ID)
+	require.NoError(t, err)
+	daemon.waitRunStatus(t, consumerRun.ID, run.RunApplied)
 
 	got := daemon.getCurrentState(t, ctx, consumer.ID)
-	if assert.Contains(t, got.Outputs, "foo") {
-		assert.Equal(t, "bar", got.Outputs["foo"])
+	if assert.Contains(t, got.Outputs, "remote_foo", got.Outputs) {
+		assert.Equal(t, `"bar"`, string(got.Outputs["remote_foo"].Value))
 	}
 }

internal/integration/retry_run_ui_test.go

@@ -19,19 +19,9 @@ func TestIntegration_RetryRunUI(t *testing.T) {
 	cv := daemon.createAndUploadConfigurationVersion(t, ctx, ws, &configversion.CreateOptions{
 		Speculative: internal.Bool(true),
 	})
-	// watch run events
-	sub, unsub := daemon.Runs.Watch(ctx)
-	defer unsub()
 	// create a run and wait for it reach planned-and-finished state
 	r := daemon.createRun(t, ctx, ws, cv, nil)
-	for event := range sub {
-		if event.Payload.Status == run.RunErrored {
-			t.Fatal("run unexpectedly errored")
-		}
-		if event.Payload.Status == run.RunPlannedAndFinished {
-			break
-		}
-	}
+	daemon.waitRunStatus(t, r.ID, run.RunPlannedAndFinished)
 
 	// open browser, go to run, and click retry
 	browser.New(t, ctx, func(page playwright.Page) {

internal/integration/run_api_test.go

@@ -49,9 +49,6 @@ func TestIntegration_RunAPI(t *testing.T) {
 		})
 		require.NoError(t, err)
 
-		sub, unsub := daemon.Runs.Watch(ctx)
-		defer unsub()
-
 		created, err := tfeClient.Runs.Create(ctx, tfe.RunCreateOptions{
 			// no config version ID specified
 			Workspace: &tfe.Workspace{
@@ -61,19 +58,10 @@ func TestIntegration_RunAPI(t *testing.T) {
 		require.NoError(t, err)
 
 		// wait for run to reach planned status
-		for event := range sub {
-			switch event.Payload.Status {
-			case run.RunErrored:
-				t.Fatal("run unexpectedly errored")
-			case run.RunPlanned:
-				// run should have planned two resources (defined in the config from the
-				// github repo)
-				planned, err := daemon.Runs.Get(ctx, testutils.ParseID(t, created.ID))
-				require.NoError(t, err)
+		planned := daemon.waitRunStatus(t, testutils.ParseID(t, created.ID), run.RunPlanned)
 
-				assert.Equal(t, 2, planned.Plan.ResourceReport.Additions)
-				return // success
-			}
-		}
+		// run should have planned two resources (defined in the config from the
+		// github repo)
+		assert.Equal(t, 2, planned.Plan.ResourceReport.Additions)
 	})
 }

internal/integration/run_error_test.go

@@ -77,10 +77,6 @@ func TestRunError(t *testing.T) {
 			err = daemon.Configs.UploadConfig(ctx, cv.ID, tarball)
 			require.NoError(t, err)
 
-			// watch run events
-			runsSub, runsUnsub := daemon.Runs.Watch(ctx)
-			defer runsUnsub()
-
 			// watch log events
 			logsSub, logsUnsub := daemon.Logs.WatchLogs(ctx)
 			defer logsUnsub()
@@ -102,7 +98,7 @@ func TestRunError(t *testing.T) {
 					if errorRegex.Match(event.Payload.Data) {
 						gotErrorLogs = true
 					}
-				case event := <-runsSub:
+				case event := <-daemon.runEvents:
 					if event.Payload.Status == run.RunErrored {
 						gotErrorStatus = true
 					}

internal/integration/run_status_test.go

@@ -27,10 +27,6 @@ func TestIntegration_RunStatus(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	// watch run events
-	sub, unsub := daemon.Runs.Watch(ctx)
-	defer unsub()
-
 	// directory for root module
 	root := t.TempDir()
 
@@ -95,16 +91,10 @@ output "cat_name" { value = random_pet.cat.id }
 
 			// create run and wait for it to reach wanted status
 			created := daemon.createRun(t, ctx, ws, cv, nil)
-			for event := range sub {
-				r := event.Payload
-				if r.ID == created.ID && r.Status == step.wantStatus {
-					// status matches, now check whether reports match as well
-					assert.Equal(t, &step.wantResourceReport, event.Payload.Plan.ResourceReport)
-					assert.Equal(t, &step.wantOutputReport, r.Plan.OutputReport)
-					break
-				}
-				require.False(t, r.Done(), "run unexpectedly finished with status %s", r.Status)
-			}
+			updated := daemon.waitRunStatus(t, created.ID, step.wantStatus)
+			// status matches, now check whether reports match as well
+			assert.Equal(t, &step.wantResourceReport, updated.Plan.ResourceReport)
+			assert.Equal(t, &step.wantOutputReport, updated.Plan.OutputReport)
 		})
 	}
 }

internal/integration/state_ui_test.go

@@ -15,28 +15,15 @@ func TestIntegration_StateUI(t *testing.T) {
 	integrationTest(t)
 
 	daemon, org, ctx := setup(t, nil)
-
-	// watch run events
-	sub, unsub := daemon.Runs.Watch(ctx)
-	defer unsub()
-
-	// create run and wait for it to complete
 	ws := daemon.createWorkspace(t, ctx, org)
 	cv := daemon.createAndUploadConfigurationVersion(t, ctx, ws, nil)
-	_ = daemon.createRun(t, ctx, ws, cv, nil)
-applied:
-	for event := range sub {
-		r := event.Payload
-		switch r.Status {
-		case run.RunApplied:
-			break applied
-		case run.RunPlanned:
-			err := daemon.Runs.Apply(ctx, r.ID)
-			require.NoError(t, err)
-		case run.RunErrored:
-			t.Fatal("run unexpectedly finished with an error")
-		}
-	}
+
+	// create run and wait for it to complete
+	r := daemon.createRun(t, ctx, ws, cv, nil)
+	planned := daemon.waitRunStatus(t, r.ID, run.RunPlanned)
+	err := daemon.Runs.Apply(ctx, planned.ID)
+	require.NoError(t, err)
+	daemon.waitRunStatus(t, r.ID, run.RunApplied)
 
 	browser.New(t, ctx, func(page playwright.Page) {
 		_, err := page.Goto(workspaceURL(daemon.System.Hostname(), org.Name, ws.Name))

internal/integration/terraform_cli_cancel_test.go

@@ -23,10 +23,6 @@ func TestIntegration_TerraformCLICancel(t *testing.T) {
 
 	svc, org, ctx := setup(t, nil)
 
-	// watch run events
-	runsSub, runsUnsub := svc.Runs.Watch(ctx)
-	defer runsUnsub()
-
 	// Canceling a run is not straight-forward, because to do so reliably the
 	// terraform apply should be interrupted precisely when it is in mid-flow,
 	// i.e. while it is planning. To achieve this, the test uses the 'http'
@@ -81,7 +77,7 @@ data "http" "wait" {
 	require.NoError(t, <-tferr, string(testutils.ReadFile(t, out.Name())))
 	t.Log(string(testutils.ReadFile(t, out.Name())))
 
-	for event := range runsSub {
+	for event := range svc.runEvents {
 		r := event.Payload
 		if r.Status == run.RunCanceled {
 			break

internal/integration/timeout_test.go

@@ -28,10 +28,6 @@ func TestIntegration_Timeout(t *testing.T) {
 	})
 	ws := svc.createWorkspace(t, ctx, org)
 
-	// watch run events
-	runsSub, runsUnsub := svc.Runs.Watch(ctx)
-	defer runsUnsub()
-
 	// Setup a http server, to which the terraform 'http' data source will
 	// connect, causing it to hang, thereby keeping OTF run in the planning
 	// state.
@@ -56,17 +52,10 @@ data "http" "wait" {
 	err = svc.Configs.UploadConfig(ctx, cv.ID, tarball)
 	require.NoError(t, err)
 
-	// create run and wait for it to finish
+	// create run and wait for it to enter canceled state
 	run := svc.createRun(t, ctx, ws, cv, nil)
-	for event := range runsSub {
-		if event.Payload.ID != run.ID {
-			continue
-		}
-		if event.Payload.Done() {
-			run = event.Payload
-			break
-		}
-	}
+	run = svc.waitRunStatus(t, run.ID, otfrun.RunCanceled)
+
 	// run should have reached planning state before being timed out and being
 	// forced into a canceled state.
 	_, err = run.StatusTimestamp(otfrun.RunPlanning)

internal/integration/workspace_api_test.go

@@ -11,7 +11,7 @@ import (
 	tfe "github.com/hashicorp/go-tfe"
 	"github.com/leg100/otf/internal"
 	"github.com/leg100/otf/internal/github"
-	"github.com/leg100/otf/internal/run"
+	otfrun "github.com/leg100/otf/internal/run"
 	"github.com/leg100/otf/internal/testutils"
 	"github.com/leg100/otf/internal/tfeapi/types"
 	"github.com/leg100/otf/internal/vcs"
@@ -98,20 +98,10 @@ func TestIntegration_WorkspaceAPI_CreateConnected(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	// watch run events
-	runsSub, runsUnsub := daemon.Runs.Watch(ctx)
-	defer runsUnsub()
-
-	_, err = daemon.Runs.Create(ctx, testutils.ParseID(t, ws.ID), run.CreateOptions{})
+	run, err := daemon.Runs.Create(ctx, testutils.ParseID(t, ws.ID), otfrun.CreateOptions{})
 	require.NoError(t, err)
 
-	for event := range runsSub {
-		r := event.Payload
-		if r.Status == run.RunPlanned {
-			// status matches, now check whether reports match as well
-			assert.Equal(t, &run.Report{Additions: 2}, r.Plan.ResourceReport)
-			break
-		}
-		require.False(t, r.Done(), "run unexpectedly finished with status %s", r.Status)
-	}
+	run = daemon.waitRunStatus(t, run.ID, otfrun.RunPlanned)
+	// status matches, now check whether reports match as well
+	assert.Equal(t, &otfrun.Report{Additions: 2}, run.Plan.ResourceReport)
 }

internal/notifications/notifier.go

@@ -67,7 +67,7 @@ func NewNotifier(opts NotifierOptions) *Notifier {
 
 // Start the notifier daemon. Should be started in a go-routine.
 func (s *Notifier) Start(ctx context.Context) error {
-	// subscribe to both run events and notification config events
+	// subscribe to notification config events
 	subRuns, unsubRuns := s.runs.Watch(ctx)
 	defer unsubRuns()
 	subConfigs, unsubConfigs := s.notifications.Watch(ctx)