weatherfeeder/cmd/weatherfeeder/main_test.go

package main

import (
	"context"
	"reflect"
	"strings"
	"testing"
	"time"

	"gitea.maximumdirect.net/ejr/feedkit/config"
	fkevent "gitea.maximumdirect.net/ejr/feedkit/event"
	fkpipeline "gitea.maximumdirect.net/ejr/feedkit/pipeline"
	fkprocessors "gitea.maximumdirect.net/ejr/feedkit/processors"
	fkdedupe "gitea.maximumdirect.net/ejr/feedkit/processors/dedupe"
	fknormalize "gitea.maximumdirect.net/ejr/feedkit/processors/normalize"

	wfnormalizers "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers"
)

type testInput struct {
	name string
}

func (s testInput) Name() string { return s.name }

type testKindSource struct {
	testInput
	kind fkevent.Kind
}

func (s testKindSource) Kind() fkevent.Kind { return s.kind }

type testKindsSource struct {
	testInput
	kinds []fkevent.Kind
}

func (s testKindsSource) Kinds() []fkevent.Kind { return s.kinds }

func TestValidateSourceExpectedKindsLegacyKindFallback(t *testing.T) {
	sc := config.SourceConfig{Kind: "observation"}
	in := testKindSource{
		testInput: testInput{name: "test"},
		kind:      fkevent.Kind("observation"),
	}

	if err := validateSourceExpectedKinds(sc, in); err != nil {
		t.Fatalf("validateSourceExpectedKinds() unexpected error: %v", err)
	}
}

func TestValidateSourceExpectedKindsSubsetAllowed(t *testing.T) {
	sc := config.SourceConfig{Kinds: []string{"observation"}}
	in := testKindsSource{
		testInput: testInput{name: "test"},
		kinds:     []fkevent.Kind{"observation", "forecast"},
	}

	if err := validateSourceExpectedKinds(sc, in); err != nil {
		t.Fatalf("validateSourceExpectedKinds() unexpected error: %v", err)
	}
}

func TestValidateSourceExpectedKindsMismatchFails(t *testing.T) {
	sc := config.SourceConfig{Kinds: []string{"alert"}}
	in := testKindsSource{
		testInput: testInput{name: "test"},
		kinds:     []fkevent.Kind{"observation", "forecast"},
	}

	err := validateSourceExpectedKinds(sc, in)
	if err == nil {
		t.Fatalf("validateSourceExpectedKinds() expected mismatch error, got nil")
	}
	if !strings.Contains(err.Error(), "configured expected kind") {
		t.Fatalf("validateSourceExpectedKinds() error %q does not include expected message", err)
	}
}

func TestValidateSourceExpectedKindsNoMetadataSkipsCheck(t *testing.T) {
	sc := config.SourceConfig{Kinds: []string{"alert"}}
	in := testInput{name: "test"}

	if err := validateSourceExpectedKinds(sc, in); err != nil {
		t.Fatalf("validateSourceExpectedKinds() unexpected error: %v", err)
	}
}

func TestParseExpectedKindsRejectsEmptyValues(t *testing.T) {
	if _, err := parseExpectedKinds([]string{""}); err == nil {
		t.Fatalf("parseExpectedKinds() expected error for empty kind")
	}
}

func TestExampleConfigLoads(t *testing.T) {
	if _, err := config.Load("config.yml"); err != nil {
		t.Fatalf("config.Load(config.yml) unexpected error: %v", err)
	}
}

func TestProcessorRegistryBuildsNormalizeThenDedupeChain(t *testing.T) {
	chain, err := buildProcessorChainForTests()
	if err != nil {
		t.Fatalf("BuildChain() unexpected error: %v", err)
	}
	if len(chain) != 2 {
		t.Fatalf("BuildChain() expected 2 processors, got %d", len(chain))
	}

	pl := &fkpipeline.Pipeline{Processors: chain}
	if len(pl.Processors) != 2 {
		t.Fatalf("pipeline expected 2 processors, got %d", len(pl.Processors))
	}
}

func TestNormalizeNoMatchPassThrough(t *testing.T) {
	chain, err := buildProcessorChainForTests()
	if err != nil {
		t.Fatalf("BuildChain() unexpected error: %v", err)
	}

	pl := &fkpipeline.Pipeline{Processors: chain}
	in := fkevent.Event{
		ID:        "evt-no-match",
		Kind:      fkevent.Kind("observation"),
		Source:    "test",
		EmittedAt: time.Now().UTC(),
		Schema:    "raw.weatherfeeder.unknown.v1",
		Payload: map[string]any{
			"ok": true,
		},
	}

	out, err := pl.Process(context.Background(), in)
	if err != nil {
		t.Fatalf("Pipeline.Process() unexpected error: %v", err)
	}
	if out == nil {
		t.Fatalf("Pipeline.Process() returned nil output")
	}
	if !reflect.DeepEqual(*out, in) {
		t.Fatalf("Pipeline.Process() expected passthrough output, got %#v", *out)
	}
}

func TestDedupeDropsSecondEventWithSameID(t *testing.T) {
	chain, err := buildProcessorChainForTests()
	if err != nil {
		t.Fatalf("BuildChain() unexpected error: %v", err)
	}

	pl := &fkpipeline.Pipeline{Processors: chain}
	in := fkevent.Event{
		ID:        "evt-dedupe-1",
		Kind:      fkevent.Kind("observation"),
		Source:    "test",
		EmittedAt: time.Now().UTC(),
		Schema:    "raw.weatherfeeder.unknown.v1",
		Payload: map[string]any{
			"ok": true,
		},
	}

	first, err := pl.Process(context.Background(), in)
	if err != nil {
		t.Fatalf("first Pipeline.Process() unexpected error: %v", err)
	}
	if first == nil {
		t.Fatalf("first Pipeline.Process() unexpectedly dropped event")
	}

	second, err := pl.Process(context.Background(), in)
	if err != nil {
		t.Fatalf("second Pipeline.Process() unexpected error: %v", err)
	}
	if second != nil {
		t.Fatalf("second Pipeline.Process() expected dedupe drop, got %#v", *second)
	}
}

func buildProcessorChainForTests() ([]fkprocessors.Processor, error) {
	normalizers := wfnormalizers.RegisterBuiltins(nil)

	procReg := fkprocessors.NewRegistry()
	procReg.Register("normalize", func() (fkprocessors.Processor, error) {
		return fknormalize.NewProcessor(normalizers, false), nil
	})
	procReg.Register("dedupe", fkdedupe.Factory(dedupeMaxEntries))

	return procReg.BuildChain([]string{"normalize", "dedupe"})
}