NODEDC_PLATFORM/services/external-data-plane/test/data-product-delivery.test.mjs

import assert from "node:assert/strict";

import {
  assertPublishMatchesDefinition,
  classifyReplacementGeneration,
} from "../src/data-product-delivery.mjs";
import { normalizeDataProductDefinition } from "../src/data-product-policy.mjs";

const definition = normalizeDataProductDefinition({
  id: "test.positions.current.v2",
  version: "2.0.0",
  ontologyRevision: "ontology.test.positions.v1",
  deliveryMode: "snapshot+patch",
  semanticTypes: ["map.moving_object"],
  fields: ["display_name", "geometry", "movement_state", "signal_state", "speed_kph"],
  fieldContracts: {
    display_name: { type: "string", required: true },
    geometry: { type: "point", required: false },
    movement_state: { type: "string", required: true, enum: ["moving", "stopped"] },
    signal_state: { type: "string", required: true, enum: ["active", "inactive"] },
    speed_kph: { type: "number", required: false, minimum: 0 },
  },
  history: { mode: "none", retentionDays: 1 },
});

const fact = {
  sourceId: "unit-1",
  semanticType: "map.moving_object",
  observedAt: "2026-07-20T10:00:00.000Z",
  geometry: { type: "Point", coordinates: [37.6173, 55.7558] },
  attributes: {
    display_name: "Unit 1",
    movement_state: "moving",
    signal_state: "active",
    speed_kph: 12,
  },
};
const batch = { facts: [fact] };
assert.doesNotThrow(() => assertPublishMatchesDefinition(batch, definition));

assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{ ...fact, attributes: { ...fact.attributes, movement_state: "teleporting" } }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_value_forbidden",
);
assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{ ...fact, attributes: { ...fact.attributes, signal_state: "invented" } }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_value_forbidden",
);
assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{
      ...fact,
      attributes: Object.fromEntries(Object.entries(fact.attributes).filter(([field]) => field !== "signal_state")),
    }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_required",
);
assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{ ...fact, attributes: { ...fact.attributes, speed_kph: "12" } }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_type_invalid",
);

const telemetryDefinition = normalizeDataProductDefinition({
  id: "test.telemetry.current.v1",
  version: "1.0.0",
  ontologyRevision: "ontology.test.telemetry.v1",
  deliveryMode: "snapshot+patch",
  semanticTypes: ["map.moving_object"],
  fields: ["sensor_readings"],
  fieldContracts: {
    sensor_readings: { type: "telemetry_readings", required: false },
  },
  history: { mode: "none", retentionDays: 1 },
});
const validReadings = [{
  id: "battery.voltage",
  label: "Battery voltage",
  value: 51.7,
  unit: "V",
  observedAt: "2026-07-22T08:00:00.000Z",
}];
assert.doesNotThrow(() => assertPublishMatchesDefinition({ facts: [{
  sourceId: "unit-telemetry",
  semanticType: "map.moving_object",
  observedAt: "2026-07-22T08:00:00.000Z",
  attributes: { sensor_readings: validReadings },
}] }, telemetryDefinition));
assert.throws(
  () => assertPublishMatchesDefinition({ facts: [{
    sourceId: "unit-telemetry",
    semanticType: "map.moving_object",
    attributes: { sensor_readings: [{ ...validReadings[0], value: { raw: 51.7 } }] },
  }] }, telemetryDefinition),
  (error) => error?.status === 422 && error?.code === "data_product_field_type_invalid",
);
assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{ ...fact, attributes: { ...fact.attributes, speed_kph: -1 } }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_value_out_of_range",
);
assert.throws(
  () => assertPublishMatchesDefinition({
    facts: [{ ...fact, geometry: { type: "Point", coordinates: [181, 55.7558] } }],
  }, definition),
  (error) => error?.status === 422 && error?.code === "data_product_field_type_invalid",
);

const legacyDefinition = normalizeDataProductDefinition({
  id: "test.positions.current.v1",
  version: "1.0.0",
  ontologyRevision: "ontology.test.positions.v1",
  deliveryMode: "snapshot+patch",
  semanticTypes: ["map.moving_object"],
  fields: ["movement_state", "signal_state"],
  history: { mode: "none", retentionDays: 1 },
});
assert.doesNotThrow(() => assertPublishMatchesDefinition({
  facts: [{
    sourceId: "unit-legacy",
    semanticType: "map.moving_object",
    attributes: { movement_state: "teleporting", signal_state: "invented" },
  }],
}, legacyDefinition));

const acceptedGeneration = "2026-07-15T10:10:00.000Z";
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: null,
  generationAt: acceptedGeneration,
}), "advance");
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: acceptedGeneration,
  generationAt: "2026-07-15T10:11:00.000Z",
}), "advance");
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: acceptedGeneration,
  generationAt: acceptedGeneration,
  incomingCount: 903,
  currentCount: 903,
  matchingCount: 903,
}), "replay");
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: acceptedGeneration,
  generationAt: acceptedGeneration,
  incomingCount: 903,
  currentCount: 903,
  matchingCount: 902,
}), "conflict");
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: acceptedGeneration,
  generationAt: acceptedGeneration,
  incomingCount: 902,
  currentCount: 903,
  matchingCount: 902,
}), "conflict");
assert.equal(classifyReplacementGeneration({
  currentGenerationAt: acceptedGeneration,
  generationAt: "2026-07-15T10:09:00.000Z",
  incomingCount: 903,
  currentCount: 903,
  matchingCount: 903,
}), "stale");

console.log("external-data-plane data product delivery policy: ok");