weatherfeeder/internal/sinks/postgres/schema_test.go

package postgres

import (
	"strings"
	"testing"
)

func TestWeatherPostgresSchemaShape(t *testing.T) {
	s := PostgresSchema()
	if s.MapEvent == nil {
		t.Fatalf("PostgresSchema().MapEvent is nil")
	}

	wantTables := map[string]bool{
		tableObservations:                  true,
		tableObservationPresentWeather:     true,
		tableForecasts:                     true,
		tableForecastPeriods:               true,
		tableForecastDiscussions:           true,
		tableForecastDiscussionKeyMessages: true,
		tableWeatherStoryRuns:              true,
		tableWeatherStories:                true,
		tableAlertRuns:                     true,
		tableAlerts:                        true,
		tableAlertReferences:               true,
		tableOutlookRuns:                   true,
		tableOutlooks:                      true,
	}

	if len(s.Tables) != len(wantTables) {
		t.Fatalf("PostgresSchema().Tables len = %d, want %d", len(s.Tables), len(wantTables))
	}

	seenIndexes := map[string]bool{}
	for _, tbl := range s.Tables {
		if !wantTables[tbl.Name] {
			t.Fatalf("unexpected table %q in schema", tbl.Name)
		}
		if tbl.PruneColumn == "" {
			t.Fatalf("table %q missing prune column", tbl.Name)
		}
		for _, idx := range tbl.Indexes {
			if seenIndexes[idx.Name] {
				t.Fatalf("duplicate index name %q", idx.Name)
			}
			seenIndexes[idx.Name] = true
		}
	}
}

func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
	runColumns := columnsForTable(t, tableOutlookRuns)
	for _, col := range []string{"event_id", "event_kind", "event_source", "event_schema", "event_emitted_at", "event_effective_at", "location_id", "location_name", "latitude", "longitude", "as_of", "issued_at", "outlook_count"} {
		if !runColumns[col] {
			t.Fatalf("%s missing %s column", tableOutlookRuns, col)
		}
	}
	assertTablePrimaryKey(t, tableOutlookRuns, []string{"event_id"})
	assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_location_as_of", []string{"location_id", "as_of"})
	assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_as_of", []string{"as_of"})

	outlookColumns := columnsForTable(t, tableOutlooks)
	for _, col := range []string{"run_event_id", "outlook_index", "as_of", "product", "day", "outlook_type", "label", "label_text", "severity_rank", "valid_from", "valid_to", "issued_at", "expires_at", "forecaster", "headline", "summary", "discussion", "source_url", "image_url", "contains_location", "geometry_json"} {
		if !outlookColumns[col] {
			t.Fatalf("%s missing %s column", tableOutlooks, col)
		}
	}
	assertTablePrimaryKey(t, tableOutlooks, []string{"run_event_id", "outlook_index"})
	assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_contains_valid", []string{"contains_location", "valid_from", "valid_to"})
	assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_day_type_label", []string{"day", "outlook_type", "label"})
	assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_valid", []string{"valid_from", "valid_to"})
}

func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
	runColumns := columnsForTable(t, tableWeatherStoryRuns)
	if !runColumns["as_of"] {
		t.Fatalf("%s missing as_of column", tableWeatherStoryRuns)
	}
	if !runColumns["story_count"] {
		t.Fatalf("%s missing story_count column", tableWeatherStoryRuns)
	}

	storyColumns := columnsForTable(t, tableWeatherStories)
	for _, col := range []string{"start_time", "end_time", "updated_at", "title", "description", "alt_text", "priority", "story_order", "download_url"} {
		if !storyColumns[col] {
			t.Fatalf("%s missing %s column", tableWeatherStories, col)
		}
	}
}

func assertTablePrimaryKey(t *testing.T, table string, want []string) {
	t.Helper()
	for _, tbl := range PostgresSchema().Tables {
		if tbl.Name != table {
			continue
		}
		if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") {
			t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want)
		}
		return
	}
	t.Fatalf("missing table %q", table)
}

func assertTableIndex(t *testing.T, table string, name string, want []string) {
	t.Helper()
	for _, tbl := range PostgresSchema().Tables {
		if tbl.Name != table {
			continue
		}
		for _, idx := range tbl.Indexes {
			if idx.Name == name {
				if strings.Join(idx.Columns, ",") != strings.Join(want, ",") {
					t.Fatalf("%s index %s columns = %#v, want %#v", table, name, idx.Columns, want)
				}
				return
			}
		}
		t.Fatalf("%s missing index %s", table, name)
	}
	t.Fatalf("missing table %q", table)
}

func columnsForTable(t *testing.T, table string) map[string]bool {
	t.Helper()

	schema := PostgresSchema()
	for _, tbl := range schema.Tables {
		if tbl.Name != table {
			continue
		}
		cols := make(map[string]bool, len(tbl.Columns))
		for _, col := range tbl.Columns {
			cols[col.Name] = true
		}
		return cols
	}
	t.Fatalf("missing table %q", table)
	return nil
}