DanKotowski commented on code in PR #21840: URL: https://github.com/apache/beam/pull/21840#discussion_r897278363
########## sdks/go/pkg/beam/io/fhirio/utils_test.go: ########## @@ -0,0 +1,105 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fhirio + +import ( + "errors" + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "net/http" + "testing" +) + +var ( + fakeRequestReturnErrorMessage = "internal error" + requestReturnErrorFakeClient = &fakeFhirStoreClient{ + fakeReadResources: func(resource string) (*http.Response, error) { + return nil, errors.New(fakeRequestReturnErrorMessage) + }, + fakeExecuteBundles: func(storePath string, bundle []byte) (*http.Response, error) { + return nil, errors.New(fakeRequestReturnErrorMessage) + }, + } + + fakeBadStatus = "403 Forbidden" + badStatusFakeResponse = &http.Response{Status: fakeBadStatus} + badStatusFakeClient = &fakeFhirStoreClient{ + fakeReadResources: func(resource string) (*http.Response, error) { + return badStatusFakeResponse, nil + }, + fakeExecuteBundles: func(storePath string, bundle []byte) (*http.Response, error) { + return badStatusFakeResponse, nil + }, + } + + fakeBodyReaderErrorMessage = "ReadAll fail" + bodyReaderErrorFakeResponse = &http.Response{ + Body: &fakeReaderCloser{ + fakeRead: func([]byte) (int, error) { + return 0, errors.New(fakeBodyReaderErrorMessage) + }, + }, Status: "200 Ok"} + bodyReaderErrorFakeClient = &fakeFhirStoreClient{ + fakeReadResources: func(resource string) (*http.Response, error) { + return bodyReaderErrorFakeResponse, nil + }, + fakeExecuteBundles: func(storePath string, bundle []byte) (*http.Response, error) { + return bodyReaderErrorFakeResponse, nil + }, + } +) + +type fakeFhirStoreClient struct { + fakeReadResources func(string) (*http.Response, error) + fakeExecuteBundles func(storePath string, bundle []byte) (*http.Response, error) +} + +func (c *fakeFhirStoreClient) executeBundle(storePath string, bundle []byte) (*http.Response, error) { + return c.fakeExecuteBundles(storePath, bundle) +} + +func (c *fakeFhirStoreClient) readResource(resourcePath string) (*http.Response, error) { + return c.fakeReadResources(resourcePath) +} + +// Useful to fake the Body of a http.Response. +type fakeReaderCloser struct { + fakeRead func([]byte) (int, error) +} + +func (*fakeReaderCloser) Close() error { + return nil +} + +func (m *fakeReaderCloser) Read(b []byte) (int, error) { + return m.fakeRead(b) +} + +func validateResourceErrorCounter(t *testing.T, pipelineResult beam.PipelineResult, expectedCount int) { Review Comment: This method should not performa the t.Fatalf but return an error back to the testing function that can then check for error and display appropriate message within the tests context. Additionally if you were to keep this helper util you would likely want to call t.Helper(). ########## sdks/go/pkg/beam/io/fhirio/execute_bundles.go: ########## @@ -0,0 +1,146 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package fhirio provides an API for reading and writing resources to Google +// Cloud Healthcare Fhir stores. +// Experimental. +package fhirio + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "strings" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +const ( + bundleResponseTypeBatch = "batch-response" + bundleResponseTypeTransaction = "transaction-response" +) + +func init() { + register.DoFn4x0[context.Context, []byte, func(string), func(string)]((*executeBundlesFn)(nil)) + register.Emitter1[string]() +} + +type executeBundlesFn struct { Review Comment: nit: Should this be executeBundlesFn or just executeBundleFn, I think each item in the PCollection is an individual FHIR bundle? ########## sdks/go/test/integration/io/fhirio/fhirio_test.go: ########## @@ -192,12 +204,29 @@ func TestFhirIO_InvalidRead(t *testing.T) { passert.Count(s, failedReads, "", 1) passert.Empty(s, resources) passert.True(s, failedReads, func(errorMsg string) bool { - return strings.Contains(errorMsg, "bad status [404]") + return strings.Contains(errorMsg, "404") Review Comment: Could we use the net/http constants here instead of hard coding "404" ########## sdks/go/test/integration/io/fhirio/fhirio_test.go: ########## @@ -192,12 +204,29 @@ func TestFhirIO_InvalidRead(t *testing.T) { passert.Count(s, failedReads, "", 1) passert.Empty(s, resources) passert.True(s, failedReads, func(errorMsg string) bool { - return strings.Contains(errorMsg, "bad status [404]") + return strings.Contains(errorMsg, "404") }) ptest.RunAndValidate(t, p) } +func TestFhirIO_ExecuteBundles(t *testing.T) { + integration.CheckFilters(t) + checkFlags(t) + + fhirStorePath, teardownFhirStore := setupEmptyFhirStore(t) + defer teardownFhirStore() + + p, s, bundles := ptest.CreateList(readPrettyBundles()) + successBodies, failures := fhirio.ExecuteBundles(s, fhirStorePath, bundles) + passert.Count(s, successBodies, "", 2) + passert.Count(s, failures, "", 2) + passert.True(s, failures, func(errorMsg string) bool { + return strings.Contains(errorMsg, "400") Review Comment: Same comment as above. ########## sdks/go/pkg/beam/io/fhirio/execute_bundles_test.go: ########## @@ -0,0 +1,79 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fhirio + +import ( + "bytes" + "net/http" + "strings" + "testing" + + "github.com/apache/beam/sdks/v2/go/pkg/beam/testing/passert" + "github.com/apache/beam/sdks/v2/go/pkg/beam/testing/ptest" +) + +func TestExecuteBundles(t *testing.T) { Review Comment: Could we add another test cases for ExecuteBundles that is successful ########## sdks/go/test/integration/io/fhirio/fhirio_test.go: ########## @@ -75,13 +84,16 @@ func setupFhirStore(t *testing.T) (string, []string, func()) { healthcareDataset := fmt.Sprintf(datasetPathFmt, *gcpopts.Project, *gcpopts.Region) Review Comment: Is this the standard way to define a dataset for Apache beam? Would we ever want to potentially have different datasets for different tests? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
