weatherfeeder/cmd/weatherfeeder/main_test.go

package main

import (
	"context"
	"path/filepath"
	"reflect"
	"sort"
	"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"
	fkscheduler "gitea.maximumdirect.net/ejr/feedkit/scheduler"
	fksources "gitea.maximumdirect.net/ejr/feedkit/sources"

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

type testInput struct {
	name string
}

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

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

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

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

	if err := fksources.ValidateExpectedKinds(sc, in); err != nil {
		t.Fatalf("ValidateExpectedKinds() 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 := fksources.ValidateExpectedKinds(sc, in)
	if err == nil {
		t.Fatalf("ValidateExpectedKinds() expected mismatch error, got nil")
	}
	if !strings.Contains(err.Error(), "configured expected kind") {
		t.Fatalf("ValidateExpectedKinds() 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 := fksources.ValidateExpectedKinds(sc, in); err != nil {
		t.Fatalf("ValidateExpectedKinds() unexpected error: %v", err)
	}
}

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

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

	assertConfigSourcesBuildSchedulerJobs(t, cfg)
}

func TestMaintainedConfigExamplesLoad(t *testing.T) {
	paths, err := filepath.Glob("../../examples/*.yml")
	if err != nil {
		t.Fatalf("filepath.Glob examples: %v", err)
	}
	sort.Strings(paths)
	if len(paths) == 0 {
		t.Fatalf("expected maintained config examples")
	}

	for _, path := range paths {
		t.Run(filepath.Base(path), func(t *testing.T) {
			cfg, err := config.Load(path)
			if err != nil {
				t.Fatalf("config.Load(%s) unexpected error: %v", path, err)
			}
			assertConfigSourcesBuildSchedulerJobs(t, cfg)
		})
	}
}

func assertConfigSourcesBuildSchedulerJobs(t *testing.T, cfg *config.Config) {
	t.Helper()

	reg := fksources.NewRegistry()
	wfsources.RegisterBuiltins(reg)

	for i, sc := range cfg.Sources {
		in, err := reg.BuildInput(sc)
		if err != nil {
			t.Fatalf("BuildInput(sources[%d]) error = %v", i, err)
		}

		if err := fksources.ValidateExpectedKinds(sc, in); err != nil {
			t.Fatalf("ValidateExpectedKinds(sources[%d]) error = %v", i, err)
		}

		job, err := fkscheduler.JobFromSourceConfig(in, sc)
		if err != nil {
			t.Fatalf("JobFromSourceConfig(sources[%d]) error = %v", i, err)
		}
		if job.Source == nil {
			t.Fatalf("JobFromSourceConfig(sources[%d]) returned nil source", i)
		}
	}
}

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"})
}