test: add missing tests for Entity+Edge model migration (#513)

* test: add missing tests for Entity+Edge model migration

Restore and strengthen test coverage that was lost during the
Asset → Entity+Edge migration. Adds new test files for previously
untested components and extends existing test files with edge-case
coverage.

New test files:
- processors/enrich: Init validation, Process with nil/existing props, edge passthrough
- processors/labels: Init validation, Process with nil/existing labels, merge, edge passthrough
- sinks/console: Init, Sink single/multiple/empty/with-edges, Close
- sinks/kafka: Init config validation (missing brokers, topic, etc.)

Strengthened tests:
- models/record: multiple edges, nil entity
- models/util: nil/empty props, nested props sanitization, JSON with/without edges
- processors/script: modify entity name, nil properties, edge preservation
- sinks/compass: empty batch handling
- sinks/file: empty batch, records with edges (ndjson + yaml)

* docs: update context_graph.md to reflect Entity+Edge model

Replace outdated Asset terminology with the current Record/Entity/Edge
model that was introduced in the migration.

* fix: update Makefile proto generation to use meteor/v1beta1

Update PROTON_COMMIT to the commit that added the new Entity+Edge
protos and removed the legacy assets. Update the --path flag from
the deleted raystack/assets/v1beta2 to raystack/meteor/v1beta1.

Author: ravisuhag

Makefile

@@ -1,7 +1,7 @@
 NAME="github.com/raystack/meteor"
 VERSION=$(shell git describe --always --tags 2>/dev/null)
 COVERFILE="/tmp/app.coverprofile"
-PROTON_COMMIT := "ae895e033f71df187c62d7cf9431a2e259ddd423"
+PROTON_COMMIT := "f5514e23005e7480319a18ba905dfecaa17379f8"
 .PHONY: all build clean test
 
 all: build
@@ -36,7 +36,7 @@ test-coverage:
 generate-proto:
 	@echo " > cloning protobuf from raystack/proton"
 	@echo " > generating protobuf"
-	@buf generate --template buf.gen.yaml https://github.com/raystack/proton/archive/${PROTON_COMMIT}.zip#strip_components=1 --path raystack/assets/v1beta2
+	@buf generate --template buf.gen.yaml https://github.com/raystack/proton/archive/${PROTON_COMMIT}.zip#strip_components=1 --path raystack/meteor/v1beta1
 	@echo " > protobuf compilation finished"
 
 lint:

docs/docs/concepts/context_graph.md

@@ -19,11 +19,11 @@ Meteor operates as a metadata supply chain with three stages:
 
 ### Extract
 
-Meteor's extractors connect to 30+ data sources — databases (BigQuery, Postgres, Snowflake), BI tools (Tableau, Metabase, Superset), streaming platforms (Kafka), cloud storage (GCS), orchestrators (Optimus), and more. Each extractor produces standardized **Asset** records containing:
+Meteor's extractors connect to 30+ data sources — databases (BigQuery, Postgres, Snowflake), BI tools (Tableau, Metabase, Superset), streaming platforms (Kafka), cloud storage (GCS), orchestrators (Optimus), and more. Each extractor produces **Record**s — an **Entity** with flat properties plus **Edge**s representing relationships:
 
-- **Schema metadata** — column names, types, descriptions, constraints
-- **Lineage** — upstream and downstream asset references
-- **Ownership** — who owns and maintains the asset
+- **Schema metadata** — column names, types, descriptions, constraints (stored as entity properties)
+- **Lineage** — upstream and downstream relationships (represented as edges)
+- **Ownership** — who owns and maintains the entity (represented as edges)
 - **Service context** — source system, URLs, timestamps
 
 ### Process

models/record_test.go

@@ -34,3 +34,36 @@ func TestNewRecordWithEdges(t *testing.T) {
 	assert.Len(t, record.Edges(), 1)
 	assert.Equal(t, edge, record.Edges()[0])
 }
+
+func TestNewRecordWithMultipleEdges(t *testing.T) {
+	entity := &meteorv1beta1.Entity{
+		Urn:  "urn:test:scope:table:t1",
+		Name: "t1",
+		Type: "table",
+	}
+	lineageEdge := &meteorv1beta1.Edge{
+		SourceUrn: "urn:test:scope:table:t1",
+		TargetUrn: "urn:test:scope:table:t2",
+		Type:      "lineage",
+		Source:    "test",
+	}
+	ownerEdge := &meteorv1beta1.Edge{
+		SourceUrn: "urn:test:scope:table:t1",
+		TargetUrn: "urn:user:alice@co.com",
+		Type:      "owned_by",
+		Source:    "test",
+	}
+	record := models.NewRecord(entity, lineageEdge, ownerEdge)
+	assert.Equal(t, entity, record.Entity())
+	assert.Len(t, record.Edges(), 2)
+	assert.Equal(t, lineageEdge, record.Edges()[0])
+	assert.Equal(t, ownerEdge, record.Edges()[1])
+	assert.Equal(t, "lineage", record.Edges()[0].GetType())
+	assert.Equal(t, "owned_by", record.Edges()[1].GetType())
+}
+
+func TestNewRecordWithNilEntity(t *testing.T) {
+	record := models.NewRecord(nil)
+	assert.Nil(t, record.Entity())
+	assert.Empty(t, record.Edges())
+}

models/util_test.go

@@ -88,3 +88,66 @@ func TestRecordToJSON(t *testing.T) {
 	assert.Contains(t, string(b), `"entity"`)
 	assert.Contains(t, string(b), `"edges"`)
 }
+
+func TestNewEntityWithNilProps(t *testing.T) {
+	entity := models.NewEntity("urn:test:s:table:t1", "table", "t1", "test", nil)
+	assert.Equal(t, "urn:test:s:table:t1", entity.GetUrn())
+	assert.Equal(t, "table", entity.GetType())
+	assert.Nil(t, entity.GetProperties())
+}
+
+func TestNewEntityWithEmptyProps(t *testing.T) {
+	entity := models.NewEntity("urn:test:s:table:t1", "table", "t1", "test", map[string]any{})
+	assert.Equal(t, "urn:test:s:table:t1", entity.GetUrn())
+	assert.Nil(t, entity.GetProperties())
+}
+
+func TestNewEntityWithNestedProps(t *testing.T) {
+	entity := models.NewEntity("urn:test:s:table:t1", "table", "t1", "test", map[string]any{
+		"labels": map[string]string{"env": "production", "team": "data"},
+		"tags":   []string{"important", "verified"},
+	})
+	assert.Equal(t, "urn:test:s:table:t1", entity.GetUrn())
+	props := entity.GetProperties()
+	require.NotNil(t, props)
+	labelsVal := props.GetFields()["labels"].GetStructValue()
+	require.NotNil(t, labelsVal)
+	assert.Equal(t, "production", labelsVal.GetFields()["env"].GetStringValue())
+	assert.Equal(t, "data", labelsVal.GetFields()["team"].GetStringValue())
+	tagsVal := props.GetFields()["tags"].GetListValue()
+	require.NotNil(t, tagsVal)
+	assert.Len(t, tagsVal.GetValues(), 2)
+	assert.Equal(t, "important", tagsVal.GetValues()[0].GetStringValue())
+}
+
+func TestRecordToJSONWithoutEdges(t *testing.T) {
+	entity := &meteorv1beta1.Entity{
+		Urn:  "urn:test:s:table:t1",
+		Name: "t1",
+	}
+	record := models.NewRecord(entity)
+
+	b, err := models.RecordToJSON(record)
+	require.NoError(t, err)
+	assert.Contains(t, string(b), `"entity"`)
+	assert.NotContains(t, string(b), `"edges"`)
+}
+
+func TestRecordToJSONWithMultipleEdges(t *testing.T) {
+	entity := &meteorv1beta1.Entity{
+		Urn:  "urn:test:s:table:t1",
+		Name: "t1",
+	}
+	lineage := models.LineageEdge("urn:a", "urn:b", "test")
+	owner := models.OwnerEdge("urn:test:s:table:t1", "urn:user:bob@co.com", "test")
+	record := models.NewRecord(entity, lineage, owner)
+
+	b, err := models.RecordToJSON(record)
+	require.NoError(t, err)
+	s := string(b)
+	assert.Contains(t, s, `"entity"`)
+	assert.Contains(t, s, `"edges"`)
+	assert.Contains(t, s, `"lineage"`)
+	assert.Contains(t, s, `"owned_by"`)
+	assert.Contains(t, s, `"urn:user:bob@co.com"`)
+}

plugins/processors/enrich/processor_test.go

@@ -0,0 +1,135 @@
+//go:build plugins
+
+package enrich_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/raystack/meteor/models"
+	"github.com/raystack/meteor/plugins"
+	"github.com/raystack/meteor/plugins/processors/enrich"
+	testutils "github.com/raystack/meteor/test/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInit(t *testing.T) {
+	t.Run("should return error for invalid config", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{},
+		})
+		assert.Error(t, err)
+	})
+
+	t.Run("should return no error for valid config", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{
+				"attributes": map[string]any{
+					"team": "data-engineering",
+				},
+			},
+		})
+		assert.NoError(t, err)
+	})
+}
+
+func TestProcess(t *testing.T) {
+	t.Run("should enrich entity with nil properties", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{
+				"attributes": map[string]any{
+					"team": "data-engineering",
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		entity := models.NewEntity("urn:table:1", "table", "my-table", "bigquery", nil)
+		rec := models.NewRecord(entity)
+
+		result, err := proc.Process(context.Background(), rec)
+		require.NoError(t, err)
+
+		props := result.Entity().GetProperties().AsMap()
+		assert.Equal(t, "data-engineering", props["team"])
+	})
+
+	t.Run("should merge with existing properties", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{
+				"attributes": map[string]any{
+					"team": "data-engineering",
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		entity := models.NewEntity("urn:table:1", "table", "my-table", "bigquery", map[string]any{
+			"existing_key": "existing_value",
+		})
+		rec := models.NewRecord(entity)
+
+		result, err := proc.Process(context.Background(), rec)
+		require.NoError(t, err)
+
+		props := result.Entity().GetProperties().AsMap()
+		assert.Equal(t, "existing_value", props["existing_key"])
+		assert.Equal(t, "data-engineering", props["team"])
+	})
+
+	t.Run("should preserve edges through processing", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{
+				"attributes": map[string]any{
+					"team": "data-engineering",
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		entity := models.NewEntity("urn:table:1", "table", "my-table", "bigquery", nil)
+		lineage := models.LineageEdge("urn:table:1", "urn:table:2", "bigquery")
+		owner := models.OwnerEdge("urn:table:1", "urn:user:alice", "bigquery")
+		rec := models.NewRecord(entity, lineage, owner)
+
+		result, err := proc.Process(context.Background(), rec)
+		require.NoError(t, err)
+
+		assert.Len(t, result.Edges(), 2)
+		assert.Equal(t, lineage, result.Edges()[0])
+		assert.Equal(t, owner, result.Edges()[1])
+	})
+
+	t.Run("should only add string values and skip non-string values", func(t *testing.T) {
+		proc := enrich.New(testutils.Logger)
+		err := proc.Init(context.Background(), plugins.Config{
+			RawConfig: map[string]any{
+				"attributes": map[string]any{
+					"team":     "data-engineering",
+					"count":    42,
+					"enabled":  true,
+					"fraction": 3.14,
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		entity := models.NewEntity("urn:table:1", "table", "my-table", "bigquery", nil)
+		rec := models.NewRecord(entity)
+
+		result, err := proc.Process(context.Background(), rec)
+		require.NoError(t, err)
+
+		props := result.Entity().GetProperties().AsMap()
+		assert.Equal(t, "data-engineering", props["team"])
+		assert.Nil(t, props["count"])
+		assert.Nil(t, props["enabled"])
+		assert.Nil(t, props["fraction"])
+	})
+}