479 lines
15 KiB
Go
479 lines
15 KiB
Go
package history
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
func TestSchemaStatementsCreateCoreStables(t *testing.T) {
|
|
statements := strings.Join(SchemaStatements("test_ts"), "\n")
|
|
for _, want := range []string{"CREATE DATABASE IF NOT EXISTS test_ts", "raw_frames", "vehicle_locations"} {
|
|
if !strings.Contains(statements, want) {
|
|
t.Fatalf("schema missing %q:\n%s", want, statements)
|
|
}
|
|
}
|
|
if strings.Contains(statements, "vehicle_mileage_points") {
|
|
t.Fatalf("schema should not create duplicate mileage stable:\n%s", statements)
|
|
}
|
|
if strings.Contains(statements, "fields_json") {
|
|
t.Fatalf("raw schema should not create duplicate fields_json column:\n%s", statements)
|
|
}
|
|
locationStable := between(statements, "CREATE STABLE IF NOT EXISTS vehicle_locations", "}")
|
|
for _, field := range []string{"frame_id", "vehicle_key", "phone", "device_id"} {
|
|
if strings.Contains(locationStable, field) {
|
|
t.Fatalf("location stable should only keep core location columns and protocol/vin tags, found %s:\n%s", field, locationStable)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
|
|
if err := writer.AppendAll(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendAll() error = %v", err)
|
|
}
|
|
|
|
if got := countSQL(exec.calls, "USING raw_frames"); got != 1 {
|
|
t.Fatalf("raw child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "USING vehicle_locations"); got != 1 {
|
|
t.Fatalf("location child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "USING vehicle_mileage_points"); got != 0 {
|
|
t.Fatalf("duplicate mileage child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
|
|
t.Fatalf("raw insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 1 {
|
|
t.Fatalf("location insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 0 {
|
|
t.Fatalf("duplicate mileage insert count = %d", got)
|
|
}
|
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
|
if !strings.Contains(rawInsert, "'go_") || !strings.Contains(rawInsert, "'2e") && !strings.Contains(rawInsert, "'") {
|
|
t.Fatalf("raw insert should quote string literals: %s", rawInsert)
|
|
}
|
|
if strings.Contains(rawInsert, "VALUES (?,") {
|
|
t.Fatalf("raw insert should not use placeholders for tdengine websocket plain insert: %s", rawInsert)
|
|
}
|
|
if strings.Contains(rawInsert, "fields_json") {
|
|
t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert)
|
|
}
|
|
locationChild := findSQL(exec.calls, "USING vehicle_locations")
|
|
for _, want := range []string{"'JT808'", "'LNBVIN00000000001'"} {
|
|
if !strings.Contains(locationChild, want) {
|
|
t.Fatalf("location child should tag by protocol and vin only: %s", locationChild)
|
|
}
|
|
}
|
|
for _, field := range []string{"'JT808:013307795425'", "'13307795425'", "device_id"} {
|
|
if strings.Contains(locationChild, field) {
|
|
t.Fatalf("location child should not carry raw identity fallback tag %s: %s", field, locationChild)
|
|
}
|
|
}
|
|
locationInsert := findSQL(exec.calls, "INSERT INTO loc_")
|
|
if strings.Contains(locationInsert, "frame_id") || strings.Contains(locationInsert, "go_") {
|
|
t.Fatalf("location insert should not duplicate raw frame id: %s", locationInsert)
|
|
}
|
|
}
|
|
|
|
func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
|
|
if err := writer.AppendRawFrame(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendRawFrame() error = %v", err)
|
|
}
|
|
|
|
createChild := findSQL(exec.calls, "USING raw_frames")
|
|
if !strings.Contains(createChild, "'13307795425'") {
|
|
t.Fatalf("child table phone tag should be normalized: %s", createChild)
|
|
}
|
|
if strings.Contains(createChild, "'013307795425'") {
|
|
t.Fatalf("child table phone tag should not keep leading zero: %s", createChild)
|
|
}
|
|
refreshTag := findSQL(exec.calls, "SET TAG phone")
|
|
if !strings.Contains(refreshTag, "'13307795425'") {
|
|
t.Fatalf("phone tag refresh should use normalized phone: %s", refreshTag)
|
|
}
|
|
}
|
|
|
|
func TestWriterChunksOversizedParsedFields(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
env.ParsedFields = map[string]any{
|
|
"jt808.location.large_payload": strings.Repeat("x", 16_128),
|
|
}
|
|
|
|
if err := writer.AppendRawFrame(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendRawFrame() error = %v", err)
|
|
}
|
|
|
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
|
if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_fields"`) {
|
|
t.Fatalf("raw insert should store a parsed_fields chunk manifest: %s", rawInsert)
|
|
}
|
|
if got := countSQL(exec.calls, "USING raw_frame_payload_chunks"); got != 1 {
|
|
t.Fatalf("chunk child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO chunk_"); got < 2 {
|
|
t.Fatalf("chunk insert count = %d, calls=%v", got, exec.calls)
|
|
}
|
|
}
|
|
|
|
func TestWriterUsesEnvelopeParsedFieldsWithoutReflattening(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
env.Parsed = map[string]any{
|
|
"location": map[string]any{"speed_kmh": 99},
|
|
}
|
|
env.ParsedFields = map[string]any{
|
|
"jt808.location.speed_kmh": "12.3",
|
|
}
|
|
|
|
if err := writer.AppendRawFrame(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendRawFrame() error = %v", err)
|
|
}
|
|
|
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
|
if !strings.Contains(rawInsert, `"jt808.location.speed_kmh":"12.3"`) {
|
|
t.Fatalf("raw insert should use precomputed parsed fields: %s", rawInsert)
|
|
}
|
|
if strings.Contains(rawInsert, `"jt808.location.speed_kmh":"99"`) {
|
|
t.Fatalf("raw insert should not re-flatten parsed payload: %s", rawInsert)
|
|
}
|
|
}
|
|
|
|
func TestWriterLeavesParsedFieldsEmptyWhenNoParsedPayload(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
env.Parsed = nil
|
|
env.ParseStatus = envelope.ParseBadFrame
|
|
env.ParseError = "invalid frame checksum"
|
|
|
|
if err := writer.AppendRawFrame(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendRawFrame() error = %v", err)
|
|
}
|
|
|
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
|
if strings.Contains(rawInsert, "'{}'") || strings.Contains(rawInsert, "'null'") {
|
|
t.Fatalf("raw insert should not store empty parsed_fields payload: %s", rawInsert)
|
|
}
|
|
if !strings.Contains(rawInsert, "'invalid frame checksum'") {
|
|
t.Fatalf("raw insert should keep parse error: %s", rawInsert)
|
|
}
|
|
}
|
|
|
|
func TestTimeLiteralsUseEpochMilliseconds(t *testing.T) {
|
|
value := time.Date(2026, 7, 3, 1, 12, 59, 77_000_000, time.UTC)
|
|
|
|
if got, want := literal(value), "1783041179077"; got != want {
|
|
t.Fatalf("time literal = %s, want %s", got, want)
|
|
}
|
|
}
|
|
|
|
func TestWriterSkipsSparseDerivedRows(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
env.Fields = map[string]any{}
|
|
|
|
if err := writer.AppendAll(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendAll() error = %v", err)
|
|
}
|
|
|
|
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
|
|
t.Fatalf("raw insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 0 {
|
|
t.Fatalf("location insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 0 {
|
|
t.Fatalf("mileage insert count = %d", got)
|
|
}
|
|
}
|
|
|
|
func TestWriterSkipsLocationWhenVINIsMissing(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
env := sampleEnvelope()
|
|
env.VIN = ""
|
|
|
|
if err := writer.AppendAll(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendAll() error = %v", err)
|
|
}
|
|
|
|
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
|
|
t.Fatalf("raw insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "USING vehicle_locations"); got != 0 {
|
|
t.Fatalf("location child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 0 {
|
|
t.Fatalf("location insert count = %d", got)
|
|
}
|
|
}
|
|
|
|
func TestWriterAppendsBatchRowsByChildTable(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
first := sampleEnvelope()
|
|
second := sampleEnvelope()
|
|
second.Sequence = 2
|
|
second.EventTimeMS += 1000
|
|
second.ReceivedAtMS += 1000
|
|
|
|
if err := writer.AppendAllBatch(context.Background(), []envelope.FrameEnvelope{first, second}); err != nil {
|
|
t.Fatalf("AppendAllBatch() error = %v", err)
|
|
}
|
|
|
|
if got := countSQL(exec.calls, "USING raw_frames"); got != 1 {
|
|
t.Fatalf("raw child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "USING vehicle_locations"); got != 1 {
|
|
t.Fatalf("location child create count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
|
|
t.Fatalf("raw batch insert count = %d", got)
|
|
}
|
|
if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 1 {
|
|
t.Fatalf("location batch insert count = %d", got)
|
|
}
|
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
|
if got := strings.Count(rawInsert, "),(") + 1; got != 2 {
|
|
t.Fatalf("raw batch row count = %d, sql=%s", got, rawInsert)
|
|
}
|
|
locationInsert := findSQL(exec.calls, "INSERT INTO loc_")
|
|
if got := strings.Count(locationInsert, "),(") + 1; got != 2 {
|
|
t.Fatalf("location batch row count = %d, sql=%s", got, locationInsert)
|
|
}
|
|
}
|
|
|
|
func TestWriterWithDatabaseQualifiesTDengineTables(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriterWithDatabase(exec, "vehicle_ts")
|
|
env := sampleEnvelope()
|
|
|
|
if err := writer.AppendAll(context.Background(), env); err != nil {
|
|
t.Fatalf("AppendAll() error = %v", err)
|
|
}
|
|
|
|
for _, want := range []string{
|
|
"CREATE TABLE IF NOT EXISTS vehicle_ts.raw_",
|
|
"USING vehicle_ts.raw_frames",
|
|
"ALTER TABLE vehicle_ts.raw_",
|
|
"INSERT INTO vehicle_ts.raw_",
|
|
"CREATE TABLE IF NOT EXISTS vehicle_ts.loc_",
|
|
"USING vehicle_ts.vehicle_locations",
|
|
"INSERT INTO vehicle_ts.loc_",
|
|
} {
|
|
if !containsSQL(exec.calls, want) {
|
|
t.Fatalf("qualified sql missing %q, calls=%v", want, exec.calls)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestWriterAppendAllBatchSkipsEmptyBatch(t *testing.T) {
|
|
exec := &recordingExec{}
|
|
writer := NewWriter(exec)
|
|
|
|
if err := writer.AppendAllBatch(context.Background(), nil); err != nil {
|
|
t.Fatalf("AppendAllBatch() error = %v", err)
|
|
}
|
|
if len(exec.calls) != 0 {
|
|
t.Fatalf("exec calls = %#v, want none", exec.calls)
|
|
}
|
|
}
|
|
|
|
func TestRawSizeBytesUsesRawTextWhenHexIsEmpty(t *testing.T) {
|
|
env := envelope.FrameEnvelope{RawText: `{"code":"0F80","data":{"speed":12}}`}
|
|
if got, want := rawSizeBytes(env), len([]byte(env.RawText)); got != want {
|
|
t.Fatalf("raw text size = %d, want %d", got, want)
|
|
}
|
|
|
|
env.RawHex = "7E0200"
|
|
if got, want := rawSizeBytes(env), 3; got != want {
|
|
t.Fatalf("raw hex size = %d, want %d", got, want)
|
|
}
|
|
}
|
|
|
|
func TestSchemaStatementsCreateRawFramePayloadChunks(t *testing.T) {
|
|
statements := strings.Join(SchemaStatements("test_ts"), "\n")
|
|
for _, want := range []string{"raw_frame_payload_chunks", "payload_kind NCHAR(32)", "chunk_index INT", "chunk_text BINARY(16000)"} {
|
|
if !strings.Contains(statements, want) {
|
|
t.Fatalf("schema missing %q:\n%s", want, statements)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestWriterEnsuresSameChildTableOnceUnderConcurrency(t *testing.T) {
|
|
exec := newBlockingExec("CREATE TABLE IF NOT EXISTS raw_")
|
|
writer := NewWriter(exec)
|
|
first := sampleEnvelope()
|
|
second := sampleEnvelope()
|
|
second.Sequence = 2
|
|
second.EventID = "second"
|
|
|
|
var wg sync.WaitGroup
|
|
errs := make(chan error, 2)
|
|
wg.Add(2)
|
|
go func() {
|
|
defer wg.Done()
|
|
errs <- writer.AppendAll(context.Background(), first)
|
|
}()
|
|
<-exec.blocked
|
|
go func() {
|
|
defer wg.Done()
|
|
errs <- writer.AppendAll(context.Background(), second)
|
|
}()
|
|
time.Sleep(25 * time.Millisecond)
|
|
exec.release <- struct{}{}
|
|
wg.Wait()
|
|
close(errs)
|
|
for err := range errs {
|
|
if err != nil {
|
|
t.Fatalf("AppendAll() error = %v", err)
|
|
}
|
|
}
|
|
|
|
if got := exec.count("CREATE TABLE IF NOT EXISTS raw_"); got != 1 {
|
|
t.Fatalf("raw child CREATE count = %d, want 1; calls=%#v", got, exec.calls())
|
|
}
|
|
if got := exec.count("ALTER TABLE raw_"); got != 1 {
|
|
t.Fatalf("raw child ALTER count = %d, want 1; calls=%#v", got, exec.calls())
|
|
}
|
|
if got := exec.count("INSERT INTO raw_"); got != 2 {
|
|
t.Fatalf("raw insert count = %d, want 2; calls=%#v", got, exec.calls())
|
|
}
|
|
}
|
|
|
|
func sampleEnvelope() envelope.FrameEnvelope {
|
|
return envelope.FrameEnvelope{
|
|
Protocol: envelope.ProtocolJT808,
|
|
MessageID: "0x0200",
|
|
Sequence: 1,
|
|
VIN: "LNBVIN00000000001",
|
|
Phone: "013307795425",
|
|
SourceEndpoint: "127.0.0.1:18080",
|
|
EventTimeMS: 1782745114000,
|
|
ReceivedAtMS: 1782745114999,
|
|
RawHex: "7E0200",
|
|
Parsed: map[string]any{"message": "location"},
|
|
Fields: map[string]any{
|
|
envelope.FieldLongitude: 121.069881,
|
|
envelope.FieldLatitude: 30.590151,
|
|
envelope.FieldSpeedKMH: 23.0,
|
|
envelope.FieldTotalMileageKM: 10241.2,
|
|
"direction_deg": uint16(79),
|
|
"alarm_flag": uint32(0),
|
|
"status_flag": uint32(72),
|
|
},
|
|
ParseStatus: envelope.ParseOK,
|
|
}
|
|
}
|
|
|
|
func countSQL(calls []execCall, pattern string) int {
|
|
var count int
|
|
for _, call := range calls {
|
|
if strings.Contains(call.query, pattern) {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
func findSQL(calls []execCall, pattern string) string {
|
|
for _, call := range calls {
|
|
if strings.Contains(call.query, pattern) {
|
|
return call.query
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func containsSQL(calls []execCall, pattern string) bool {
|
|
return findSQL(calls, pattern) != ""
|
|
}
|
|
|
|
func between(value string, start string, end string) string {
|
|
startIndex := strings.Index(value, start)
|
|
if startIndex < 0 {
|
|
return ""
|
|
}
|
|
value = value[startIndex:]
|
|
endIndex := strings.Index(value, end)
|
|
if endIndex < 0 {
|
|
return value
|
|
}
|
|
return value[:endIndex]
|
|
}
|
|
|
|
type execCall struct {
|
|
query string
|
|
args []any
|
|
}
|
|
|
|
type recordingExec struct {
|
|
calls []execCall
|
|
}
|
|
|
|
func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
|
|
e.calls = append(e.calls, execCall{query: query, args: args})
|
|
return nil, nil
|
|
}
|
|
|
|
type blockingExec struct {
|
|
mu sync.Mutex
|
|
recorded []execCall
|
|
blockPattern string
|
|
blocked chan struct{}
|
|
release chan struct{}
|
|
blockOnce sync.Once
|
|
}
|
|
|
|
func newBlockingExec(blockPattern string) *blockingExec {
|
|
return &blockingExec{
|
|
blockPattern: blockPattern,
|
|
blocked: make(chan struct{}),
|
|
release: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func (e *blockingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
|
|
e.mu.Lock()
|
|
e.recorded = append(e.recorded, execCall{query: query, args: args})
|
|
e.mu.Unlock()
|
|
if strings.Contains(query, e.blockPattern) {
|
|
e.blockOnce.Do(func() {
|
|
close(e.blocked)
|
|
<-e.release
|
|
})
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
func (e *blockingExec) calls() []execCall {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
out := make([]execCall, len(e.recorded))
|
|
copy(out, e.recorded)
|
|
return out
|
|
}
|
|
|
|
func (e *blockingExec) count(pattern string) int {
|
|
return countSQL(e.calls(), pattern)
|
|
}
|