Function testIntegrationSQS

input/sqs_test.go:355–437

(tc sqsIntegrationTestCase)

}

func testIntegrationSQS(tc sqsIntegrationTestCase) func(t *testing.T) {
	return func(t *testing.T) {
		toml := `
[fields]
Names=["bucket", "path", "filename"]

[csv]
field_separator=","

[input]
Name="sqs"
[input.config]
Bucket=%q
MessageFormat="plain"
QueuePrefixes=[%v]
QueueNames=[%v]
FilePathFilter=".*"

[output]
Name="RawRecorder"
Procs=1
fields=["bucket", "path", "filename"]
`

		/* Configure the pipeline */
		comp := baker.Components{
			Inputs:  []baker.InputDesc{SQSDesc},
			Outputs: []baker.OutputDesc{outputtest.RawRecorderDesc},
		}

		prefixes := []string{}
		for _, pref := range tc.queuePrefixes {
			prefixes = append(prefixes, `"`+pref+`"`)
		}
		queues := []string{}
		for _, name := range tc.queueNames {
			queues = append(queues, `"`+name+`"`)
		}

		r := strings.NewReader(fmt.Sprintf(toml, tc.bucket, strings.Join(prefixes, ","), strings.Join(queues, ",")))
		cfg, err := baker.NewConfigFromToml(r, comp)
		if err != nil {
			t.Fatal(err)
		}

		topo, err := baker.NewTopologyFromConfig(cfg)
		if err != nil {
			t.Fatal(err)
		}

		// Replace aws services interfaces with mocks.
		topo.Input.(*SQS).svc = &mockSQSClient{
			queues: tc.messages,
		}
		topo.Input.(*SQS).s3Input.SetS3API(newMockedS3FromFS(os.DirFS("testdata/sqstest")))

		/* Run the pipeline */
		topo.Start()