(callStop bool)
| 50 | } |
| 51 | |
| 52 | func testIntegrationS3(callStop bool) func(t *testing.T) { |
| 53 | return func(t *testing.T) { |
| 54 | defer testutil.DisableLogging()() |
| 55 | |
| 56 | toml := ` |
| 57 | [fields] |
| 58 | names=["f0", "f1", "f2", "f3"] |
| 59 | |
| 60 | [input] |
| 61 | name="records" |
| 62 | |
| 63 | [output] |
| 64 | name="custom" |
| 65 | procs=1 |
| 66 | |
| 67 | [upload] |
| 68 | name="s3" |
| 69 | |
| 70 | [upload.config] |
| 71 | sourcebasepath=%q |
| 72 | stagingpath=%q |
| 73 | bucket="my-bucket" |
| 74 | interval="10ms" |
| 75 | retries=1 |
| 76 | concurrency=1` |
| 77 | |
| 78 | /* Configure the pipeline */ |
| 79 | |
| 80 | comp := baker.Components{ |
| 81 | Inputs: []baker.InputDesc{inputtest.RecordsDesc}, |
| 82 | Outputs: []baker.OutputDesc{customOutputDesc}, |
| 83 | Uploads: []baker.UploadDesc{S3Desc}, |
| 84 | } |
| 85 | |
| 86 | basePath, stagingPath := t.TempDir(), t.TempDir() |
| 87 | r := strings.NewReader(fmt.Sprintf(toml, basePath, stagingPath)) |
| 88 | cfg, err := baker.NewConfigFromToml(r, comp) |
| 89 | if err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | |
| 93 | topo, err := baker.NewTopologyFromConfig(cfg) |
| 94 | if err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | |
| 98 | // Upload: mock AWS S3 |
| 99 | u := topo.Upload.(*S3) |
| 100 | s, ops, params := mockS3Service(false) |
| 101 | u.uploader = s3manager.NewUploaderWithClient(s) |
| 102 | |
| 103 | // Output: provide the path in which files should be written |
| 104 | o := topo.Output[0].(*customOutput) |
| 105 | o.path = basePath |
| 106 | |
| 107 | const nfiles = 100 |
| 108 | |
| 109 | // Input: create some dummy records |
no test coverage detected