69 lines
2.8 KiB
TypeScript
69 lines
2.8 KiB
TypeScript
|
import { describe, expect, it } from 'vitest'
|
||
|
|
import type { Curve96 } from '@vpp/domain'
|
||
|
|
import { IngestionPipeline } from '../src/ingest.js'
|
||
|
|
import type { IngestSnapshot } from '../src/ingest.js'
|
||
|
|
import { MemorySnapshotStore } from '../src/snapshot.js'
|
||
|
|
import { MemoryTimeSeriesStore } from '../src/timeseries.js'
|
||
|
|
|
||
|
|
const curve = (fill: string): Curve96 => ({
|
||
|
|
interval_minutes: 15,
|
||
|
|
date: '2026-03-14',
|
||
|
|
values: Array.from({ length: 96 }, () => fill),
|
||
|
|
})
|
||
|
|
|
||
|
|
const clock = () => '2026-03-14T08:00:00Z'
|
||
|
|
|
||
|
|
const make = () => {
|
||
|
|
const timeseries = new MemoryTimeSeriesStore(clock)
|
||
|
|
const snapshots = new MemorySnapshotStore()
|
||
|
|
return { timeseries, snapshots, pipeline: new IngestionPipeline({ timeseries, snapshots, clock }) }
|
||
|
|
}
|
||
|
|
|
||
|
|
describe('ingestion pipeline skeleton', () => {
|
||
|
|
it('snapshots then stores a curve that passes the quality gate', () => {
|
||
|
|
const { timeseries, snapshots, pipeline } = make()
|
||
|
|
const result = pipeline.ingestCurve('load:agg-01', curve('12.5'), 'meter-file')
|
||
|
|
expect(result.accepted).toBe(true)
|
||
|
|
if (!result.accepted) throw new Error('unreachable')
|
||
|
|
expect(result.record.version).toBe(1)
|
||
|
|
expect(timeseries.latest('load:agg-01', '2026-03-14')).toEqual(result.record)
|
||
|
|
|
||
|
|
const snap = snapshots.get(result.snapshot_ref) as IngestSnapshot
|
||
|
|
expect(snap.series_id).toBe('load:agg-01')
|
||
|
|
expect(snap.source).toBe('meter-file')
|
||
|
|
expect(snap.quality.ok).toBe(true)
|
||
|
|
expect(snap.curve).toEqual(curve('12.5'))
|
||
|
|
expect(pipeline.quarantined()).toHaveLength(0)
|
||
|
|
})
|
||
|
|
|
||
|
|
it('quarantines a curve that fails the gate and keeps it out of the store', () => {
|
||
|
|
const { timeseries, snapshots, pipeline } = make()
|
||
|
|
const result = pipeline.ingestCurve('load:agg-01', curve('0'), 'edge')
|
||
|
|
expect(result.accepted).toBe(false)
|
||
|
|
if (result.accepted) throw new Error('unreachable')
|
||
|
|
expect(result.issues[0]).toMatch(/flat-zero/)
|
||
|
|
expect(timeseries.latest('load:agg-01', '2026-03-14')).toBeUndefined()
|
||
|
|
|
||
|
|
// Quarantined input is still evidence: tagged with its issues and snapshotted.
|
||
|
|
const q = pipeline.quarantined()
|
||
|
|
expect(q).toHaveLength(1)
|
||
|
|
expect(q[0]).toMatchObject({ series_id: 'load:agg-01', date: '2026-03-14', source: 'edge' })
|
||
|
|
const snap = snapshots.get(q[0]!.snapshot_ref) as IngestSnapshot
|
||
|
|
expect(snap.quality.ok).toBe(false)
|
||
|
|
})
|
||
|
|
|
||
|
|
it('accepts a custom gate so stricter checks can be plugged in later', () => {
|
||
|
|
const timeseries = new MemoryTimeSeriesStore(clock)
|
||
|
|
const snapshots = new MemorySnapshotStore()
|
||
|
|
const pipeline = new IngestionPipeline({
|
||
|
|
timeseries,
|
||
|
|
snapshots,
|
||
|
|
clock,
|
||
|
|
gate: () => ({ ok: false, issues: ['always rejected'] }),
|
||
|
|
})
|
||
|
|
const result = pipeline.ingestCurve('load:x', curve('1.0'), 'edge')
|
||
|
|
expect(result.accepted).toBe(false)
|
||
|
|
expect(timeseries.seriesIds()).toEqual([])
|
||
|
|
})
|
||
|
|
})
|