Repository: apache/camel-k
Pull request #3857 (2022-11-29 08:43:14)

fix(source): dataformat dependencies not resolved (1.10.x)
<!-- Description -->

Backport #3856 to 1.10.x


<!--
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
fix(source): dataformat dependencies not resolved
```



Patch:
diff --git a/pkg/apis/camel/v1/camelcatalog_types_support.go b/pkg/apis/camel/v1/camelcatalog_types_support.go
index 9c0ba615b..89c4c16eb 100644
--- a/pkg/apis/camel/v1/camelcatalog_types_support.go
+++ b/pkg/apis/camel/v1/camelcatalog_types_support.go
@@ -74,20 +74,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/util/camel/camel_runtime_catalog.go b/pkg/util/camel/camel_runtime_catalog.go
index 511541911..28e52e4f6 100644
--- a/pkg/util/camel/camel_runtime_catalog.go
+++ b/pkg/util/camel/camel_runtime_catalog.go
@@ -159,9 +159,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 be2b86ddc..783662b0d 100644
--- a/pkg/util/source/inspector.go
+++ b/pkg/util/source/inspector.go
@@ -248,6 +248,8 @@ func (i *baseInspector) discoverDependencies(source v1.SourceSpec, meta *Metadat
 				for _, dep := range candidateComp.GetConsumerDependencyIDs(scheme.ID) {
 					i.addDependency(dep, meta)
 				}
+				// some components require additional dependency resolution based on URI
+				i.addDependenciesFromURI(uri, scheme, meta)
 			}
 		}
 	}
@@ -260,6 +262,8 @@ func (i *baseInspector) discoverDependencies(source v1.SourceSpec, meta *Metadat
 				for _, dep := range candidateComp.GetProducerDependencyIDs(scheme.ID) {
 					i.addDependency(dep, meta)
 				}
+				// some components require additional dependency resolution based on URI
+				i.addDependenciesFromURI(uri, scheme, meta)
 			}
 		}
 	}
@@ -306,6 +310,22 @@ func (i *baseInspector) addDependency(dependency string, meta *Metadata) {
 	meta.Dependencies.Add(dependency)
 }
 
+func (i *baseInspector) addDependenciesFromURI(uri string, scheme *v1.CamelScheme, meta *Metadata) {
+	if scheme.ID == "dataformat" {
+		// dataformat:name:(marshal|unmarshal)[?options]
+		parts := strings.Split(uri, ":")
+		if len(parts) < 3 {
+			return
+		}
+		name := parts[1]
+		df := i.catalog.GetArtifactByDataFormat(name)
+		if df == nil {
+			return
+		}
+		meta.Dependencies.Add(df.GetDependencyID())
+	}
+}
+
 // hasOnlyPassiveEndpoints returns true if the source has no endpoint that needs to remain always active.
 func (i *baseInspector) hasOnlyPassiveEndpoints(fromURIs []string) bool {
 	passivePlusHTTP := make(map[string]bool)
diff --git a/pkg/util/source/inspector_xml.go b/pkg/util/source/inspector_xml.go
index 51baaee24..a854bebf4 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.Dependencies.Add(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 a28ac260e..df3fb077d 100644
--- a/pkg/util/source/inspector_yaml.go
+++ b/pkg/util/source/inspector_yaml.go
@@ -65,9 +65,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/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 1a85b1fd5..8bf25330a 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,19 +113,19 @@ func TestYAMLDependencies(t *testing.T) {
 	}{
 		{
 			name:                "consumer",
-			source:              YAMLRouteConsumer,
+			source:              yamlRouteConsumer,
 			dependencies:        []string{`mvn:org.apache.camel.k:camel-k-knative-consumer`},
 			missingDependencies: []string{`mvn:org.apache.camel.k:camel-k-knative-producer`},
 		},
 		{
 			name:                "producer",
-			source:              YAMLRouteProducer,
+			source:              yamlRouteProducer,
 			dependencies:        []string{`mvn:org.apache.camel.k:camel-k-knative-producer`},
 			missingDependencies: []string{`mvn:org.apache.camel.k:camel-k-knative-consumer`},
 		},
 		{
 			name:   "transformer",
-			source: YAMLRouteTransformer,
+			source: yamlRouteTransformer,
 			dependencies: []string{
 				`mvn:org.apache.camel.k:camel-k-knative-producer`,
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
@@ -132,14 +133,14 @@ func TestYAMLDependencies(t *testing.T) {
 		},
 		{
 			name:   "invalid",
-			source: YAMLInvalid,
+			source: yamlInvalid,
 			dependencies: []string{
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
 			},
 		},
 		{
 			name:   "in-depth",
-			source: YAMLInDepthChannel,
+			source: yamlInDepthChannel,
 			dependencies: []string{
 				`mvn:org.apache.camel.k:camel-k-knative-producer`,
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
@@ -147,7 +148,7 @@ func TestYAMLDependencies(t *testing.T) {
 		},
 		{
 			name:   "wire-tap-knative",
-			source: YAMLWireTapKnativeEIP,
+			source: yamlWireTapKnativeEIP,
 			dependencies: []string{
 				`mvn:org.apache.camel.k:camel-k-knative-producer`,
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
@@ -155,40 +156,31 @@ func TestYAMLDependencies(t *testing.T) {
 		},
 		{
 			name:   "wire-tap-jms",
-			source: YAMLWireTapJmsEIP,
+			source: yamlWireTapJmsEIP,
 			dependencies: []string{
 				`mvn:org.apache.camel.k:camel-k-knative-consumer`,
 				`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"
@@ -209,7 +201,7 @@ const YAMLRestDSL = `
           uri: "log:out"
 `
 
-const YAMLRestDSLWithRoute = `
+const yamlRestDSLWithRoute = `
 - route:
     id: "flow"
     group: "routes"
@@ -234,30 +226,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:
@@ -270,7 +255,7 @@ const YAMLFromDSL = `
       - to: "log:info"
 `
 
-const YAMLFromDSLWithRoute = `
+const yamlFromDSLWithRoute = `
 - route:
     id: route1
     from:
@@ -286,30 +271,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:
@@ -317,7 +296,7 @@ const YAMLJSONMarshal = `
         json: {}
 `
 
-const YAMLJSONUnmarshal = `
+const yamlJSONUnmarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -325,7 +304,7 @@ const YAMLJSONUnmarshal = `
         json: {}
 `
 
-const YAMLJSONGsonMarshal = `
+const yamlJSONGsonMarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -334,7 +313,7 @@ const YAMLJSONGsonMarshal = `
           library: Gson
 `
 
-const YAMLJSONUnknownMarshal = `
+const yamlJSONUnknownMarshal = `
 - from:
     uri: timer:tick
     steps:
@@ -349,60 +328,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:
@@ -411,7 +436,7 @@ const YAMLKameletEipMap = `
 `
 
 // #nosec G101
-const YAMLKameletEipMapWithParams = `
+const yamlKameletEipMapWithParams = `
 - from:
     uri: timer:tick
     steps:
@@ -421,7 +446,7 @@ const YAMLKameletEipMapWithParams = `
           baz:test
 `
 
-const YAMLKameletEndpoint = `
+const yamlKameletEndpoint = `
 - from:
     uri: timer:tick
     steps:
@@ -434,58 +459,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:
@@ -497,7 +506,7 @@ const YAMLKameletExplicitParams = `
           cloudEventsSpecVersion: "1.0"
 `
 
-const YAMLKameletExplicitNumericParams = `
+const yamlKameletExplicitNumericParams = `
 - from:
     uri: timer:tick
     parameters:
@@ -513,47 +522,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)
+}