Method TestJobWatcher

database/watcher/watcher_store_test.go:38–152

()

}

func (s *WatcherStoreTestSuite) TestJobWatcher() {
	consumer, err := watcher.RegisterConsumer(
		s.ctx, "job-test",
		watcher.WithEntityTypeFilter(common.JobEntityType),
		watcher.WithAny(
			watcher.WithOperationTypeFilter(common.CreateOperation),
			watcher.WithOperationTypeFilter(common.UpdateOperation),
			watcher.WithOperationTypeFilter(common.DeleteOperation)),
	)
	s.Require().NoError(err)
	s.Require().NotNil(consumer)
	s.T().Cleanup(func() { consumer.Close() })
	consumeEvents(consumer)

	jobParams := params.Job{
		WorkflowJobID: 2,
		RunID:         2,
		Action:        "test-action",
		Conclusion:    "started",
		Status:        "in_progress",
		Name:          "test-job",
	}

	job, err := s.store.CreateOrUpdateJob(s.ctx, jobParams)
	s.Require().NoError(err)

	select {
	case event := <-consumer.Watch():
		s.Require().Equal(common.ChangePayload{
			EntityType: common.JobEntityType,
			Operation:  common.CreateOperation,
			Payload:    job,
		}, event)
		asJob, ok := event.Payload.(params.Job)
		s.Require().True(ok)
		s.Require().Equal(job.ID, int64(1))
		s.Require().Equal(asJob.ID, int64(1))
	case <-time.After(1 * time.Second):
		s.T().Fatal("expected payload not received")
	}

	job.Conclusion = "success"
	updatedJob, err := s.store.CreateOrUpdateJob(s.ctx, job)
	s.Require().NoError(err)

	select {
	case event := <-consumer.Watch():
		s.Require().Equal(common.ChangePayload{
			EntityType: common.JobEntityType,
			Operation:  common.UpdateOperation,
			Payload:    updatedJob,
		}, event)
	case <-time.After(1 * time.Second):
		s.T().Fatal("expected payload not received")
	}

	entityID, err := uuid.NewUUID()
	s.Require().NoError(err)