pipeline_test.go

package dyfields_test

import (
	"encoding/json"
	"os"
	"testing"

	"github.com/BrobridgeOrg/dyfields"
)

// The pipeline scenario is the hard case the design was stress-tested against:
// two levels of object_list, a cross-level `^.` reference, secrets at both
// levels, and an edit that deletes an entry from the middle of a list.

func readFile(t *testing.T, name string) []byte {
	t.Helper()
	b, err := os.ReadFile("testdata/" + name)
	if err != nil {
		t.Fatalf("read %s: %v", name, err)
	}
	return b
}

func readDoc(t *testing.T, name string) dyfields.ValueDocument {
	t.Helper()
	var d dyfields.ValueDocument
	if err := json.Unmarshal(readFile(t, name), &d); err != nil {
		t.Fatalf("unmarshal %s: %v", name, err)
	}
	return d
}

func canonical(t *testing.T, v any) string {
	t.Helper()
	b, err := json.MarshalIndent(v, "", "  ")
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	return string(b)
}

func pipelineCompiled(t *testing.T) *dyfields.Compiled {
	t.Helper()
	c, err := dyfields.Load(readFile(t, "pipeline_schema.json"))
	if err != nil {
		t.Fatalf("load pipeline schema: %v", err)
	}
	return c
}

func TestPipelineSchemaCompiles(t *testing.T) {
	c := pipelineCompiled(t)

	// Both list levels hold a secret somewhere below them, so both need a
	// client-generated $id: the secret key cannot be written without one.
	if !c.RequiresItemID("sinks") {
		t.Error("sinks should require a client-generated $id")
	}
	if !c.RequiresItemID("sinks.mappings") {
		t.Error("sinks.mappings should require a client-generated $id")
	}
}

func TestPipelineSchemaRoundTrip(t *testing.T) {
	original := readFile(t, "pipeline_schema.json")
	s, err := dyfields.Parse(original)
	if err != nil {
		t.Fatalf("parse: %v", err)
	}
	out, err := s.MarshalIndent()
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	again, err := dyfields.Parse(out)
	if err != nil {
		t.Fatalf("re-parse: %v", err)
	}
	if canonical(t, s) != canonical(t, again) {
		t.Fatal("schema changed when written out and read back")
	}
}

func TestPipelineExistingDocumentValidates(t *testing.T) {
	c := pipelineCompiled(t)
	before := readDoc(t, "pipeline_before.json")

	out, errs := c.Validate(before)
	if errs.Len() != 0 {
		t.Fatalf("stored document should validate, got: %v", errs)
	}
	// A document that already conforms must come back unchanged; otherwise
	// every round trip through the server would rewrite stored data.
	if got, want := canonical(t, out), canonical(t, before); got != want {
		t.Errorf("Validate changed a conforming document\n--- got\n%s\n--- want\n%s", got, want)
	}
}

func TestPipelineApplyMatchesScenario(t *testing.T) {
	c := pipelineCompiled(t)
	before := readDoc(t, "pipeline_before.json")

	var patch dyfields.Patch
	if err := json.Unmarshal(readFile(t, "pipeline_patch.json"), &patch); err != nil {
		t.Fatalf("unmarshal patch: %v", err)
	}

	got, errs := c.Apply(before, patch)
	if errs.Len() != 0 {
		t.Fatalf("apply should succeed, got: %v", errs)
	}
	want := readDoc(t, "pipeline_after.json")
	if canonical(t, got) != canonical(t, want) {
		t.Fatalf("apply result differs\n--- got\n%s\n--- want\n%s", canonical(t, got), canonical(t, want))
	}
}

// The single most important guarantee in the design: deleting an entry from the
// middle of a list must not shift anyone else's secret onto the wrong entry.
func TestPipelineSecretsStayAlignedAcrossDeletion(t *testing.T) {
	c := pipelineCompiled(t)
	before := readDoc(t, "pipeline_before.json")
	var patch dyfields.Patch
	if err := json.Unmarshal(readFile(t, "pipeline_patch.json"), &patch); err != nil {
		t.Fatalf("unmarshal patch: %v", err)
	}

	out, _ := c.Apply(before, patch)

	if got := out.Secrets["sinks.s_k1.mappings.m_3.secret_salt"]; got != "salt-for-phone" {
		t.Errorf("m_3 lost its own salt: %q", got)
	}
	for _, gone := range []string{
		"sinks.s_k1.mappings.m_2.secret_salt", // the deleted mapping
		"sinks.s_h1.auth_token",               // the deleted sink
		"webhook_secret",                      // the switched-off group
	} {
		if _, ok := out.Secrets[gone]; ok {
			t.Errorf("secret %q should have been removed", gone)
		}
	}
	if got := out.Secrets["source_password"]; got != "pg-secret" {
		// An untouched secret comes back as an empty string from the form,
		// which must mean "unchanged", not "clear it".
		t.Errorf("untouched secret was not preserved: %q", got)
	}
}

func TestPipelineImpactOfWarnsBeforeDataIsLost(t *testing.T) {
	c := pipelineCompiled(t)
	before := readDoc(t, "pipeline_before.json")

	// Switching a sink from kafka to http invalidates a whole subtree.
	patch := dyfields.Patch{Values: map[string]any{
		"sinks": []any{map[string]any{
			"$id":      "s_k1",
			"type":     "http",
			"endpoint": "https://elsewhere.internal/ingest",
		}},
	}}

	lost := c.ImpactOf(before, patch)
	want := map[string]bool{
		"sinks.s_k1.brokers":                    true,
		"sinks.s_k1.topic":                      true,
		"sinks.s_k1.mappings.m_1.partition_key": true,
		"sinks.s_k1.mappings.m_2.partition_key": true,
		"sinks.s_k1.mappings.m_3.partition_key": true,
	}
	got := map[string]bool{}
	for _, p := range lost {
		got[p] = true
	}
	for p := range want {
		if !got[p] {
			t.Errorf("ImpactOf did not report %q as lost (got %v)", p, lost)
		}
	}
}

func TestPipelineReadOnlyIsEnforcedOnApply(t *testing.T) {
	c := pipelineCompiled(t)
	before := readDoc(t, "pipeline_before.json")

	// is_running is true, so source_type is locked.
	patch := dyfields.Patch{Values: map[string]any{"source_type": "mysql"}}
	_, errs := c.Apply(before, patch)
	if !errs.Has("readonly") {
		t.Fatalf("changing a locked field should be rejected, got: %v", errs)
	}

	// Validate alone cannot see this, because it has nothing to compare
	// against. That is a documented boundary, not an oversight.
	tampered := before.Clone()
	tampered.Values["source_type"] = "mysql"
	if _, errs := c.Validate(tampered); errs.Len() != 0 {
		t.Fatalf("Validate does not check readonly, got: %v", errs)
	}
}

func TestPipelineMissingItemIDForSecretBearingEntry(t *testing.T) {
	c := pipelineCompiled(t)
	doc := readDoc(t, "pipeline_before.json")

	sinks := doc.Values["sinks"].([]any)
	sink := sinks[0].(map[string]any)
	maps := sink["mappings"].([]any)
	maps = append(maps, map[string]any{
		"source_field": "ip", "target_field": "client_ip", "transform": "mask",
	})
	sink["mappings"] = maps

	_, errs := c.Validate(doc)
	if !errs.Has("missing_item_id") {
		t.Fatalf("an entry with a secret must carry a $id, got: %v", errs)
	}
}

func TestPipelineDuplicateIsReportedWithoutLeakingTheValue(t *testing.T) {
	c := pipelineCompiled(t)
	doc := readDoc(t, "pipeline_before.json")

	sinks := doc.Values["sinks"].([]any)
	sink := sinks[0].(map[string]any)
	maps := sink["mappings"].([]any)
	first := maps[0].(map[string]any)
	maps[2].(map[string]any)["target_field"] = first["target_field"]

	_, errs := c.Validate(doc)
	if !errs.Has("duplicate") {
		t.Fatalf("duplicate target_field should be rejected, got: %v", errs)
	}
	for _, e := range errs {
		if e.Code == "duplicate" && e.ItemPath == "" {
			t.Error("a duplicate error must say which entry it is about")
		}
	}
}

func TestPipelineCrossLevelReferenceControlsVisibility(t *testing.T) {
	c := pipelineCompiled(t)
	doc := readDoc(t, "pipeline_before.json")

	vis := c.Visible(doc)
	// The kafka sink's mappings can be partition keys; the http sink's cannot,
	// because partition_key reads `^.type` from the sink it belongs to.
	if !vis.Fields["sinks[0].mappings[0].partition_key"] {
		t.Error("partition_key should be visible under a kafka sink")
	}
	if vis.Fields["sinks[1].mappings[0].partition_key"] {
		t.Error("partition_key should be hidden under an http sink")
	}
}

func TestPipelineHiddenValuesAreRemoved(t *testing.T) {
	c := pipelineCompiled(t)
	doc := readDoc(t, "pipeline_before.json")

	// A partition_key on the http sink's mapping is not merely ignored: an
	// invisible field has no value at all.
	sinks := doc.Values["sinks"].([]any)
	http := sinks[1].(map[string]any)
	http["mappings"].([]any)[0].(map[string]any)["partition_key"] = true

	out, errs := c.Validate(doc)
	if errs.Len() != 0 {
		t.Fatalf("unexpected errors: %v", errs)
	}
	m := out.Values["sinks"].([]any)[1].(map[string]any)["mappings"].([]any)[0].(map[string]any)
	if _, ok := m["partition_key"]; ok {
		t.Error("an invisible field kept its value")
	}
}

func TestPipelineRedactHidesEverySecret(t *testing.T) {
	c := pipelineCompiled(t)
	doc := readDoc(t, "pipeline_before.json")

	pub := c.Redact(doc)
	b, err := json.Marshal(pub)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	for _, secret := range []string{"pg-secret", "salt-for-email", "bearer-xyz", "hook-abc"} {
		if contains(string(b), secret) {
			t.Errorf("redacted document still contains %q", secret)
		}
	}
	if len(pub.SecretsSet) != len(doc.Secrets) {
		t.Errorf("expected %d secret paths, got %d", len(doc.Secrets), len(pub.SecretsSet))
	}
}

func contains(haystack, needle string) bool {
	return len(needle) > 0 && len(haystack) >= len(needle) &&
		func() bool {
			for i := 0; i+len(needle) <= len(haystack); i++ {
				if haystack[i:i+len(needle)] == needle {
					return true
				}
			}
			return false
		}()
}