281 lines
8.2 KiB
Go
281 lines
8.2 KiB
Go
// conditions_read_test.go validates current-conditions repository read flow.
|
|
// Layer: adapters/outbound/postgres conditions read tests.
|
|
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"database/sql/driver"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
const currentConditionsTestDriverName = "weatherapi_current_conditions_read_test"
|
|
|
|
func init() {
|
|
sql.Register(currentConditionsTestDriverName, currentConditionsTestDriver{})
|
|
}
|
|
|
|
func TestCurrentConditionsUsesConsensusConditionCode(t *testing.T) {
|
|
repo, closeDB := openCurrentConditionsTestRepository(t,
|
|
currentConditionsAggregateQuery(currentConditionsAggregateRow(3), nil),
|
|
currentConditionsConditionCodeCandidatesQuery([][]driver.Value{
|
|
{"source-a", int64(1)},
|
|
{"source-b", int64(2)},
|
|
{"source-c", int64(95)},
|
|
}, nil),
|
|
)
|
|
defer closeDB()
|
|
|
|
conditions, err := repo.CurrentConditions(context.Background(), 15)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
if conditions == nil {
|
|
t.Fatal("expected current conditions")
|
|
}
|
|
if conditions.ConditionCode != 1 {
|
|
t.Fatalf("expected consensus condition code 1, got %d", conditions.ConditionCode)
|
|
}
|
|
if conditions.TemperatureC == nil || *conditions.TemperatureC != 15.5 {
|
|
t.Fatalf("expected temperature 15.5, got %v", conditions.TemperatureC)
|
|
}
|
|
if conditions.WindDirectionDegrees == nil || *conditions.WindDirectionDegrees != 182.5 {
|
|
t.Fatalf("expected wind direction 182.5, got %v", conditions.WindDirectionDegrees)
|
|
}
|
|
if conditions.IsDay == nil || !*conditions.IsDay {
|
|
t.Fatalf("expected isDay true, got %v", conditions.IsDay)
|
|
}
|
|
assertCurrentConditionsTestQueriesConsumed(t)
|
|
}
|
|
|
|
func TestCurrentConditionsNoSamplesSkipsConditionCodeCandidates(t *testing.T) {
|
|
repo, closeDB := openCurrentConditionsTestRepository(t,
|
|
currentConditionsAggregateQuery([]driver.Value{int64(0), nil, nil, nil, nil, nil, nil, nil}, nil),
|
|
)
|
|
defer closeDB()
|
|
|
|
conditions, err := repo.CurrentConditions(context.Background(), 15)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
if conditions != nil {
|
|
t.Fatalf("expected nil current conditions, got %+v", conditions)
|
|
}
|
|
assertCurrentConditionsTestQueriesConsumed(t)
|
|
}
|
|
|
|
func TestCurrentConditionsConditionCodeCandidateQueryUsesLatestPerSourceOrdering(t *testing.T) {
|
|
query := compactSQL(queryCurrentConditionsConditionCodeCandidates)
|
|
want := "PARTITION BY event_source ORDER BY observed_at DESC, event_emitted_at DESC"
|
|
if !strings.Contains(query, want) {
|
|
t.Fatalf("expected condition code candidate query to contain %q, got %q", want, query)
|
|
}
|
|
if !strings.Contains(query, "SELECT event_source, condition_code") {
|
|
t.Fatalf("expected condition code candidate query to select event_source and condition_code, got %q", query)
|
|
}
|
|
}
|
|
|
|
func TestCurrentConditionsAggregateQueryDoesNotSelectConditionCode(t *testing.T) {
|
|
query := compactSQL(queryCurrentConditions)
|
|
if strings.Contains(query, "condition_code") {
|
|
t.Fatalf("expected aggregate query not to select condition_code, got %q", query)
|
|
}
|
|
}
|
|
|
|
func TestCurrentConditionsConditionCodeCandidateQueryWrapsErrors(t *testing.T) {
|
|
repo, closeDB := openCurrentConditionsTestRepository(t,
|
|
currentConditionsAggregateQuery(currentConditionsAggregateRow(1), nil),
|
|
scriptedCurrentConditionsQuery{
|
|
name: "condition code candidates",
|
|
query: queryCurrentConditionsConditionCodeCandidates,
|
|
args: []driver.Value{int64(15)},
|
|
err: errors.New("candidate query unavailable"),
|
|
},
|
|
)
|
|
defer closeDB()
|
|
|
|
_, err := repo.CurrentConditions(context.Background(), 15)
|
|
if err == nil {
|
|
t.Fatal("expected error")
|
|
}
|
|
if !strings.Contains(err.Error(), "query current conditions condition code candidates") {
|
|
t.Fatalf("expected candidate query context, got %v", err)
|
|
}
|
|
}
|
|
|
|
func openCurrentConditionsTestRepository(t *testing.T, queries ...scriptedCurrentConditionsQuery) (*Repository, func()) {
|
|
t.Helper()
|
|
currentConditionsTestScript.set(queries)
|
|
|
|
db, err := sql.Open(currentConditionsTestDriverName, "")
|
|
if err != nil {
|
|
t.Fatalf("open test db: %v", err)
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
|
|
return NewRepository(db), func() {
|
|
_ = db.Close()
|
|
currentConditionsTestScript.set(nil)
|
|
}
|
|
}
|
|
|
|
func assertCurrentConditionsTestQueriesConsumed(t *testing.T) {
|
|
t.Helper()
|
|
if remaining := currentConditionsTestScript.remaining(); remaining != 0 {
|
|
t.Fatalf("expected all scripted queries consumed, got %d remaining", remaining)
|
|
}
|
|
}
|
|
|
|
func currentConditionsAggregateQuery(row []driver.Value, nextErr error) scriptedCurrentConditionsQuery {
|
|
return scriptedCurrentConditionsQuery{
|
|
name: "aggregate",
|
|
query: queryCurrentConditions,
|
|
args: []driver.Value{int64(15)},
|
|
columns: []string{"sample_count", "temperature_c", "apparent_temperature_c", "dewpoint_c", "relative_humidity_percent", "wind_speed_kmh", "wind_direction_degrees", "is_day"},
|
|
rows: [][]driver.Value{row},
|
|
nextErr: nextErr,
|
|
}
|
|
}
|
|
|
|
func currentConditionsConditionCodeCandidatesQuery(rows [][]driver.Value, nextErr error) scriptedCurrentConditionsQuery {
|
|
return scriptedCurrentConditionsQuery{
|
|
name: "condition code candidates",
|
|
query: queryCurrentConditionsConditionCodeCandidates,
|
|
args: []driver.Value{int64(15)},
|
|
columns: []string{"event_source", "condition_code"},
|
|
rows: rows,
|
|
nextErr: nextErr,
|
|
}
|
|
}
|
|
|
|
func currentConditionsAggregateRow(sampleCount int64) []driver.Value {
|
|
return []driver.Value{
|
|
sampleCount,
|
|
float64(15.5),
|
|
float64(14.2),
|
|
float64(10.1),
|
|
float64(72),
|
|
float64(24.8),
|
|
float64(182.5),
|
|
true,
|
|
}
|
|
}
|
|
|
|
type currentConditionsTestDriver struct{}
|
|
|
|
func (currentConditionsTestDriver) Open(string) (driver.Conn, error) {
|
|
return currentConditionsTestConn{}, nil
|
|
}
|
|
|
|
type currentConditionsTestConn struct{}
|
|
|
|
func (currentConditionsTestConn) Prepare(string) (driver.Stmt, error) {
|
|
return nil, errors.New("prepare is not supported")
|
|
}
|
|
|
|
func (currentConditionsTestConn) Close() error {
|
|
return nil
|
|
}
|
|
|
|
func (currentConditionsTestConn) Begin() (driver.Tx, error) {
|
|
return nil, errors.New("transactions are not supported")
|
|
}
|
|
|
|
func (currentConditionsTestConn) QueryContext(_ context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
|
|
return currentConditionsTestScript.next(query, args)
|
|
}
|
|
|
|
type scriptedCurrentConditionsQuery struct {
|
|
name string
|
|
query string
|
|
args []driver.Value
|
|
columns []string
|
|
rows [][]driver.Value
|
|
err error
|
|
nextErr error
|
|
}
|
|
|
|
type currentConditionsTestScriptState struct {
|
|
mu sync.Mutex
|
|
queries []scriptedCurrentConditionsQuery
|
|
}
|
|
|
|
var currentConditionsTestScript currentConditionsTestScriptState
|
|
|
|
func (s *currentConditionsTestScriptState) set(queries []scriptedCurrentConditionsQuery) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.queries = append([]scriptedCurrentConditionsQuery(nil), queries...)
|
|
}
|
|
|
|
func (s *currentConditionsTestScriptState) remaining() int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return len(s.queries)
|
|
}
|
|
|
|
func (s *currentConditionsTestScriptState) next(query string, args []driver.NamedValue) (driver.Rows, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
if len(s.queries) == 0 {
|
|
return nil, fmt.Errorf("unexpected query: %s", compactSQL(query))
|
|
}
|
|
next := s.queries[0]
|
|
s.queries = s.queries[1:]
|
|
|
|
if compactSQL(query) != compactSQL(next.query) {
|
|
return nil, fmt.Errorf("expected %s query %q, got %q", next.name, compactSQL(next.query), compactSQL(query))
|
|
}
|
|
if len(args) != len(next.args) {
|
|
return nil, fmt.Errorf("expected %s args %v, got %v", next.name, next.args, namedValues(args))
|
|
}
|
|
for i, arg := range args {
|
|
if arg.Value != next.args[i] {
|
|
return nil, fmt.Errorf("expected %s arg %d to be %v, got %v", next.name, i, next.args[i], arg.Value)
|
|
}
|
|
}
|
|
if next.err != nil {
|
|
return nil, next.err
|
|
}
|
|
return ¤tConditionsTestRows{
|
|
columns: append([]string(nil), next.columns...),
|
|
rows: append([][]driver.Value(nil), next.rows...),
|
|
nextErr: next.nextErr,
|
|
}, nil
|
|
}
|
|
|
|
type currentConditionsTestRows struct {
|
|
columns []string
|
|
rows [][]driver.Value
|
|
index int
|
|
nextErr error
|
|
}
|
|
|
|
func (r *currentConditionsTestRows) Columns() []string {
|
|
return r.columns
|
|
}
|
|
|
|
func (r *currentConditionsTestRows) Close() error {
|
|
return nil
|
|
}
|
|
|
|
func (r *currentConditionsTestRows) Next(dest []driver.Value) error {
|
|
if r.index >= len(r.rows) {
|
|
if r.nextErr != nil {
|
|
err := r.nextErr
|
|
r.nextErr = nil
|
|
return err
|
|
}
|
|
return io.EOF
|
|
}
|
|
copy(dest, r.rows[r.index])
|
|
r.index++
|
|
return nil
|
|
}
|