This is an automated email from the ASF dual-hosted git repository.
danoliveira pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b9d99405aba Add ExecuteBundles transform to Go FhirIO (#21840)
b9d99405aba is described below
commit b9d99405abaf74889eb9cf05eec4d67ea275a6b4
Author: Lucas Nogueira <lnogu...@uwaterloo.ca>
AuthorDate: Tue Jun 21 20:30:22 2022 -0400
Add ExecuteBundles transform to Go FhirIO (#21840)
* squashed rebase conflict commits
* add execute bundles transform with integration test
* adjust import to follow convention
* fix variable scope
* adjust read unit tests
* fix bad status test case
* improve read unit test assertions and naming
* add unit tests
* add comment to execute bundles transform
* include error reason
* make variable exported to fix integration test
* adjust integration tests after merge
* update license comment
* unify unit test utilities in single file
* add comment explaining the purpose of unexported transform function
* remove unnecessary generic usage
* remove coded mistakenly added
* improve transaction vs batch bundle comment
* use net/http constants instead of hardcoded values
* return error instead of t.Fail inside helper function
* improve execute bundles DoFn identifier
* adjust import spacing
* improve error message
---
sdks/go/pkg/beam/io/fhirio/common.go | 56 ++++++++
sdks/go/pkg/beam/io/fhirio/execute_bundles.go | 146 +++++++++++++++++++++
sdks/go/pkg/beam/io/fhirio/execute_bundles_test.go | 82 ++++++++++++
sdks/go/pkg/beam/io/fhirio/fakes_test.go | 39 ------
sdks/go/pkg/beam/io/fhirio/read.go | 45 +++----
sdks/go/pkg/beam/io/fhirio/read_test.go | 55 ++------
sdks/go/pkg/beam/io/fhirio/utils_test.go | 107 +++++++++++++++
sdks/go/test/integration/io/fhirio/fhirio_test.go | 46 +++++--
8 files changed, 456 insertions(+), 120 deletions(-)
diff --git a/sdks/go/pkg/beam/io/fhirio/common.go
b/sdks/go/pkg/beam/io/fhirio/common.go
index 98bc10e3306..666f9d6f0e0 100644
--- a/sdks/go/pkg/beam/io/fhirio/common.go
+++ b/sdks/go/pkg/beam/io/fhirio/common.go
@@ -19,10 +19,16 @@
package fhirio
import (
+ "bytes"
"context"
+ "io"
"net/http"
+ "regexp"
+ "time"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
"google.golang.org/api/healthcare/v1"
"google.golang.org/api/option"
)
@@ -32,8 +38,54 @@ const (
baseMetricPrefix = "fhirio/"
)
+func executeRequestAndRecordLatency(ctx context.Context, latencyMs
*beam.Distribution, requestSupplier func() (*http.Response, error))
(*http.Response, error) {
+ timeBeforeReadRequest := time.Now()
+ response, err := requestSupplier()
+ latencyMs.Update(ctx, time.Since(timeBeforeReadRequest).Milliseconds())
+ return response, err
+}
+
+func extractBodyFrom(response *http.Response) (string, error) {
+ if isBadStatusCode(response.Status) {
+ return "", errors.Errorf("response contains bad status: [%v]",
response.Status)
+ }
+
+ bodyBytes, err := io.ReadAll(response.Body)
+ if err != nil {
+ return "", err
+ }
+
+ return string(bodyBytes), nil
+}
+
+func isBadStatusCode(status string) bool {
+ // 2XXs are successes, otherwise failure.
+ isMatch, err := regexp.MatchString("^2\\d{2}", status)
+ if err != nil {
+ return true
+ }
+ return !isMatch
+}
+
+type fhirioFnCommon struct {
+ client fhirStoreClient
+ resourcesErrorCount beam.Counter
+ resourcesSuccessCount beam.Counter
+ latencyMs beam.Distribution
+}
+
+func (fnc *fhirioFnCommon) setup(namespace string) {
+ if fnc.client == nil {
+ fnc.client = newFhirStoreClient()
+ }
+ fnc.resourcesErrorCount = beam.NewCounter(namespace,
baseMetricPrefix+"resource_error_count")
+ fnc.resourcesSuccessCount = beam.NewCounter(namespace,
baseMetricPrefix+"resource_success_count")
+ fnc.latencyMs = beam.NewDistribution(namespace,
baseMetricPrefix+"latency_ms")
+}
+
type fhirStoreClient interface {
readResource(resourcePath string) (*http.Response, error)
+ executeBundle(storePath string, bundle []byte) (*http.Response, error)
}
type fhirStoreClientImpl struct {
@@ -51,3 +103,7 @@ func newFhirStoreClient() *fhirStoreClientImpl {
func (c *fhirStoreClientImpl) readResource(resourcePath string)
(*http.Response, error) {
return c.fhirService.Read(resourcePath).Do()
}
+
+func (c *fhirStoreClientImpl) executeBundle(storePath string, bundle []byte)
(*http.Response, error) {
+ return c.fhirService.ExecuteBundle(storePath,
bytes.NewReader(bundle)).Do()
+}
diff --git a/sdks/go/pkg/beam/io/fhirio/execute_bundles.go
b/sdks/go/pkg/beam/io/fhirio/execute_bundles.go
new file mode 100644
index 00000000000..c96d72fae39
--- /dev/null
+++ b/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)]((*executeBundleFn)(nil))
+ register.Emitter1[string]()
+}
+
+type executeBundleFn struct {
+ fhirioFnCommon
+ successesCount beam.Counter
+ // Path to FHIR store where bundle requests will be executed on.
+ FhirStorePath string
+}
+
+func (fn executeBundleFn) String() string {
+ return "executeBundleFn"
+}
+
+func (fn *executeBundleFn) Setup() {
+ fn.fhirioFnCommon.setup(fn.String())
+ fn.successesCount = beam.NewCounter(fn.String(),
baseMetricPrefix+"success_count")
+}
+
+func (fn *executeBundleFn) ProcessElement(ctx context.Context, inputBundleBody
[]byte, emitSuccess, emitFailure func(string)) {
+ response, err := executeRequestAndRecordLatency(ctx, &fn.latencyMs,
func() (*http.Response, error) {
+ return fn.client.executeBundle(fn.FhirStorePath,
inputBundleBody)
+ })
+ if err != nil {
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitFailure(errors.Wrap(err, "execute bundle request returned
error").Error())
+ return
+ }
+
+ body, err := extractBodyFrom(response)
+ if err != nil {
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitFailure(errors.Wrap(err, "could not extract body from
execute bundles response").Error())
+ return
+ }
+
+ fn.processResponseBody(ctx, body, emitSuccess, emitFailure)
+}
+
+func (fn *executeBundleFn) processResponseBody(ctx context.Context, body
string, emitSuccess, emitFailure func(string)) {
+ var bodyFields struct {
+ Type string `json:"type"`
+ Entries []interface{} `json:"entry"`
+ }
+
+ err := json.NewDecoder(strings.NewReader(body)).Decode(&bodyFields)
+ if err != nil {
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitFailure(errors.Wrap(err, "could not parse body from execute
bundle response").Error())
+ return
+ }
+
+ if bodyFields.Entries == nil {
+ return
+ }
+
+ // A BATCH bundle returns a success response even if entries have
failures, as
+ // entries are executed separately. However, TRANSACTION bundles should
return
+ // error response (in client.executeBundle call) if any entry fails.
Therefore,
+ // for BATCH bundles we need to parse the error and success counters.
+ switch bodyFields.Type {
+ case bundleResponseTypeTransaction:
+ fn.resourcesSuccessCount.Inc(ctx,
int64(len(bodyFields.Entries)))
+ emitSuccess(body)
+ case bundleResponseTypeBatch:
+ for _, entry := range bodyFields.Entries {
+ var entryFields struct {
+ Response struct {
+ Status string `json:"status"`
+ } `json:"response"`
+ }
+ entryBytes, _ := json.Marshal(entry)
+ _ =
json.NewDecoder(bytes.NewReader(entryBytes)).Decode(&entryFields)
+ if entryFields.Response.Status == "" {
+ continue
+ }
+
+ if isBadStatusCode(entryFields.Response.Status) {
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitFailure(errors.Errorf("execute bundles
entry contains bad status: [%v]", entryFields.Response.Status).Error())
+ } else {
+ fn.resourcesSuccessCount.Inc(ctx, 1)
+ emitSuccess(string(entryBytes))
+ }
+ }
+ }
+
+ fn.successesCount.Inc(ctx, 1)
+}
+
+// ExecuteBundles performs all the requests in the specified bundles on a given
+// FHIR store. This transform takes a path to a FHIR store and a PCollection of
+// bundles as JSON-encoded strings. It executes the requests defined on the
+// bundles on the FHIR store located on the provided path. It outputs two
+// PCollection<string>, the first containing the response bodies of the
+// successfully performed requests and the second one error messages of the
+// requests that failed to be executed.
+// See:
https://cloud.google.com/healthcare-api/docs/samples/healthcare-fhir-execute-bundle
+func ExecuteBundles(s beam.Scope, fhirStorePath string, bundles
beam.PCollection) (beam.PCollection, beam.PCollection) {
+ s = s.Scope("fhirio.ExecuteBundles")
+ return executeBundles(s, fhirStorePath, bundles, nil)
+}
+
+// This is useful as an entry point for testing because we can provide a fake
FHIR store client.
+func executeBundles(s beam.Scope, fhirStorePath string, bundles
beam.PCollection, client fhirStoreClient) (beam.PCollection, beam.PCollection) {
+ return beam.ParDo2(s, &executeBundleFn{fhirioFnCommon:
fhirioFnCommon{client: client}, FhirStorePath: fhirStorePath}, bundles)
+}
diff --git a/sdks/go/pkg/beam/io/fhirio/execute_bundles_test.go
b/sdks/go/pkg/beam/io/fhirio/execute_bundles_test.go
new file mode 100644
index 00000000000..c0eb1174bb1
--- /dev/null
+++ b/sdks/go/pkg/beam/io/fhirio/execute_bundles_test.go
@@ -0,0 +1,82 @@
+// 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) {
+ testCases := []struct {
+ name string
+ client fhirStoreClient
+ containedError string
+ }{
+ {
+ name: "Execute Bundles request returns error",
+ client: requestReturnErrorFakeClient,
+ containedError: fakeRequestReturnErrorMessage,
+ },
+ {
+ name: "Execute Bundles request returns bad
status",
+ client: badStatusFakeClient,
+ containedError: fakeBadStatus,
+ },
+ {
+ name: "Execute Bundles request response body
fails to be read",
+ client: bodyReaderErrorFakeClient,
+ containedError: fakeBodyReaderErrorMessage,
+ },
+ {
+ name: "Execute Bundles request response body failed to
be decoded",
+ client: &fakeFhirStoreClient{
+ fakeExecuteBundles: func(storePath string,
bundle []byte) (*http.Response, error) {
+ return &http.Response{
+ Body: &fakeReaderCloser{
+ fakeRead: func(t
[]byte) (int, error) {
+ return
bytes.NewReader([]byte("")).Read(t)
+ },
+ }, Status: "200 Ok"}, nil
+ },
+ },
+ containedError: "EOF",
+ },
+ }
+
+ testBundles := [][]byte{[]byte("foo"), []byte("bar")}
+ for _, testCase := range testCases {
+ t.Run(testCase.name, func(t *testing.T) {
+ p, s, bundles := ptest.CreateList(testBundles)
+ successfulBodies, failures := executeBundles(s, "bla",
bundles, testCase.client)
+ passert.Empty(s, successfulBodies)
+ passert.Count(s, failures, "", len(testBundles))
+ passert.True(s, failures, func(errorMsg string) bool {
+ return strings.Contains(errorMsg,
testCase.containedError)
+ })
+ pipelineResult := ptest.RunAndValidate(t, p)
+ err := validateResourceErrorCounter(pipelineResult,
len(testBundles))
+ if err != nil {
+ t.Fatalf("validateResourceErrorCounter returned
error [%v]", err.Error())
+ }
+ })
+ }
+}
diff --git a/sdks/go/pkg/beam/io/fhirio/fakes_test.go
b/sdks/go/pkg/beam/io/fhirio/fakes_test.go
deleted file mode 100644
index 4e1a51aeb23..00000000000
--- a/sdks/go/pkg/beam/io/fhirio/fakes_test.go
+++ /dev/null
@@ -1,39 +0,0 @@
-// 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 "net/http"
-
-type fakeFhirStoreClient struct {
- fakeReadResources func(string) (*http.Response, error)
-}
-
-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)
-}
diff --git a/sdks/go/pkg/beam/io/fhirio/read.go
b/sdks/go/pkg/beam/io/fhirio/read.go
index a710c92a869..41c53a540b6 100644
--- a/sdks/go/pkg/beam/io/fhirio/read.go
+++ b/sdks/go/pkg/beam/io/fhirio/read.go
@@ -20,8 +20,7 @@ package fhirio
import (
"context"
- "io"
- "time"
+ "net/http"
"github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
@@ -34,10 +33,7 @@ func init() {
}
type readResourceFn struct {
- client fhirStoreClient
- readResourceErrors beam.Counter
- readResourceSuccess beam.Counter
- readResourceLatencyMs beam.Distribution
+ fhirioFnCommon
}
func (fn readResourceFn) String() string {
@@ -45,40 +41,28 @@ func (fn readResourceFn) String() string {
}
func (fn *readResourceFn) Setup() {
- if fn.client == nil {
- fn.client = newFhirStoreClient()
- }
- fn.readResourceErrors = beam.NewCounter(fn.String(),
baseMetricPrefix+"read_resource_error_count")
- fn.readResourceSuccess = beam.NewCounter(fn.String(),
baseMetricPrefix+"read_resource_success_count")
- fn.readResourceLatencyMs = beam.NewDistribution(fn.String(),
baseMetricPrefix+"read_resource_latency_ms")
+ fn.fhirioFnCommon.setup(fn.String())
}
func (fn *readResourceFn) ProcessElement(ctx context.Context, resourcePath
string, emitResource, emitDeadLetter func(string)) {
- timeBeforeReadRequest := time.Now()
- response, err := fn.client.readResource(resourcePath)
- fn.readResourceLatencyMs.Update(ctx,
time.Since(timeBeforeReadRequest).Milliseconds())
-
+ response, err := executeRequestAndRecordLatency(ctx, &fn.latencyMs,
func() (*http.Response, error) {
+ return fn.client.readResource(resourcePath)
+ })
if err != nil {
- fn.readResourceErrors.Inc(ctx, 1)
- emitDeadLetter(errors.Wrapf(err, "failed fetching resource
[%s]", resourcePath).Error())
- return
- }
-
- if response.StatusCode != 200 {
- fn.readResourceErrors.Inc(ctx, 1)
- emitDeadLetter(errors.Errorf("fetched resource [%s] returned
bad status [%d]", resourcePath, response.StatusCode).Error())
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitDeadLetter(errors.Wrapf(err, "read resource request
returned error on input: [%v]", resourcePath).Error())
return
}
- bytes, err := io.ReadAll(response.Body)
+ body, err := extractBodyFrom(response)
if err != nil {
- fn.readResourceErrors.Inc(ctx, 1)
- emitDeadLetter(errors.Wrapf(err, "error reading response body
of resource [%s]", resourcePath).Error())
+ fn.resourcesErrorCount.Inc(ctx, 1)
+ emitDeadLetter(errors.Wrapf(err, "could not extract body from
read resource [%v] response", resourcePath).Error())
return
}
- fn.readResourceSuccess.Inc(ctx, 1)
- emitResource(string(bytes))
+ fn.resourcesSuccessCount.Inc(ctx, 1)
+ emitResource(body)
}
// Read fetches resources from Google Cloud Healthcare FHIR stores based on the
@@ -93,6 +77,7 @@ func Read(s beam.Scope, resourcePaths beam.PCollection)
(beam.PCollection, beam.
return read(s, resourcePaths, nil)
}
+// This is useful as an entry point for testing because we can provide a fake
FHIR store client.
func read(s beam.Scope, resourcePaths beam.PCollection, client
fhirStoreClient) (beam.PCollection, beam.PCollection) {
- return beam.ParDo2(s, &readResourceFn{client: client}, resourcePaths)
+ return beam.ParDo2(s, &readResourceFn{fhirioFnCommon:
fhirioFnCommon{client: client}}, resourcePaths)
}
diff --git a/sdks/go/pkg/beam/io/fhirio/read_test.go
b/sdks/go/pkg/beam/io/fhirio/read_test.go
index 7b8cba1af8d..39ae484b92b 100644
--- a/sdks/go/pkg/beam/io/fhirio/read_test.go
+++ b/sdks/go/pkg/beam/io/fhirio/read_test.go
@@ -16,8 +16,6 @@
package fhirio
import (
- "errors"
- "net/http"
"strings"
"testing"
@@ -32,35 +30,19 @@ func TestRead(t *testing.T) {
containedError string
}{
{
- name: "Read Request Failed",
- client: &fakeFhirStoreClient{
- fakeReadResources: func(resource string)
(*http.Response, error) {
- return nil, errors.New("")
- },
- },
- containedError: "failed fetching resource",
+ name: "Read request returns error",
+ client: requestReturnErrorFakeClient,
+ containedError: fakeRequestReturnErrorMessage,
},
{
- name: "Read Request Returns Bad Status",
- client: &fakeFhirStoreClient{
- fakeReadResources: func(resource string)
(*http.Response, error) {
- return &http.Response{StatusCode: 403},
nil
- },
- },
- containedError: "returned bad status",
+ name: "Read request returns bad status",
+ client: badStatusFakeClient,
+ containedError: fakeBadStatus,
},
{
- name: "Response body fails to be parsed",
- client: &fakeFhirStoreClient{
- fakeReadResources: func(resource string)
(*http.Response, error) {
- return &http.Response{Body:
&fakeReaderCloser{
- fakeRead: func([]byte) (int,
error) {
- return 0, errors.New("")
- },
- }, StatusCode: 200}, nil
- },
- },
- containedError: "error reading response body",
+ name: "Read request response body fails to be
read",
+ client: bodyReaderErrorFakeClient,
+ containedError: fakeBodyReaderErrorMessage,
},
}
@@ -75,23 +57,10 @@ func TestRead(t *testing.T) {
return strings.Contains(errorMsg,
testCase.containedError)
})
pipelineResult := ptest.RunAndValidate(t, p)
- counterResults :=
pipelineResult.Metrics().AllMetrics().Counters()
-
- if len(counterResults) != 1 {
- t.Fatalf("counterResults got length %v,
expected %v", len(counterResults), 1)
- }
- counterResult := counterResults[0]
-
- expectedCounterName :=
"fhirio/read_resource_error_count"
- if counterResult.Name() != expectedCounterName {
- t.Fatalf("counterResult.Name() is '%v',
expected '%v'", counterResult.Name(), expectedCounterName)
+ err := validateResourceErrorCounter(pipelineResult,
len(testResourcePaths))
+ if err != nil {
+ t.Fatalf("validateResourceErrorCounter returned
error [%v]", err.Error())
}
-
- expectedCounterResult := int64(len(testResourcePaths))
- if counterResult.Result() != expectedCounterResult {
- t.Fatalf("counterResult.Result() is %v,
expected %v", counterResult.Result(), expectedCounterResult)
- }
-
})
}
}
diff --git a/sdks/go/pkg/beam/io/fhirio/utils_test.go
b/sdks/go/pkg/beam/io/fhirio/utils_test.go
new file mode 100644
index 00000000000..a8ffefd207c
--- /dev/null
+++ b/sdks/go/pkg/beam/io/fhirio/utils_test.go
@@ -0,0 +1,107 @@
+// 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"
+ "fmt"
+ "net/http"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+)
+
+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(pipelineResult beam.PipelineResult,
expectedCount int) error {
+ counterResults := pipelineResult.Metrics().AllMetrics().Counters()
+ if len(counterResults) != 1 {
+ return fmt.Errorf("counterResults got length %v, expected %v",
len(counterResults), 1)
+ }
+ counterResult := counterResults[0]
+
+ expectedCounterName := "fhirio/resource_error_count"
+ if counterResult.Name() != expectedCounterName {
+ return fmt.Errorf("counterResult.Name() is '%v', expected
'%v'", counterResult.Name(), expectedCounterName)
+ }
+
+ if counterResult.Result() != int64(expectedCount) {
+ return fmt.Errorf("counterResult.Result() is %v, expected %v",
counterResult.Result(), expectedCount)
+ }
+ return nil
+}
diff --git a/sdks/go/test/integration/io/fhirio/fhirio_test.go
b/sdks/go/test/integration/io/fhirio/fhirio_test.go
index 538d1c6b78f..560e74f4522 100644
--- a/sdks/go/test/integration/io/fhirio/fhirio_test.go
+++ b/sdks/go/test/integration/io/fhirio/fhirio_test.go
@@ -24,6 +24,7 @@ import (
"flag"
"fmt"
"math/big"
+ "net/http"
"os"
"strconv"
"strings"
@@ -62,11 +63,20 @@ func checkFlags(t *testing.T) {
}
}
+func setupFhirStoreWithData(t *testing.T) (string, []string, func()) {
+ return setupFhirStore(t, true)
+}
+
+func setupEmptyFhirStore(t *testing.T) (string, func()) {
+ storePath, _, teardownFunc := setupFhirStore(t, false)
+ return storePath, teardownFunc
+}
+
// Sets up a test fhir store by creating and populating data to it for testing
// purposes. It returns the name of the created store path, a slice of the
// resource paths to be used in tests, and a function to teardown what has been
// set up.
-func setupFhirStore(t *testing.T) (string, []string, func()) {
+func setupFhirStore(t *testing.T, shouldPopulateStore bool) (string, []string,
func()) {
t.Helper()
if storeService == nil || storeManagementService == nil {
t.Fatal("Healthcare Services were not initialized")
@@ -75,13 +85,16 @@ func setupFhirStore(t *testing.T) (string, []string,
func()) {
healthcareDataset := fmt.Sprintf(datasetPathFmt, *gcpopts.Project,
*gcpopts.Region)
createdFhirStore, err := createStore(healthcareDataset)
if err != nil {
- t.Fatal("Test store failed to be created")
+ t.Fatalf("Test store failed to be created. Reason: %v",
err.Error())
}
createdFhirStorePath := createdFhirStore.Name
- resourcePaths := populateStore(createdFhirStorePath)
- if len(resourcePaths) == 0 {
- t.Fatal("No data got populated to test")
+ var resourcePaths []string
+ if shouldPopulateStore {
+ resourcePaths = populateStore(createdFhirStorePath)
+ if len(resourcePaths) == 0 {
+ t.Fatal("No data got populated to test")
+ }
}
return createdFhirStorePath, resourcePaths, func() {
@@ -168,7 +181,7 @@ func TestFhirIO_Read(t *testing.T) {
integration.CheckFilters(t)
checkFlags(t)
- _, testResourcePaths, teardownFhirStore := setupFhirStore(t)
+ _, testResourcePaths, teardownFhirStore := setupFhirStoreWithData(t)
defer teardownFhirStore()
p, s, resourcePaths := ptest.CreateList(testResourcePaths)
@@ -183,7 +196,7 @@ func TestFhirIO_InvalidRead(t *testing.T) {
integration.CheckFilters(t)
checkFlags(t)
- fhirStorePath, _, teardownFhirStore := setupFhirStore(t)
+ fhirStorePath, _, teardownFhirStore := setupFhirStoreWithData(t)
defer teardownFhirStore()
invalidResourcePath := fhirStorePath + "/fhir/Patient/invalid"
@@ -192,12 +205,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,
strconv.Itoa(http.StatusNotFound))
})
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,
strconv.Itoa(http.StatusBadRequest))
+ })
+ ptest.RunAndValidate(t, p)
+}
+
func TestMain(m *testing.M) {
flag.Parse()
beam.Init()
