fixup: add conformance test

Signed-off-by: Lance Ball <[email protected]>

Author: lance

package.json

@@ -12,7 +12,7 @@
     "lint:fix": "eslint 'src/**/*.{js,ts}' 'test/**/*.{js,ts}' --fix",
     "apretest": "npm run lint && npm run conformance",
     "test": "mocha --require ts-node/register ./test/integration/**/kafka_tests.ts",
-    "conformance": "cucumber-js ./conformance/features/http-protocol-binding.feature -p default",
+    "conformance": "cucumber-js ./conformance/features/*-protocol-binding.feature -p default",
     "coverage": "nyc --reporter=lcov --reporter=text npm run test",
     "coverage-publish": "wget -qO - https://coverage.codacy.com/get.sh | bash -s report -l JavaScript -r coverage/lcov.info",
     "generate-docs": "typedoc --excludeNotDocumented --out docs src",

src/message/kafka/index.ts

@@ -175,10 +175,15 @@ function parseBinary<T>(message: KafkaMessage<T>): CloudEvent<T> {
   const eventObj: { [key: string ]: unknown } = {};
   const headers = { ...message.headers };
 
+  eventObj.datacontenttype = headers[CONSTANTS.HEADER_CONTENT_TYPE];
+
   for (const key in KAFKA_CE_HEADERS) {
     const h = KAFKA_CE_HEADERS[key];
     if (!!headers[h]) {
       eventObj[HEADER_MAP[h]] = headers[h];
+      if (h === KAFKA_CE_HEADERS.TIME) {
+        eventObj.time = new Date(eventObj.time as string).toISOString();
+      }
       delete headers[h];
     }
   }
@@ -211,9 +216,10 @@ function parseStructured<T>(message: KafkaMessage<T>): CloudEvent<T> {
   if (!message.headers[CONSTANTS.HEADER_CONTENT_TYPE]?.startsWith(CONSTANTS.MIME_CE_JSON)) {
     throw new ValidationError(`Unsupported event encoding ${message.headers[CONSTANTS.HEADER_CONTENT_TYPE]}`);
   }
-
+  const eventObj = JSON.parse(message.value as string);
+  eventObj.time = new Date(eventObj.time).toISOString();
   return new CloudEvent({
-    ...JSON.parse(message.value as string),
+    ...eventObj,
     partitionkey: message.key,
   }, false);
 }
@@ -242,7 +248,7 @@ function parseBatched<T>(message: KafkaMessage<T>): CloudEvent<T>[] {
 function extractBinaryData<T>(message: KafkaMessage<T>): T {
   let data = message.value as T;
   // If the event data is JSON, go ahead and parse it
-  const datacontenttype = message.headers[KAFKA_CE_HEADERS.DATACONTENTTYPE] as string;
+  const datacontenttype = message.headers[CONSTANTS.HEADER_CONTENT_TYPE] as string;
   if (!!datacontenttype && datacontenttype.startsWith(CONSTANTS.MIME_JSON)) {
     if (typeof message.value === "string") {
       data = JSON.parse(message.value);

test/conformance/steps.ts

@@ -3,17 +3,41 @@
  SPDX-License-Identifier: Apache-2.0
 */
 
-/* eslint-disable @typescript-eslint/ban-ts-comment */
-
 import { assert } from "chai";
 import { Given, When, Then, World } from "@cucumber/cucumber";
-import { Message, Headers, HTTP } from "../../src";
+import { Message, Headers, HTTP, KafkaMessage, Kafka } from "../../src";
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const { HTTPParser } = require("http-parser-js");
 
 const parser = new HTTPParser(HTTPParser.REQUEST);
 
+Given("Kafka Protocol Binding is supported", function (this: World) {
+  return true;
+});
+
+Given("a Kafka message with payload:", function (request: string) {
+  // Create a KafkaMessage from the incoming HTTP request
+  const value = Buffer.from(request);
+  const message: KafkaMessage = {
+    key: "",
+    headers: {},
+    body: value,
+    value,
+  };
+  this.message = message;
+  return true;
+});
+
+Then("Kafka headers:", function (attributes: { rawTable: [] }) {
+  this.message.headers = tableToObject(attributes.rawTable);
+});
+
+When("parsed as Kafka message", function () {
+  this.cloudevent = Kafka.toEvent(this.message);
+  return true;
+});
+
 Given("HTTP Protocol Binding is supported", function (this: World) {
   return true;
 });