This is an automated email from the ASF dual-hosted git repository.
nferraro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k.git
The following commit(s) were added to refs/heads/master by this push:
new d8708ca fix(api): Declare integration Flow as json.RawMessage
d8708ca is described below
commit d8708cadd68e01890c822f4f20ffe2a9501a75d7
Author: Antonin Stefanutti <anto...@stefanutti.fr>
AuthorDate: Fri Jul 3 19:31:46 2020 +0200
fix(api): Declare integration Flow as json.RawMessage
---
deploy/crd-integration.yaml | 4 +-
.../integrations.camel.apache.org.crd.yaml | 4 +-
deploy/resources.go | 4 +-
e2e/knative/knative_platform_test.go | 20 ++++--
go.sum | 4 ++
helm/camel-k/crds/crd-integration.yaml | 4 +-
pkg/apis/camel/go.mod | 3 +-
pkg/apis/camel/go.sum | 2 +
pkg/apis/camel/v1/common_types.go | 7 +-
pkg/apis/camel/v1/zz_generated.deepcopy.go | 27 +++++++-
pkg/client/camel/go.sum | 1 +
pkg/cmd/run.go | 7 +-
pkg/trait/init.go | 13 ++--
pkg/util/digest/digest.go | 9 ++-
pkg/util/flow/flow.go | 76 ++++++++++++++++++++++
pkg/util/flow/flow_test.go | 53 +++++++++++++++
16 files changed, 210 insertions(+), 28 deletions(-)
diff --git a/deploy/crd-integration.yaml b/deploy/crd-integration.yaml
index 0366bde..6157ce4 100644
--- a/deploy/crd-integration.yaml
+++ b/deploy/crd-integration.yaml
@@ -88,9 +88,7 @@ spec:
type: array
flows:
items:
- description: Flow is an unstructured object representing a
Camel Flow
- in YAML/JSON DSL
- type: string
+ type: object
type: array
kit:
type: string
diff --git
a/deploy/olm-catalog/camel-k-dev/1.1.0-snapshot/integrations.camel.apache.org.crd.yaml
b/deploy/olm-catalog/camel-k-dev/1.1.0-snapshot/integrations.camel.apache.org.crd.yaml
index 0366bde..6157ce4 100644
---
a/deploy/olm-catalog/camel-k-dev/1.1.0-snapshot/integrations.camel.apache.org.crd.yaml
+++
b/deploy/olm-catalog/camel-k-dev/1.1.0-snapshot/integrations.camel.apache.org.crd.yaml
@@ -88,9 +88,7 @@ spec:
type: array
flows:
items:
- description: Flow is an unstructured object representing a
Camel Flow
- in YAML/JSON DSL
- type: string
+ type: object
type: array
kit:
type: string
diff --git a/deploy/resources.go b/deploy/resources.go
index 573461f..307bd13 100644
--- a/deploy/resources.go
+++ b/deploy/resources.go
@@ -140,9 +140,9 @@ var assets = func() http.FileSystem {
"/crd-integration.yaml": &vfsgen۰CompressedFileInfo{
name: "crd-integration.yaml",
modTime: time.Time{},
- uncompressedSize: 11649,
+ uncompressedSize: 11528,
- compressedContent:
[]byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\x1a\x4d\x6f\xe3\xb8\xf5\xae\x5f\xf1\x10\x1f\x66\x17\x88\xed\xce\xee\xa5\x70\x4f\xae\x27\x41\xdd\xc9\x38\x81\xe5\xd9\xc5\x1c\x69\xe9\x59\x66\x43\x91\x2a\x49\xd9\x49\x8b\xfe\xf7\xe2\x91\x92\x2c\xc9\x92\xe3\x38\x33\x58\x60\x11\xdd\x2c\xbe\xef\x6f\x3e\x79\x00\xc3\xef\xf7\x04\x03\xb8\xe3\x11\x4a\x83\x31\x58\x05\x76\x8b\x30\xcd\x58\xb4\x45\x08\xd5\xc6\xee\x99\x46\xb8\x55\xb9\x8c\x99\xe5\x4a\xc2\x4f\xd3\xf0\xf6\x67\xc8\x
[...]
+ compressedContent:
[]byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\x1a\x4d\x73\xe2\x38\xf6\xee\x5f\xf1\x2a\x1c\x7a\xa6\x2a\xc0\xf6\xcc\x65\x8b\x3d\xb1\x74\x52\xcb\x76\x9a\xa4\x80\x9e\xa9\x39\x0a\xfb\x61\xb4\x91\x25\xad\x24\x43\xb2\x5b\xfb\xdf\xb7\x9e\x64\x1b\x1b\x6c\x42\x48\x77\x4d\x55\x57\x7c\xc3\x7a\xdf\xdf\x7a\xa6\x07\xfd\x6f\xf7\x44\x3d\xb8\xe3\x31\x4a\x8b\x09\x38\x05\x6e\x83\x30\xd6\x2c\xde\x20\x2c\xd4\xda\xed\x98\x41\xb8\x55\xb9\x4c\x98\xe3\x4a\xc2\x4f\xe3\xc5\xed\xcf\x90\x
[...]
},
"/operator-deployment.yaml": &vfsgen۰CompressedFileInfo{
name: "operator-deployment.yaml",
diff --git a/e2e/knative/knative_platform_test.go
b/e2e/knative/knative_platform_test.go
index f3aa846..857a72f 100644
--- a/e2e/knative/knative_platform_test.go
+++ b/e2e/knative/knative_platform_test.go
@@ -25,11 +25,16 @@ import (
"strings"
"testing"
+ . "github.com/onsi/gomega"
+
+ "github.com/stretchr/testify/assert"
+
+ corev1 "k8s.io/api/core/v1"
+
. "github.com/apache/camel-k/e2e/support"
"github.com/apache/camel-k/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/pkg/util/flow"
"github.com/apache/camel-k/pkg/util/knative"
- . "github.com/onsi/gomega"
- corev1 "k8s.io/api/core/v1"
)
func TestKnativePlatformTest(t *testing.T) {
@@ -53,11 +58,12 @@ func TestKnativePlatformTest(t *testing.T) {
// Change something in the integration to produce a
redeploy
Expect(UpdateIntegration(ns, "yaml", func(it
*v1.Integration) {
it.Spec.Profile = ""
- var flows []v1.Flow
- for _, flow := range it.Spec.Flows {
- flows = append(flows,
v1.Flow(strings.ReplaceAll(string(flow), "string!", "string!!!")))
- }
- it.Spec.Flows = flows
+ content, err := flow.Marshal(it.Spec.Flows)
+ assert.NoError(t, err)
+ newData := strings.ReplaceAll(string(content),
"string!", "string!!!")
+ newFlows, err := flow.UnmarshalString(newData)
+ assert.NoError(t, err)
+ it.Spec.Flows = newFlows
})).To(BeNil())
// Spec profile should be reset by "kamel run"
Eventually(IntegrationSpecProfile(ns,
"yaml")).Should(Equal(v1.TraitProfile("")))
diff --git a/go.sum b/go.sum
index c5771d7..3f5902b 100644
--- a/go.sum
+++ b/go.sum
@@ -1375,12 +1375,16 @@ rsc.io/binaryregexp v0.2.0/go.mod
h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8
rsc.io/letsencrypt v0.0.3/go.mod
h1:buyQKZ6IXrRnB7TdkHP0RyEybLx18HHyOSoTyoOLqNY=
sigs.k8s.io/controller-runtime v0.5.2
h1:pyXbUfoTo+HA3jeIfr0vgi+1WtmNh0CwlcnQGLXwsSw=
sigs.k8s.io/controller-runtime v0.5.2/go.mod
h1:JZUwSMVbxDupo0lTJSSFP5pimEyxGynROImSsqIOx1A=
+sigs.k8s.io/controller-tools v0.0.0-20200528125929-5c0c6ae3b64b/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/controller-tools v0.2.4/go.mod
h1:m/ztfQNocGYBgTTCmFdnK94uVvgxeZeE3LtJvd/jIzA=
sigs.k8s.io/controller-tools v0.2.8/go.mod
h1:9VKHPszmf2DHz/QmHkcfZoewO6BL7pPs9uAiBVsaJSE=
+sigs.k8s.io/controller-tools v0.3.0/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/kustomize v2.0.3+incompatible/go.mod
h1:MkjgH3RdOWrievjo6c9T245dYlB5QeXV4WCbnt/PEpU=
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e
h1:4Z09Hglb792X0kfOBBJUPFEyvVfQWrYT/l8h5EKA6JQ=
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e/go.mod
h1:wWxsB5ozmmv/SG7nM11ayaAW51xMvak/t1r0CSlcokI=
sigs.k8s.io/structured-merge-diff/v2 v2.0.1/go.mod
h1:Wb7vfKAodbKgf6tn1Kl0VvGj7mRH6DGaRcixXEJXTsE=
sigs.k8s.io/yaml v1.1.0 h1:4A07+ZFc2wgJwo8YNlQpr1rVlgUDlxXHhPJciaPY5gs=
sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
+sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
vbom.ml/util v0.0.0-20160121211510-db5cfe13f5cc/go.mod
h1:so/NYdZXCz+E3ZpW0uAoCj6uzU2+8OWDFv/HxUSs7kI=
diff --git a/helm/camel-k/crds/crd-integration.yaml
b/helm/camel-k/crds/crd-integration.yaml
index 0366bde..6157ce4 100644
--- a/helm/camel-k/crds/crd-integration.yaml
+++ b/helm/camel-k/crds/crd-integration.yaml
@@ -88,9 +88,7 @@ spec:
type: array
flows:
items:
- description: Flow is an unstructured object representing a
Camel Flow
- in YAML/JSON DSL
- type: string
+ type: object
type: array
kit:
type: string
diff --git a/pkg/apis/camel/go.mod b/pkg/apis/camel/go.mod
index d50e11b..c7bcb9f 100644
--- a/pkg/apis/camel/go.mod
+++ b/pkg/apis/camel/go.mod
@@ -5,6 +5,7 @@ go 1.13
require (
k8s.io/api v0.18.2
k8s.io/apimachinery v0.18.2
- sigs.k8s.io/controller-tools v0.3.0 // indirect
+ // Required to get
https://github.com/kubernetes-sigs/controller-tools/pull/428
+ sigs.k8s.io/controller-tools v0.0.0-20200528125929-5c0c6ae3b64b //
indirect
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e //
indirect
)
diff --git a/pkg/apis/camel/go.sum b/pkg/apis/camel/go.sum
index afd0e7e..60acd80 100644
--- a/pkg/apis/camel/go.sum
+++ b/pkg/apis/camel/go.sum
@@ -415,6 +415,8 @@ k8s.io/kube-openapi
v0.0.0-20200121204235-bf4fb3bd569c/go.mod h1:GRQhZsXIAJ1xR0C
k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89
h1:d4vVOjXm687F1iLSP2q3lyPPuyvTUt3aVoBpi2DqRsU=
k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod
h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.7/go.mod
h1:PHgbrJT7lCHcxMU+mDHEm+nx46H4zuuHZkDP6icnhu0=
+sigs.k8s.io/controller-tools v0.0.0-20200528125929-5c0c6ae3b64b
h1:jVf/McoMd0tHALAJrr4VgEVakuOhEYQ+m00kJTseL3s=
+sigs.k8s.io/controller-tools v0.0.0-20200528125929-5c0c6ae3b64b/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/controller-tools v0.3.0
h1:y3YD99XOyWaXkiF1kd41uRvfp/64teWcrEZFuHxPhJ4=
sigs.k8s.io/controller-tools v0.3.0/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e
h1:4Z09Hglb792X0kfOBBJUPFEyvVfQWrYT/l8h5EKA6JQ=
diff --git a/pkg/apis/camel/v1/common_types.go
b/pkg/apis/camel/v1/common_types.go
index ee660f7..47d4afc 100644
--- a/pkg/apis/camel/v1/common_types.go
+++ b/pkg/apis/camel/v1/common_types.go
@@ -18,6 +18,8 @@ limitations under the License.
package v1
import (
+ "encoding/json"
+
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -133,4 +135,7 @@ type ResourceCondition interface {
}
// Flow is an unstructured object representing a Camel Flow in YAML/JSON DSL
-type Flow string
+// +kubebuilder:validation:Type=object
+type Flow struct {
+ json.RawMessage `json:",inline"`
+}
diff --git a/pkg/apis/camel/v1/zz_generated.deepcopy.go
b/pkg/apis/camel/v1/zz_generated.deepcopy.go
index 3cd5a5f..f8f17f1 100644
--- a/pkg/apis/camel/v1/zz_generated.deepcopy.go
+++ b/pkg/apis/camel/v1/zz_generated.deepcopy.go
@@ -5,6 +5,8 @@
package v1
import (
+ json "encoding/json"
+
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
runtime "k8s.io/apimachinery/pkg/runtime"
@@ -618,6 +620,27 @@ func (in *FailureRecovery) DeepCopy() *FailureRecovery {
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver,
writing into out. in must be non-nil.
+func (in *Flow) DeepCopyInto(out *Flow) {
+ *out = *in
+ if in.RawMessage != nil {
+ in, out := &in.RawMessage, &out.RawMessage
+ *out = make(json.RawMessage, len(*in))
+ copy(*out, *in)
+ }
+ return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver,
creating a new Flow.
+func (in *Flow) DeepCopy() *Flow {
+ if in == nil {
+ return nil
+ }
+ out := new(Flow)
+ in.DeepCopyInto(out)
+ return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver,
writing into out. in must be non-nil.
func (in *ImageTask) DeepCopyInto(out *ImageTask) {
*out = *in
in.ContainerTask.DeepCopyInto(&out.ContainerTask)
@@ -1086,7 +1109,9 @@ func (in *IntegrationSpec) DeepCopyInto(out
*IntegrationSpec) {
if in.Flows != nil {
in, out := &in.Flows, &out.Flows
*out = make([]Flow, len(*in))
- copy(*out, *in)
+ for i := range *in {
+ (*in)[i].DeepCopyInto(&(*out)[i])
+ }
}
if in.Resources != nil {
in, out := &in.Resources, &out.Resources
diff --git a/pkg/client/camel/go.sum b/pkg/client/camel/go.sum
index 29c91ed..13fbe75 100644
--- a/pkg/client/camel/go.sum
+++ b/pkg/client/camel/go.sum
@@ -464,6 +464,7 @@ modernc.org/mathutil v1.0.0/go.mod
h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03
modernc.org/strutil v1.0.0/go.mod
h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs=
modernc.org/xc v1.0.0/go.mod h1:mRNCo0bvLjGhHO9WsyuKVU4q0ceiDDDoEeWDJHrNx8I=
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.7/go.mod
h1:PHgbrJT7lCHcxMU+mDHEm+nx46H4zuuHZkDP6icnhu0=
+sigs.k8s.io/controller-tools v0.0.0-20200528125929-5c0c6ae3b64b/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/controller-tools v0.3.0/go.mod
h1:enhtKGfxZD1GFEoMgP8Fdbu+uKQ/cq1/WGJhdVChfvI=
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e
h1:4Z09Hglb792X0kfOBBJUPFEyvVfQWrYT/l8h5EKA6JQ=
sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e/go.mod
h1:wWxsB5ozmmv/SG7nM11ayaAW51xMvak/t1r0CSlcokI=
diff --git a/pkg/cmd/run.go b/pkg/cmd/run.go
index fce05e7..5927a47 100644
--- a/pkg/cmd/run.go
+++ b/pkg/cmd/run.go
@@ -36,6 +36,7 @@ import (
"github.com/apache/camel-k/pkg/client"
"github.com/apache/camel-k/pkg/trait"
"github.com/apache/camel-k/pkg/util"
+ "github.com/apache/camel-k/pkg/util/flow"
"github.com/apache/camel-k/pkg/util/gzip"
"github.com/apache/camel-k/pkg/util/kubernetes"
k8slog "github.com/apache/camel-k/pkg/util/kubernetes/log"
@@ -471,7 +472,11 @@ func (o *runCmdOptions) updateIntegrationCode(c
client.Client, sources []string)
}
if o.UseFlows && (strings.HasSuffix(source, ".yaml") ||
strings.HasSuffix(source, ".yml")) {
- integration.Spec.AddFlows(v1.Flow(data))
+ flows, err := flow.UnmarshalString(data)
+ if err != nil {
+ return nil, err
+ }
+ integration.Spec.AddFlows(flows...)
} else {
integration.Spec.AddSources(v1.SourceSpec{
DataSpec: v1.DataSpec{
diff --git a/pkg/trait/init.go b/pkg/trait/init.go
index d060021..32236e9 100644
--- a/pkg/trait/init.go
+++ b/pkg/trait/init.go
@@ -25,9 +25,10 @@ import (
v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
"github.com/apache/camel-k/pkg/util"
+ "github.com/apache/camel-k/pkg/util/flow"
)
-const flowsInternalSourceName = "camel-k-embedded-flow-%d.yaml"
+const flowsInternalSourceName = "camel-k-embedded-flow.yaml"
// Internal trait
type initTrait struct {
@@ -52,11 +53,15 @@ func (t *initTrait) Apply(e *Environment) error {
if e.IntegrationInPhase(v1.IntegrationPhaseInitialization) {
// Flows need to be turned into a generated source
- for i, flow := range e.Integration.Spec.Flows {
+ if len(e.Integration.Spec.Flows) > 0 {
+ content, err := flow.Marshal(e.Integration.Spec.Flows)
+ if err != nil {
+ return err
+ }
e.Integration.Status.AddOrReplaceGeneratedSources(v1.SourceSpec{
DataSpec: v1.DataSpec{
- Name:
fmt.Sprintf(flowsInternalSourceName, i),
- Content: string(flow),
+ Name: flowsInternalSourceName,
+ Content: string(content),
},
})
}
diff --git a/pkg/util/digest/digest.go b/pkg/util/digest/digest.go
index bc4ff63..1a39800 100644
--- a/pkg/util/digest/digest.go
+++ b/pkg/util/digest/digest.go
@@ -31,6 +31,7 @@ import (
v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
"github.com/apache/camel-k/pkg/util"
"github.com/apache/camel-k/pkg/util/defaults"
+ "github.com/apache/camel-k/pkg/util/flow"
)
// ComputeForIntegration a digest of the fields that are relevant for the
deployment
@@ -67,8 +68,12 @@ func ComputeForIntegration(integration *v1.Integration)
(string, error) {
}
// Integration flows
- for _, flow := range integration.Spec.Flows {
- if _, err := hash.Write([]byte(flow)); err != nil {
+ if len(integration.Spec.Flows) > 0 {
+ flows, err := flow.Marshal(integration.Spec.Flows)
+ if err != nil {
+ return "", err
+ }
+ if _, err := hash.Write(flows); err != nil {
return "", err
}
}
diff --git a/pkg/util/flow/flow.go b/pkg/util/flow/flow.go
new file mode 100644
index 0000000..aaabd4d
--- /dev/null
+++ b/pkg/util/flow/flow.go
@@ -0,0 +1,76 @@
+/*
+ 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 flow
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "io"
+ "io/ioutil"
+
+ yaml2 "gopkg.in/yaml.v2"
+
+ "k8s.io/apimachinery/pkg/util/yaml"
+
+ v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+)
+
+// UnmarshalString reads flows contained in a string
+func UnmarshalString(flowsString string) ([]v1.Flow, error) {
+ return Unmarshal(bytes.NewReader([]byte(flowsString)))
+}
+
+// Unmarshal flows from a stream
+func Unmarshal(reader io.Reader) ([]v1.Flow, error) {
+ buffered, err := ioutil.ReadAll(reader)
+ if err != nil {
+ return nil, err
+ }
+ var flows []v1.Flow
+ // Using the Kubernetes decoder to turn them into JSON before unmarshal.
+ // This avoids having map[interface{}]interface{} objects which are not
JSON compatible.
+ jsonData, err := yaml.ToJSON(buffered)
+ if err != nil {
+ return nil, err
+ }
+
+ if err = json.Unmarshal(jsonData, &flows); err != nil {
+ return nil, err
+ }
+ return flows, err
+}
+
+// Marshal flows as byte array
+func Marshal(flows []v1.Flow) ([]byte, error) {
+ data, err := json.Marshal(&flows)
+ if err != nil {
+ return nil, err
+ }
+ jsondata := make([]map[string]interface{}, 0)
+ err = json.Unmarshal(data, &jsondata)
+ if err != nil {
+ return nil, fmt.Errorf("error unmarshalling json: %v", err)
+ }
+ yamldata, err := yaml2.Marshal(&jsondata)
+ if err != nil {
+ return nil, fmt.Errorf("error marshalling to yaml: %v", err)
+ }
+
+ return yamldata, nil
+}
diff --git a/pkg/util/flow/flow_test.go b/pkg/util/flow/flow_test.go
new file mode 100644
index 0000000..4afb2ef
--- /dev/null
+++ b/pkg/util/flow/flow_test.go
@@ -0,0 +1,53 @@
+/*
+ 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 flow
+
+import (
+ "bytes"
+ "encoding/json"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestReadWriteYaml(t *testing.T) {
+ // yaml in conventional form as marshalled by the go runtime
+ yaml := `- from:
+ steps:
+ - to: log:info
+ uri: timer:tick
+`
+
+ yamlReader := bytes.NewReader([]byte(yaml))
+ flows, err := Unmarshal(yamlReader)
+ assert.NoError(t, err)
+ assert.NotNil(t, flows)
+ assert.Len(t, flows, 1)
+
+ flow := map[string]interface{}{}
+ err = json.Unmarshal(flows[0].RawMessage, &flow)
+ assert.NoError(t, err)
+
+ assert.NotNil(t, flow["from"])
+ assert.Nil(t, flow["xx"])
+
+ data, err := Marshal(flows)
+ assert.NoError(t, err)
+ assert.NotNil(t, data)
+ assert.Equal(t, yaml, string(data))
+}
