Repository: apache/camel-k
Pull request #3828 (2022-11-19 00:00:00)

chore(test) : TestRunAndLog to pass on mac os
* Fix: #3813

Signed-off-by: yugo-horie <u5.horie@gmail.com>

<!-- Description -->

#3813 

We divided the test case each OS.

<!--
Enter your extended release note in the below block. If the PR requires
additional action from users switching to the new release, include the string
"action required". If no release note is required, write "NONE". 

You can (optionally) mark this PR with labels "kind/bug" or "kind/feature" to make sure
the text is added to the right section of the release notes. 
-->

**Release Note**
```release-note
TestRunAndLog to pass both linux and MacOS
```



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e3debdb13..1ae169fe3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,12 +6,14 @@
 
 **Closed issues:**
 
+- Remove deprecated Kubernetes API [\#3840](https://github.com/apache/camel-k/issues/3840)
 - gc trait fails for: cannot list resource \"endpoints\" in API group \"submariner.io\" [\#3833](https://github.com/apache/camel-k/issues/3833)
 - Bundle m2 camel k runtime distribution [\#3826](https://github.com/apache/camel-k/issues/3826)
 - E2E - Builder tests are constantly failing due to "Out of date version of GraalVM detected: native-image 21.3.0.0-Final" [\#3817](https://github.com/apache/camel-k/issues/3817)
 - query regarding using camel k [\#3815](https://github.com/apache/camel-k/issues/3815)
 - FTP sink does not create directories when writing [\#3812](https://github.com/apache/camel-k/issues/3812)
 - Release 1.10.3 [\#3796](https://github.com/apache/camel-k/issues/3796)
+- Deprecate overlapping Kamel CLI features in favour of Camel JBang [\#3790](https://github.com/apache/camel-k/issues/3790)
 - Remove code deprecated before 1.8 \(LTS\) [\#3787](https://github.com/apache/camel-k/issues/3787)
 - Native checks to run on a nightly basis instead of on each PR [\#3784](https://github.com/apache/camel-k/issues/3784)
 - Ready condition message not always taken from Camel Health Check [\#3761](https://github.com/apache/camel-k/issues/3761)
@@ -20,6 +22,7 @@
 - Drop support for Openshift 3 and related E2E test [\#3421](https://github.com/apache/camel-k/issues/3421)
 - Turn AWS region parameter into an enum? [\#3289](https://github.com/apache/camel-k/issues/3289)
 - Install Camel-K Operator based on Helm Chart with --global option [\#3054](https://github.com/apache/camel-k/issues/3054)
+- Camel 3.15 - Data formats renamed [\#2880](https://github.com/apache/camel-k/issues/2880)
 
 ## [1.10.4-nightly](https://github.com/apache/camel-k/tree/1.10.4-nightly) (2022-11-14)
 
diff --git a/pkg/apis/camel/v1/camelcatalog_types_support.go b/pkg/apis/camel/v1/camelcatalog_types_support.go
index 1106a8b51..3434e255b 100644
--- a/pkg/apis/camel/v1/camelcatalog_types_support.go
+++ b/pkg/apis/camel/v1/camelcatalog_types_support.go
@@ -101,20 +101,20 @@ func (in *CamelArtifact) GetDependencyID() string {
 	}
 }
 
-func (in *CamelArtifact) GetConsumerDependencyIDs(schemeID string) (deps []string) {
+func (in *CamelArtifact) GetConsumerDependencyIDs(schemeID string) []string {
 	return in.getDependencyIDs(schemeID, consumerScheme)
 }
 
-func (in *CamelArtifact) GetProducerDependencyIDs(schemeID string) (deps []string) {
+func (in *CamelArtifact) GetProducerDependencyIDs(schemeID string) []string {
 	return in.getDependencyIDs(schemeID, producerScheme)
 }
 
-func (in *CamelArtifact) getDependencyIDs(schemeID string, scope func(CamelScheme) CamelSchemeScope) (deps []string) {
+func (in *CamelArtifact) getDependencyIDs(schemeID string, scope func(CamelScheme) CamelSchemeScope) []string {
 	ads := in.getDependencies(schemeID, scope)
 	if ads == nil {
-		return deps
+		return nil
 	}
-	deps = make([]string, 0, len(ads))
+	deps := make([]string, 0, len(ads))
 	for _, ad := range ads {
 		deps = append(deps, ad.GetDependencyID())
 	}
diff --git a/pkg/cmd/init.go b/pkg/cmd/init.go
index 59347482e..1187b23bb 100644
--- a/pkg/cmd/init.go
+++ b/pkg/cmd/init.go
@@ -36,10 +36,11 @@ func newCmdInit(rootCmdOptions *RootCmdOptions) (*cobra.Command, *initCmdOptions
 		RootCmdOptions: rootCmdOptions,
 	}
 	cmd := cobra.Command{
-		Use:     "init [flags] IntegrationFile.java",
-		Short:   "Initialize empty Camel K files",
-		Long:    `Initialize empty Camel K integrations and other resources.`,
-		PreRunE: decode(&options),
+		Use:        "init [flags] IntegrationFile.java",
+		Short:      "Initialize empty Camel K files",
+		Long:       `Initialize empty Camel K integrations and other resources.`,
+		Deprecated: "consider using Camel JBang instead (https://camel.apache.org/manual/camel-jbang.html)",
+		PreRunE:    decode(&options),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if err := options.validate(cmd, args); err != nil {
 				return err
diff --git a/pkg/cmd/local.go b/pkg/cmd/local.go
index 70a0f7a15..da816af9a 100644
--- a/pkg/cmd/local.go
+++ b/pkg/cmd/local.go
@@ -42,6 +42,7 @@ func newCmdLocal(rootCmdOptions *RootCmdOptions) (*cobra.Command, *LocalCmdOptio
 		Use:               "local [sub-command]",
 		Short:             "Perform integration actions locally.",
 		Long:              `Perform integration actions locally given a set of input integration files.`,
+		Deprecated:        "consider using Camel JBang instead (https://camel.apache.org/manual/camel-jbang.html)",
 		PersistentPreRunE: options.persistentPreRun,
 		Annotations: map[string]string{
 			offlineCommandLabel: "true",
diff --git a/pkg/cmd/local_build.go b/pkg/cmd/local_build.go
index 6bd124a1e..8ebb2e00c 100644
--- a/pkg/cmd/local_build.go
+++ b/pkg/cmd/local_build.go
@@ -33,10 +33,11 @@ func newCmdLocalBuild(localCmdOptions *LocalCmdOptions) (*cobra.Command, *localB
 	}
 
 	cmd := cobra.Command{
-		Use:     "build [options]",
-		Short:   "Build integration images locally.",
-		Long:    `Build integration images locally for containerized integrations.`,
-		PreRunE: decode(&options),
+		Use:        "build [options]",
+		Short:      "Build integration images locally.",
+		Long:       `Build integration images locally for containerized integrations.`,
+		Deprecated: "it may be removed in future versions",
+		PreRunE:    decode(&options),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if err := options.validate(cmd, args); err != nil {
 				return err
diff --git a/pkg/cmd/local_inspect.go b/pkg/cmd/local_inspect.go
index 7b47e3374..7ffdf8af2 100644
--- a/pkg/cmd/local_inspect.go
+++ b/pkg/cmd/local_inspect.go
@@ -36,7 +36,8 @@ func newCmdLocalInspect(localCmdOptions *LocalCmdOptions) (*cobra.Command, *loca
 		Long: `Output dependencies for a list of integration files. By default this command returns the
 top level dependencies only. When --all-dependencies is enabled, the transitive dependencies
 will be generated by calling Maven and then printed in the selected output format.`,
-		PreRunE: decode(&options),
+		Deprecated: "it may be removed in future versions",
+		PreRunE:    decode(&options),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if err := options.validate(args); err != nil {
 				return err
diff --git a/pkg/cmd/local_run.go b/pkg/cmd/local_run.go
index e88ff0903..c9bf2ff88 100644
--- a/pkg/cmd/local_run.go
+++ b/pkg/cmd/local_run.go
@@ -35,10 +35,11 @@ func newCmdLocalRun(localCmdOptions *LocalCmdOptions) (*cobra.Command, *localRun
 	}
 
 	cmd := cobra.Command{
-		Use:     "run [integration files]",
-		Short:   "Run integration locally.",
-		Long:    `Run integration locally using the input integration files.`,
-		PreRunE: decode(&options),
+		Use:        "run [integration files]",
+		Short:      "Run integration locally.",
+		Long:       `Run integration locally using the input integration files.`,
+		Deprecated: "consider using Camel JBang instead (https://camel.apache.org/manual/camel-jbang.html)",
+		PreRunE:    decode(&options),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if err := options.validate(args); err != nil {
 				return err
diff --git a/pkg/trait/pdb.go b/pkg/trait/pdb.go
index 6f02265f1..78d960d9f 100644
--- a/pkg/trait/pdb.go
+++ b/pkg/trait/pdb.go
@@ -20,7 +20,7 @@ package trait
 import (
 	"fmt"
 
-	"k8s.io/api/policy/v1beta1"
+	policyv1 "k8s.io/api/policy/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/pointer"
@@ -72,18 +72,18 @@ func (t *pdbTrait) Apply(e *Environment) error {
 	return nil
 }
 
-func (t *pdbTrait) podDisruptionBudgetFor(integration *v1.Integration) *v1beta1.PodDisruptionBudget {
-	pdb := &v1beta1.PodDisruptionBudget{
+func (t *pdbTrait) podDisruptionBudgetFor(integration *v1.Integration) *policyv1.PodDisruptionBudget {
+	pdb := &policyv1.PodDisruptionBudget{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "PodDisruptionBudget",
-			APIVersion: v1beta1.SchemeGroupVersion.String(),
+			APIVersion: policyv1.SchemeGroupVersion.String(),
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      integration.Name,
 			Namespace: integration.Namespace,
 			Labels:    integration.Labels,
 		},
-		Spec: v1beta1.PodDisruptionBudgetSpec{
+		Spec: policyv1.PodDisruptionBudgetSpec{
 			Selector: &metav1.LabelSelector{
 				MatchLabels: map[string]string{
 					v1.IntegrationLabel: integration.Name,
diff --git a/pkg/util/camel/camel_runtime_catalog.go b/pkg/util/camel/camel_runtime_catalog.go
index 25da254e4..3305bd10b 100644
--- a/pkg/util/camel/camel_runtime_catalog.go
+++ b/pkg/util/camel/camel_runtime_catalog.go
@@ -186,9 +186,8 @@ func (c *RuntimeCatalog) DecodeComponent(uri string) (*v1.CamelArtifact, *v1.Cam
 		return nil, nil
 	}
 	uriStart := uriSplit[0]
-	scheme, ok := c.GetScheme(uriStart)
 	var schemeRef *v1.CamelScheme
-	if ok {
+	if scheme, ok := c.GetScheme(uriStart); ok {
 		schemeRef = &scheme
 	}
 	return c.GetArtifactByScheme(uriStart), schemeRef
diff --git a/pkg/util/source/inspector.go b/pkg/util/source/inspector.go
index b8334bae6..694cad3b8 100644
--- a/pkg/util/source/inspector.go
+++ b/pkg/util/source/inspector.go
@@ -353,6 +353,29 @@ func (i *baseInspector) addDependencies(uri string, meta *Metadata, consumer boo
 		meta.AddDependency(dep)
 	}
 
+	// some components require additional dependency resolution based on URI
+	if err := i.addDependenciesFromURI(uri, scheme, meta); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (i *baseInspector) addDependenciesFromURI(uri string, scheme *v1.CamelScheme, meta *Metadata) error {
+	if scheme.ID == "dataformat" {
+		// dataformat:name:(marshal|unmarshal)[?options]
+		parts := strings.Split(uri, ":")
+		if len(parts) < 3 {
+			return fmt.Errorf("invalid dataformat uri: %s", uri)
+		}
+		name := parts[1]
+		df := i.catalog.GetArtifactByDataFormat(name)
+		if df == nil {
+			return fmt.Errorf("dataformat %q not found: %s", name, uri)
+		}
+		meta.AddDependency(df.GetDependencyID())
+	}
+
 	return nil
 }
 
diff --git a/pkg/util/source/inspector_xml.go b/pkg/util/source/inspector_xml.go
index 3e55eb3b9..f1fc4be71 100644
--- a/pkg/util/source/inspector_xml.go
+++ b/pkg/util/source/inspector_xml.go
@@ -48,6 +48,16 @@ func (i XMLInspector) Extract(source v1.SourceSpec, meta *Metadata) error {
 				meta.RequiredCapabilities.Add(v1.CapabilityRest)
 			case "circuitBreaker":
 				meta.RequiredCapabilities.Add(v1.CapabilityCircuitBreaker)
+			case "json":
+				dataFormatID := defaultJSONDataFormat
+				for _, a := range se.Attr {
+					if a.Name.Local == "library" {
+						dataFormatID = strings.ToLower(a.Value)
+					}
+				}
+				if dfDep := i.catalog.GetArtifactByDataFormat(dataFormatID); dfDep != nil {
+					meta.AddDependency(dfDep.GetDependencyID())
+				}
 			case "language":
 				for _, a := range se.Attr {
 					if a.Name.Local == "language" {
diff --git a/pkg/util/source/inspector_yaml.go b/pkg/util/source/inspector_yaml.go
index e672f763d..42cd9f58e 100644
--- a/pkg/util/source/inspector_yaml.go
+++ b/pkg/util/source/inspector_yaml.go
@@ -69,9 +69,7 @@ func (i YAMLInspector) parseStep(key string, content interface{}, meta *Metadata
 		meta.RequiredCapabilities.Add(v1.CapabilityRest)
 	case "circuitBreaker":
 		meta.RequiredCapabilities.Add(v1.CapabilityCircuitBreaker)
-	case "unmarshal":
-		fallthrough
-	case "marshal":
+	case "marshal", "unmarshal":
 		if cm, ok := content.(map[interface{}]interface{}); ok {
 			if js, jsOk := cm["json"]; jsOk {
 				dataFormatID := defaultJSONDataFormat


Test patch:
diff --git a/e2e/global/common/traits/pdb_test.go b/e2e/global/common/traits/pdb_test.go
index 6a18b463a..e0a69993c 100644
--- a/e2e/global/common/traits/pdb_test.go
+++ b/e2e/global/common/traits/pdb_test.go
@@ -31,7 +31,7 @@ import (
 	. "github.com/onsi/gomega/gstruct"
 
 	corev1 "k8s.io/api/core/v1"
-	policy "k8s.io/api/policy/v1beta1"
+	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -109,7 +109,7 @@ func TestPodDisruptionBudgetTrait(t *testing.T) {
 		// Eviction attempt
 		pods := IntegrationPods(ns, name)()
 		Expect(pods).To(HaveLen(2))
-		err := TestClient().CoreV1().Pods(ns).Evict(TestContext, &policy.Eviction{
+		err := TestClient().CoreV1().Pods(ns).EvictV1(TestContext, &policyv1.Eviction{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: pods[0].Name,
 			},
@@ -141,7 +141,7 @@ func TestPodDisruptionBudgetTrait(t *testing.T) {
 
 		pods = IntegrationPods(ns, name)()
 		Expect(pods).To(HaveLen(3))
-		Expect(TestClient().CoreV1().Pods(ns).Evict(TestContext, &policy.Eviction{
+		Expect(TestClient().CoreV1().Pods(ns).EvictV1(TestContext, &policyv1.Eviction{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: pods[0].Name,
 			},
@@ -152,11 +152,11 @@ func TestPodDisruptionBudgetTrait(t *testing.T) {
 	})
 }
 
-func podDisruptionBudget(ns string, name string) func() *policy.PodDisruptionBudget {
-	return func() *policy.PodDisruptionBudget {
-		pdb := policy.PodDisruptionBudget{
+func podDisruptionBudget(ns string, name string) func() *policyv1.PodDisruptionBudget {
+	return func() *policyv1.PodDisruptionBudget {
+		pdb := policyv1.PodDisruptionBudget{
 			TypeMeta: metav1.TypeMeta{
-				APIVersion: policy.SchemeGroupVersion.String(),
+				APIVersion: policyv1.SchemeGroupVersion.String(),
 				Kind:       "PodDisruptionBudget",
 			},
 			ObjectMeta: metav1.ObjectMeta{
diff --git a/pkg/trait/pdb_test.go b/pkg/trait/pdb_test.go
index af6e27da4..63857dd05 100644
--- a/pkg/trait/pdb_test.go
+++ b/pkg/trait/pdb_test.go
@@ -24,7 +24,7 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/api/policy/v1beta1"
+	policyv1 "k8s.io/api/policy/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 
@@ -73,7 +73,7 @@ func TestPdbIsCreatedWithMinAvailable(t *testing.T) {
 	assert.Equal(t, int32(2), pdb.Spec.MinAvailable.IntVal)
 }
 
-func pdbCreatedCheck(t *testing.T, pdbTrait *pdbTrait, environment *Environment) *v1beta1.PodDisruptionBudget {
+func pdbCreatedCheck(t *testing.T, pdbTrait *pdbTrait, environment *Environment) *policyv1.PodDisruptionBudget {
 	t.Helper()
 
 	err := pdbTrait.Apply(environment)
@@ -87,9 +87,9 @@ func pdbCreatedCheck(t *testing.T, pdbTrait *pdbTrait, environment *Environment)
 	return pdb
 }
 
-func findPdb(resources *kubernetes.Collection) *v1beta1.PodDisruptionBudget {
+func findPdb(resources *kubernetes.Collection) *policyv1.PodDisruptionBudget {
 	for _, a := range resources.Items() {
-		if pdb, ok := a.(*v1beta1.PodDisruptionBudget); ok {
+		if pdb, ok := a.(*policyv1.PodDisruptionBudget); ok {
 			return pdb
 		}
 	}
diff --git a/pkg/util/command_test.go b/pkg/util/command_test.go
index 98187b097..782f1804e 100644
--- a/pkg/util/command_test.go
+++ b/pkg/util/command_test.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"os/exec"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -30,17 +29,11 @@ import (
 var (
 	loggerInfo = func(s string) string {
 		fmt.Println("OUT:", s)
-		if strings.Contains(s, "invalid") {
-			return s
-		}
-		return ""
+		return s
 	}
 	loggerError = func(s string) string {
 		fmt.Println("ERR:", s)
-		if strings.Contains(s, "invalid") {
-			return s
-		}
-		return ""
+		return s
 	}
 )
 
@@ -52,9 +45,9 @@ func TestRunAndLog(t *testing.T) {
 }
 
 func TestRunAndLogInvalid(t *testing.T) {
-	cmd := exec.CommandContext(context.Background(), "date", "-dsa")
+	cmd := exec.CommandContext(context.Background(), "go", "help", "dsa")
 	err := RunAndLog(context.Background(), cmd, loggerInfo, loggerError)
 
 	assert.NotNil(t, err)
-	assert.ErrorContains(t, err, "exit status 1")
+	assert.Equal(t, "go help dsa: unknown help topic. Run 'go help'.: exit status 2", err.Error())
 }
diff --git a/pkg/util/source/inspector_groovy_test.go b/pkg/util/source/inspector_groovy_test.go
index d0d2a5e80..5dc3f05ae 100644
--- a/pkg/util/source/inspector_groovy_test.go
+++ b/pkg/util/source/inspector_groovy_test.go
@@ -21,34 +21,46 @@ import (
 	"fmt"
 	"testing"
 
-	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
-
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/apache/camel-k/pkg/util/camel"
 )
 
-const GroovyKameletEip = `
+func newTestGroovyInspector(t *testing.T) GroovyInspector {
+	t.Helper()
+
+	catalog, err := camel.DefaultCatalog()
+	require.NoError(t, err)
+
+	return GroovyInspector{
+		baseInspector: baseInspector{
+			catalog: catalog,
+		},
+	}
+}
+
+const groovyKameletEip = `
 from("direct:start")
     .kamelet("foo/bar?baz=test")
 `
 
-const GroovyKameletEipSingleQuote = `
+const groovyKameletEipSingleQuote = `
 from("direct:start")
     .kamelet('foo/bar?baz=test')
 `
 
-const GroovyKameletEndpoint = `
+const groovyKameletEndpoint = `
 from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
-const GroovyKameletEndpointSingleQuote = `
+const groovyKameletEndpointSingleQuote = `
 from("direct:start")
     .to('kamelet:foo/bar?baz=test')
 `
 
-const GroovyWireTapEipSingleQuote = `
+const groovyWireTapEipSingleQuote = `
 from("direct:start")
     .wireTap('kamelet:foo/bar?baz=test')
 `
@@ -59,53 +71,102 @@ func TestGroovyKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   GroovyKameletEip,
+			source:   groovyKameletEip,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   GroovyKameletEndpoint,
+			source:   groovyKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   GroovyKameletEipSingleQuote,
+			source:   groovyKameletEipSingleQuote,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   GroovyKameletEndpointSingleQuote,
+			source:   groovyKameletEndpointSingleQuote,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   GroovyWireTapEipSingleQuote,
+			source:   groovyWireTapEipSingleQuote,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestGroovyInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestGroovyKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := GroovyInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
+		})
+	}
+}
+
+const groovyJSONEip = `
+from("direct:start")
+    .unmarshal().json()
+`
+
+const groovyJSONJacksonEip = `
+from("direct:start")
+    .unmarshal().json(JsonLibrary.Jackson)
+`
+
+const groovyAvroEndpoint = `
+from("direct:start")
+    .to("dataformat:avro:marshal")
+`
+
+const groovyJacksonEndpoint = `
+from("direct:start")
+    .to("dataformat:jackson:marshal")
+`
+
+const groovyProtobufEndpoint = `
+from("direct:start")
+    .to("dataformat:protobuf:marshal")
+`
+
+func TestGroovyDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: groovyJSONEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: groovyJSONJacksonEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: groovyAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: groovyJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: groovyProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestGroovyInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestGroovyDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/inspector_java_script_test.go b/pkg/util/source/inspector_java_script_test.go
index 3ed09d928..05d5efa89 100644
--- a/pkg/util/source/inspector_java_script_test.go
+++ b/pkg/util/source/inspector_java_script_test.go
@@ -21,23 +21,36 @@ import (
 	"fmt"
 	"testing"
 
-	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/camel"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-const JavaScriptKameletEip = `
+func newTestJavaScriptInspector(t *testing.T) JavaScriptInspector {
+	t.Helper()
+
+	catalog, err := camel.DefaultCatalog()
+	require.NoError(t, err)
+
+	return JavaScriptInspector{
+		baseInspector: baseInspector{
+			catalog: catalog,
+		},
+	}
+}
+
+const javaScriptKameletEip = `
 from("direct:start")
     .kamelet("foo/bar?baz=test")
 `
 
-const JavaScriptKameletEndpoint = `
+const javaScriptKameletEndpoint = `
 from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
-const JavaScriptWireTapEipSingleQuote = `
+const javaScriptWireTapEipSingleQuote = `
 from("direct:start")
     .wireTap('kamelet:foo/bar?baz=test')
 `
@@ -48,45 +61,94 @@ func TestJavaScriptKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   JavaScriptKameletEip,
+			source:   javaScriptKameletEip,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   JavaScriptKameletEndpoint,
+			source:   javaScriptKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   JavaScriptWireTapEipSingleQuote,
+			source:   javaScriptWireTapEipSingleQuote,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestJavaScriptInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestJavaScriptKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := JavaScriptInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
+		})
+	}
+}
+
+const javaScriptJSONEip = `
+from('direct:start')
+    .unmarshal().json()
+`
+
+const javaScriptJSONJacksonEip = `
+from('direct:start')
+    .unmarshal().json(JsonLibrary.Jackson)
+`
+
+const javaScriptAvroEndpoint = `
+from('direct:start')
+    .to('dataformat:avro:marshal');
+`
+
+const javaScriptJacksonEndpoint = `
+from('direct:start')
+    .to('dataformat:jackson:marshal')
+`
+
+const javaScriptProtobufEndpoint = `
+from('direct:start')
+    .to('dataformat:protobuf:marshal')
+`
+
+func TestJavaScriptDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: javaScriptJSONEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: javaScriptJSONJacksonEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: javaScriptAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: javaScriptJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: javaScriptProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestJavaScriptInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestJavaScriptDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/inspector_java_source_test.go b/pkg/util/source/inspector_java_source_test.go
index af1a8e3af..d84a0d868 100644
--- a/pkg/util/source/inspector_java_source_test.go
+++ b/pkg/util/source/inspector_java_source_test.go
@@ -21,25 +21,38 @@ import (
 	"fmt"
 	"testing"
 
-	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/camel"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-const JavaSourceKameletEip = `
+func newTestJavaSourceInspector(t *testing.T) JavaSourceInspector {
+	t.Helper()
+
+	catalog, err := camel.DefaultCatalog()
+	require.NoError(t, err)
+
+	return JavaSourceInspector{
+		baseInspector: baseInspector{
+			catalog: catalog,
+		},
+	}
+}
+
+const javaSourceKameletEip = `
 from("direct:start")
-    .kamelet("foo/bar?baz=test")
+    .kamelet("foo/bar?baz=test");
 `
 
-const JavaSourceKameletEndpoint = `
+const javaSourceKameletEndpoint = `
 from("direct:start")
-    .to("kamelet:foo/bar?baz=test")
+    .to("kamelet:foo/bar?baz=test");
 `
 
-const JavaSourceWireTapEip = `
+const javaSourceWireTapEip = `
 from("direct:start")
-    .wireTap("kamelet:foo/bar?baz=test")
+    .wireTap("kamelet:foo/bar?baz=test");
 `
 
 func TestJavaSourceKamelet(t *testing.T) {
@@ -48,45 +61,94 @@ func TestJavaSourceKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   JavaSourceKameletEip,
+			source:   javaSourceKameletEip,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   JavaSourceKameletEndpoint,
+			source:   javaSourceKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   JavaSourceWireTapEip,
+			source:   javaSourceWireTapEip,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestJavaSourceInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestJavaSourceKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := JavaSourceInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
+		})
+	}
+}
+
+const javaSourceJSONEip = `
+from("direct:start")
+    .unmarshal().json();
+`
+
+const javaSourceJSONJacksonEip = `
+from("direct:start")
+    .unmarshal().json(JsonLibrary.Jackson);
+`
+
+const javaSourceAvroEndpoint = `
+from("direct:start")
+    .to("dataformat:avro:marshal");
+`
+
+const javaSourceJacksonEndpoint = `
+from("direct:start")
+    .to("dataformat:jackson:marshal");
+`
+
+const javaSourceProtobufEndpoint = `
+from("direct:start")
+    .to("dataformat:protobuf:marshal");
+`
+
+func TestJavaSourceDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: javaSourceJSONEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: javaSourceJSONJacksonEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: javaSourceAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: javaSourceJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: javaSourceProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestJavaSourceInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestJavaSourceDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/inspector_kotlin_test.go b/pkg/util/source/inspector_kotlin_test.go
index 1b08fe2b3..09619d426 100644
--- a/pkg/util/source/inspector_kotlin_test.go
+++ b/pkg/util/source/inspector_kotlin_test.go
@@ -21,23 +21,36 @@ import (
 	"fmt"
 	"testing"
 
-	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/camel"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-const KotlinKameletEip = `
+func newTestKotlinInspector(t *testing.T) KotlinInspector {
+	t.Helper()
+
+	catalog, err := camel.DefaultCatalog()
+	require.NoError(t, err)
+
+	return KotlinInspector{
+		baseInspector: baseInspector{
+			catalog: catalog,
+		},
+	}
+}
+
+const kotlinKameletEip = `
 from("direct:start")
     .kamelet("foo/bar?baz=test")
 `
 
-const KotlinKameletEndpoint = `
+const kotlinKameletEndpoint = `
 from("direct:start")
     .to("kamelet:foo/bar?baz=test")
 `
 
-const KotlinWireTapEip = `
+const kotlinWireTapEip = `
 from("direct:start")
     .wireTap("kamelet:foo/bar?baz=test")
 `
@@ -48,45 +61,94 @@ func TestKotlinKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   KotlinKameletEip,
+			source:   kotlinKameletEip,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   KotlinKameletEndpoint,
+			source:   kotlinKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   KotlinWireTapEip,
+			source:   kotlinWireTapEip,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestKotlinInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestKotlinKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := KotlinInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
+		})
+	}
+}
+
+const kotlinJSONEip = `
+from("direct:start")
+    .unmarshal().json()
+`
+
+const kotlinJSONJacksonEip = `
+from("direct:start")
+    .unmarshal().json(JsonLibrary.Jackson)
+`
+
+const kotlinAvroEndpoint = `
+from("direct:start")
+    .to("dataformat:avro:marshal")
+`
+
+const kotlinJacksonEndpoint = `
+from("direct:start")
+    .to("dataformat:jackson:marshal")
+`
+
+const kotlinProtobufEndpoint = `
+from("direct:start")
+    .to("dataformat:protobuf:marshal")
+`
+
+func TestKotlinDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: kotlinJSONEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: kotlinJSONJacksonEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: kotlinAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: kotlinJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: kotlinProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestKotlinInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestKotlinDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/inspector_xml_test.go b/pkg/util/source/inspector_xml_test.go
index 8552a84b7..9f9077fb9 100644
--- a/pkg/util/source/inspector_xml_test.go
+++ b/pkg/util/source/inspector_xml_test.go
@@ -21,13 +21,26 @@ import (
 	"fmt"
 	"testing"
 
-	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/camel"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-const XMLKameletEip = `
+func newTestXMLInspector(t *testing.T) XMLInspector {
+	t.Helper()
+
+	catalog, err := camel.DefaultCatalog()
+	require.NoError(t, err)
+
+	return XMLInspector{
+		baseInspector: baseInspector{
+			catalog: catalog,
+		},
+	}
+}
+
+const xmlKameletEip = `
 <camelContext xmlns="http://camel.apache.org/schema/spring">
   <route>
     <from uri="direct:start"/>
@@ -37,7 +50,7 @@ const XMLKameletEip = `
 </camelContext>
 `
 
-const XMLKameletEndpoint = `
+const xmlKameletEndpoint = `
 <camelContext xmlns="http://camel.apache.org/schema/spring">
   <route>
     <from uri="direct:start"/>
@@ -46,7 +59,7 @@ const XMLKameletEndpoint = `
 </camelContext>
 `
 
-const XMLWireTapEndpoint = `
+const xmlWireTapEndpoint = `
 <camelContext xmlns="http://camel.apache.org/schema/spring">
   <route>
     <from uri="direct:start"/>
@@ -61,45 +74,115 @@ func TestXMLKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   XMLKameletEip,
+			source:   xmlKameletEip,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   XMLKameletEndpoint,
+			source:   xmlKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   XMLWireTapEndpoint,
+			source:   xmlWireTapEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestXMLInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestXMLKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := XMLInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
+		})
+	}
+}
+
+const xmlJSONEip = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <marshal><json/></marshal>
+  </route>
+</camelContext>
+`
+
+const xmlJSONJacksonEip = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <marshal><json library="Jackson"/></marshal>
+  </route>
+</camelContext>
+`
+
+const xmlAvroEndpoint = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <to uri="dataformat:avro:marshal"/>
+  </route>
+</camelContext>
+`
+
+const xmlJacksonEndpoint = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <to uri="dataformat:jackson:marshal"/>
+  </route>
+</camelContext>
+`
+
+const xmlProtobufEndpoint = `
+<camelContext xmlns="http://camel.apache.org/schema/spring">
+  <route>
+    <from uri="direct:start"/>
+    <to uri="dataformat:protobuf:marshal"/>
+  </route>
+</camelContext>
+`
+
+func TestXMLDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: xmlJSONEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: xmlJSONJacksonEip,
+			deps:   []string{"camel:jackson"},
+		},
+		{
+			source: xmlAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: xmlJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: xmlProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestXMLInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestXMLDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/inspector_yaml_test.go b/pkg/util/source/inspector_yaml_test.go
index 212993939..28e1d261e 100644
--- a/pkg/util/source/inspector_yaml_test.go
+++ b/pkg/util/source/inspector_yaml_test.go
@@ -22,16 +22,17 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/util/camel"
 )
 
-func NewtestYAMLInspector(t *testing.T) YAMLInspector {
+func newTestYAMLInspector(t *testing.T) YAMLInspector {
 	t.Helper()
 
 	catalog, err := camel.DefaultCatalog()
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	return YAMLInspector{
 		baseInspector: baseInspector{
@@ -40,7 +41,7 @@ func NewtestYAMLInspector(t *testing.T) YAMLInspector {
 	}
 }
 
-const YAMLRouteConsumer = `
+const yamlRouteConsumer = `
 - from:
     uri: knative:endpoint/default
     steps:
@@ -48,7 +49,7 @@ const YAMLRouteConsumer = `
           uri: "log:out"
 `
 
-const YAMLRouteProducer = `
+const yamlRouteProducer = `
 - from:
     uri: timer:tick
     steps:
@@ -56,7 +57,7 @@ const YAMLRouteProducer = `
           uri: knative:endpoint/service
 `
 
-const YAMLRouteTransformer = `
+const yamlRouteTransformer = `
 - from:
     uri: knative:channel/mychannel
     steps:
@@ -64,14 +65,14 @@ const YAMLRouteTransformer = `
           uri: knative:endpoint/service
 `
 
-const YAMLInvalid = `
+const yamlInvalid = `
 - from:
     uri: knative:endpoint/default
     steps:
       - "log:out"
 `
 
-const YAMLInDepthChannel = `
+const yamlInDepthChannel = `
 - from:
     uri: knative:channel/mychannel
     steps:
@@ -83,7 +84,7 @@ const YAMLInDepthChannel = `
                 uri: knative:endpoint/service
 `
 
-const YAMLWireTapKnativeEIP = `
+const yamlWireTapKnativeEIP = `
 - from:
     uri: knative:channel/mychannel
     parameters:
@@ -93,7 +94,7 @@ const YAMLWireTapKnativeEIP = `
           uri: knative:channel/mychannel
 `
 
-const YAMLWireTapJmsEIP = `
+const yamlWireTapJmsEIP = `
 - from:
     uri: knative:channel/mychannel
     parameters:
@@ -112,78 +113,69 @@ func TestYAMLDependencies(t *testing.T) {
 	}{
 		{
 			name:         "consumer",
-			source:       YAMLRouteConsumer,
+			source:       yamlRouteConsumer,
 			dependencies: []string{`camel:knative`},
 		},
 		{
 			name:         "producer",
-			source:       YAMLRouteProducer,
+			source:       yamlRouteProducer,
 			dependencies: []string{`camel:knative`},
 		},
 		{
 			name:   "transformer",
-			source: YAMLRouteTransformer,
+			source: yamlRouteTransformer,
 			dependencies: []string{
 				`camel:knative`,
 			},
 		},
 		{
 			name:   "invalid",
-			source: YAMLInvalid,
+			source: yamlInvalid,
 			dependencies: []string{
 				`camel:knative`,
 			},
 		},
 		{
 			name:   "in-depth",
-			source: YAMLInDepthChannel,
+			source: yamlInDepthChannel,
 			dependencies: []string{
 				`camel:knative`,
 			},
 		},
 		{
 			name:   "wire-tap-knative",
-			source: YAMLWireTapKnativeEIP,
+			source: yamlWireTapKnativeEIP,
 			dependencies: []string{
 				`camel:knative`,
 			},
 		},
 		{
 			name:   "wire-tap-jms",
-			source: YAMLWireTapJmsEIP,
+			source: yamlWireTapJmsEIP,
 			dependencies: []string{
 				`camel:knative`,
 				`camel:jms`,
 			},
 		},
 	}
+
+	inspector := newTestYAMLInspector(t)
 	for i := range tests {
 		test := tests[i]
 		t.Run(test.name, func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Name:    "route.yaml",
-					Content: test.source,
-				},
-				Language: v1.LanguageYaml,
-			}
-
-			meta := NewMetadata()
-			inspector := NewtestYAMLInspector(t)
-
-			err := inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			for _, dependency := range test.dependencies {
-				assert.Contains(t, meta.Dependencies.List(), dependency)
-			}
-			for _, missingDependency := range test.missingDependencies {
-				assert.NotContains(t, meta.Dependencies.List(), missingDependency)
-			}
+			assertExtractYAML(t, inspector, test.source, func(meta *Metadata) {
+				for _, dependency := range test.dependencies {
+					assert.Contains(t, meta.Dependencies.List(), dependency)
+				}
+				for _, missingDependency := range test.missingDependencies {
+					assert.NotContains(t, meta.Dependencies.List(), missingDependency)
+				}
+			})
 		})
 	}
 }
 
-const YAMLRestDSL = `
+const yamlRestDSL = `
 - rest:
     verb: "post"
     uri: "/api/flow"
@@ -204,7 +196,7 @@ const YAMLRestDSL = `
           uri: "log:out"
 `
 
-const YAMLRestDSLWithRoute = `
+const yamlRestDSLWithRoute = `
 - route:
     id: "flow"
     group: "routes"
@@ -229,30 +221,23 @@ const YAMLRestDSLWithRoute = `
 `
 
 func TestYAMLRestDSL(t *testing.T) {
-	for name, content := range map[string]string{"YAMLRestDSL": YAMLRestDSL, "YAMLRestDSLWithRoute": YAMLRestDSLWithRoute} {
+	inspector := newTestYAMLInspector(t)
+	for name, content := range map[string]string{
+		"yamlRestDSL":          yamlRestDSL,
+		"yamlRestDSLWithRoute": yamlRestDSLWithRoute,
+	} {
 		sourceContent := content
 		t.Run(name, func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Name:    "route.yaml",
-					Content: sourceContent,
-				},
-				Language: v1.LanguageYaml,
-			}
-
-			meta := NewMetadata()
-			inspector := NewtestYAMLInspector(t)
-
-			err := inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.Has(v1.CapabilityRest))
-			assert.True(t, meta.Dependencies.Has("camel:log"))
-			assert.True(t, meta.ExposesHTTPServices)
+			assertExtractYAML(t, inspector, sourceContent, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.Has(v1.CapabilityRest))
+				assert.True(t, meta.Dependencies.Has("camel:log"))
+				assert.True(t, meta.ExposesHTTPServices)
+			})
 		})
 	}
 }
 
-const YAMLFromDSL = `
+const yamlFromDSL = `
 - from:
     uri: "timer:tick"
     parameters:
@@ -265,7 +250,7 @@ const YAMLFromDSL = `
       - to: "log:info"
 `
 
-const YAMLFromDSLWithRoute = `
+const yamlFromDSLWithRoute = `
 - route:
     id: route1
     from:
@@ -281,30 +266,24 @@ const YAMLFromDSLWithRoute = `
 `
 
 func TestYAMLRouteAndFromEquivalence(t *testing.T) {
-	for name, content := range map[string]string{"YAMLFromDSL": YAMLFromDSL, "YAMLFromDSLWithRoute": YAMLFromDSLWithRoute} {
+	inspector := newTestYAMLInspector(t)
+	for name, content := range map[string]string{
+		"yamlFromDSL":          yamlFromDSL,
+		"yamlFromDSLWithRoute": yamlFromDSLWithRoute,
+	} {
 		sourceContent := content
 		t.Run(name, func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Name:    "route.yaml",
-					Content: sourceContent,
-				},
-				Language: v1.LanguageYaml,
-			}
-
-			meta := NewMetadata()
-			inspector := NewtestYAMLInspector(t)
-			err := inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.Equal(t, meta.FromURIs, []string{"timer:tick?period=5000"})
-			assert.Equal(t, meta.ToURIs, []string{"log:info"})
-			assert.True(t, meta.Dependencies.Has("camel:log"))
-			assert.True(t, meta.Dependencies.Has("camel:timer"))
+			assertExtractYAML(t, inspector, sourceContent, func(meta *Metadata) {
+				assert.Equal(t, meta.FromURIs, []string{"timer:tick?period=5000"})
+				assert.Equal(t, meta.ToURIs, []string{"log:info"})
+				assert.True(t, meta.Dependencies.Has("camel:log"))
+				assert.True(t, meta.Dependencies.Has("camel:timer"))
+			})
 		})
 	}
 }
 
-const YAMLJSONMarshal = `
+const yamlJSONMarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -312,7 +291,7 @@ const YAMLJSONMarshal = `
         json: {}
 `
 
-const YAMLJSONUnmarshal = `
+const yamlJSONUnmarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -320,7 +299,7 @@ const YAMLJSONUnmarshal = `
         json: {}
 `
 
-const YAMLJSONGsonMarshal = `
+const yamlJSONGsonMarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -329,7 +308,7 @@ const YAMLJSONGsonMarshal = `
           library: Gson
 `
 
-const YAMLJSONUnknownMarshal = `
+const yamlJSONUnknownMarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -344,60 +323,106 @@ func TestYAMLJson(t *testing.T) {
 		dependency string
 	}{
 		{
-			source:     YAMLJSONMarshal,
+			source:     yamlJSONMarshal,
 			dependency: "camel:jackson",
 		},
 		{
-			source:     YAMLJSONUnmarshal,
+			source:     yamlJSONUnmarshal,
 			dependency: "camel:jackson",
 		},
 		{
-			source:     YAMLJSONGsonMarshal,
+			source:     yamlJSONGsonMarshal,
 			dependency: "camel:gson",
 		},
 		{
-			source:     YAMLJSONUnknownMarshal,
+			source:     yamlJSONUnknownMarshal,
 			dependency: "camel:timer",
 		},
 	}
 
+	inspector := newTestYAMLInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("%s-%d", test.dependency, i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Name:    "route.yaml",
-					Content: test.source,
-				},
-				Language: v1.LanguageYaml,
-			}
-
-			meta := NewMetadata()
-			inspector := NewtestYAMLInspector(t)
-
-			err := inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-			assert.Contains(t, meta.Dependencies.List(), test.dependency)
+			assertExtractYAML(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				assert.Contains(t, meta.Dependencies.List(), test.dependency)
+			})
+		})
+	}
+}
+
+const yamlAvroEndpoint = `
+- from:
+    uri: direct:start
+    steps:
+    - to:
+        uri: dataformat:avro:marshal
+`
+
+const yamlJacksonEndpoint = `
+- from:
+    uri: direct:start
+    steps:
+    - to:
+        uri: dataformat:jackson:marshal
+`
+
+const yamlProtobufEndpoint = `
+- from:
+    uri: direct:start
+    steps:
+    - to:
+        uri: dataformat:protobuf:marshal
+`
+
+func TestYAMLDataFormat(t *testing.T) {
+	tc := []struct {
+		source string
+		deps   []string
+	}{
+		{
+			source: yamlAvroEndpoint,
+			deps:   []string{"camel:dataformat", "camel:avro"},
+		},
+		{
+			source: yamlJacksonEndpoint,
+			deps:   []string{"camel:dataformat", "camel:jackson"},
+		},
+		{
+			source: yamlProtobufEndpoint,
+			deps:   []string{"camel:dataformat", "camel:protobuf"},
+		},
+	}
+
+	inspector := newTestYAMLInspector(t)
+	for i := range tc {
+		test := tc[i]
+		t.Run(fmt.Sprintf("TestYAMLDataFormat-%d", i), func(t *testing.T) {
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				for _, d := range test.deps {
+					assert.Contains(t, meta.Dependencies.List(), d)
+				}
+			})
 		})
 	}
 }
 
-const YAMLKameletEipNoID = `
+const yamlKameletEipNoID = `
 - from:
     uri: timer:tick
     steps:
     - kamelet: "foo"
 `
 
-const YAMLKameletEipInline = `
+const yamlKameletEipInline = `
 - from:
     uri: timer:tick
     steps:
     - kamelet: "foo/bar?baz=test"
 `
 
-const YAMLKameletEipMap = `
+const yamlKameletEipMap = `
 - from:
     uri: timer:tick
     steps:
@@ -406,7 +431,7 @@ const YAMLKameletEipMap = `
 `
 
 // #nosec G101
-const YAMLKameletEipMapWithParams = `
+const yamlKameletEipMapWithParams = `
 - from:
     uri: timer:tick
     steps:
@@ -416,7 +441,7 @@ const YAMLKameletEipMapWithParams = `
           baz:test
 `
 
-const YAMLKameletEndpoint = `
+const yamlKameletEndpoint = `
 - from:
     uri: timer:tick
     steps:
@@ -429,58 +454,42 @@ func TestYAMLKamelet(t *testing.T) {
 		kamelets []string
 	}{
 		{
-			source:   YAMLKameletEipNoID,
+			source:   yamlKameletEipNoID,
 			kamelets: []string{"foo"},
 		},
 		{
-			source:   YAMLKameletEipInline,
+			source:   yamlKameletEipInline,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   YAMLKameletEipMap,
+			source:   yamlKameletEipMap,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   YAMLKameletEipMapWithParams,
+			source:   yamlKameletEipMapWithParams,
 			kamelets: []string{"foo/bar"},
 		},
 		{
-			source:   YAMLKameletEndpoint,
+			source:   yamlKameletEndpoint,
 			kamelets: []string{"foo/bar"},
 		},
 	}
 
+	inspector := newTestYAMLInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestYAMLKamelet-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := YAMLInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			for _, k := range test.kamelets {
-				assert.Contains(t, meta.Kamelets, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				for _, k := range test.kamelets {
+					assert.Contains(t, meta.Kamelets, k)
+				}
+			})
 		})
 	}
 }
 
-const YAMLKameletExplicitParams = `
+const yamlKameletExplicitParams = `
 - from:
     uri: cron:tab
     parameters:
@@ -492,7 +501,7 @@ const YAMLKameletExplicitParams = `
           cloudEventsSpecVersion: "1.0"
 `
 
-const YAMLKameletExplicitNumericParams = `
+const yamlKameletExplicitNumericParams = `
 - from:
     uri: timer:tick
     parameters:
@@ -508,47 +517,31 @@ func TestYAMLExplicitParameters(t *testing.T) {
 		toURIs   []string
 	}{
 		{
-			source:   YAMLKameletExplicitParams,
+			source:   yamlKameletExplicitParams,
 			fromURIs: []string{"cron:tab?schedule=%2A+%2A+%2A+%2A+%3F"},
 			toURIs:   []string{"knative:channel/a?cloudEventsSpecVersion=1.0"},
 		},
 		{
-			source:   YAMLKameletExplicitNumericParams,
+			source:   yamlKameletExplicitNumericParams,
 			fromURIs: []string{"timer:tick?period=1000"},
 		},
 	}
 
+	inspector := newTestYAMLInspector(t)
 	for i := range tc {
 		test := tc[i]
 		t.Run(fmt.Sprintf("TestYAMLExplicitParameters-%d", i), func(t *testing.T) {
-			code := v1.SourceSpec{
-				DataSpec: v1.DataSpec{
-					Content: test.source,
-				},
-			}
-
-			catalog, err := camel.DefaultCatalog()
-			assert.Nil(t, err)
-
-			meta := NewMetadata()
-			inspector := YAMLInspector{
-				baseInspector: baseInspector{
-					catalog: catalog,
-				},
-			}
-
-			err = inspector.Extract(code, &meta)
-			assert.Nil(t, err)
-			assert.True(t, meta.RequiredCapabilities.IsEmpty())
-
-			assert.Len(t, meta.FromURIs, len(test.fromURIs))
-			for _, k := range test.fromURIs {
-				assert.Contains(t, meta.FromURIs, k)
-			}
-			assert.Len(t, meta.ToURIs, len(test.toURIs))
-			for _, k := range test.toURIs {
-				assert.Contains(t, meta.ToURIs, k)
-			}
+			assertExtract(t, inspector, test.source, func(meta *Metadata) {
+				assert.True(t, meta.RequiredCapabilities.IsEmpty())
+				assert.Len(t, meta.FromURIs, len(test.fromURIs))
+				for _, k := range test.fromURIs {
+					assert.Contains(t, meta.FromURIs, k)
+				}
+				assert.Len(t, meta.ToURIs, len(test.toURIs))
+				for _, k := range test.toURIs {
+					assert.Contains(t, meta.ToURIs, k)
+				}
+			})
 		})
 	}
 }
diff --git a/pkg/util/source/test_support.go b/pkg/util/source/test_support.go
new file mode 100644
index 000000000..7d1863a77
--- /dev/null
+++ b/pkg/util/source/test_support.go
@@ -0,0 +1,58 @@
+/*
+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 source
+
+import (
+	"testing"
+
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
+
+	"github.com/stretchr/testify/require"
+)
+
+func assertExtract(t *testing.T, inspector Inspector, source string, assertFn func(meta *Metadata)) {
+	t.Helper()
+
+	srcSpec := v1.SourceSpec{
+		DataSpec: v1.DataSpec{
+			Content: source,
+		},
+	}
+	meta := NewMetadata()
+	err := inspector.Extract(srcSpec, &meta)
+	require.NoError(t, err)
+
+	assertFn(&meta)
+}
+
+func assertExtractYAML(t *testing.T, inspector YAMLInspector, source string, assertFn func(meta *Metadata)) {
+	t.Helper()
+
+	srcSpec := v1.SourceSpec{
+		DataSpec: v1.DataSpec{
+			Name:    "route.yaml",
+			Content: source,
+		},
+		Language: v1.LanguageYaml,
+	}
+	meta := NewMetadata()
+	err := inspector.Extract(srcSpec, &meta)
+	require.NoError(t, err)
+
+	assertFn(&meta)
+}