Binary data converter (#90)

Author: bergundy

.github/workflows/python.yaml

@@ -22,6 +22,9 @@ jobs:
         run: 'echo head_ref: ${{ github.head_ref }}, ref: ${{ github.ref }}, python repo ref: ${{ inputs.python-repo-ref }}'
         working-directory: '.'
 
+      - name: Install Protoc
+        uses: arduino/setup-protoc@v1
+
       - name: Checkout SDK features repo
         uses: actions/checkout@v2
         with:

.gitignore

@@ -14,4 +14,11 @@ node_modules
 /tslib/
 
 # Python stuff
-__pycache__
+__pycache__
+
+# Build Go stuff
+bin
+
+# VS Code config
+.vscode
+

README.md

@@ -5,11 +5,11 @@ contains a runner and language-specific harnesses to confirm feature behavior ac
 
 These SDK features serve several purposes:
 
-* Ensure parity across SDKs by having same-feature snippets adjacent to one another
-* Confirm feature behavior across SDK versions
-* Confirm history across SDK versions
-* Document features in different SDKs
-* Easy-to-use environment for writing quick workflows in all languages/versions
+- Ensure parity across SDKs by having same-feature snippets adjacent to one another
+- Confirm feature behavior across SDK versions
+- Confirm history across SDK versions
+- Document features in different SDKs
+- Easy-to-use environment for writing quick workflows in all languages/versions
 
 ## Building
 
@@ -87,38 +87,42 @@ the `features/` directory.
 In addition to code for the feature, there are configuration settings that can be set in `.config.json`. The possible
 settings are:
 
-* `go`
-  * `minVersion` - Minimum version in Go this feature should be run in. The feature will be skipped in older versions.
+- `go`
+  - `minVersion` - Minimum version in Go this feature should be run in. The feature will be skipped in older versions.
 
 There are also files in the `history/` subdirectory which contain history files used during run. See the
 "History Checking" and "Generating History" sections for more info.
 
 ### Best Practices
 
-* Try to only demonstrate/test one feature per feature directory.
-  * Code should be kept as short and clear as possible.
-  * No need to over-assert on a bunch of values, just confirm that the feature does what is expected via its output.
-* A Go feature should be in `feature.go`.
-  * For incompatible versions, different files like `feature_pre1.11.0.go` can be present using build tags
-* A Java feature should be in `feature.java`.
-* A TypeScript feature should be in `feature.ts` for all non-workflow code and `feature.workflow.ts` for all workflow
-  code.
-* A Python feature should be in `feature.py`.
-* Add a README.md to each feature directory.
-  * README should have a title summarizing the feature (only first letter needs to be in title case), then a short
+- Try to only demonstrate/test one feature per feature directory.
+  - Code should be kept as short and clear as possible.
+  - No need to over-assert on a bunch of values, just confirm that the feature does what is expected via its output.
+- A Go feature should be in `feature.go`.
+  - For incompatible versions, different files like `feature_pre1.11.0.go` can be present using build tags
+- A Java feature should be in `feature.java`.
+- A TypeScript feature should be in `feature.ts`.
+
+  **NOTE**: TypeScript features include workflow and non workflow code in the same file. Those are run in different
+  environments so they may not share variables and the feature author should keep the workflow runtime limitations in min
+  mind when writing features.
+
+- A Python feature should be in `feature.py`.
+- Add a README.md to each feature directory.
+  - README should have a title summarizing the feature (only first letter needs to be in title case), then a short
     paragraph explaining the feature and its purpose, and then optionally another paragraph explaining details of the
     specific code steps.
-  * Other sections can also explain decisions made in language specific ways or other details about versions/approaches.
-  * Feel free to add links and more text as necessary.
-* Verification/regression feature directories for bugs should be under `features/bugs/<lang>`.
-  * Ideally the checking of the result has a version condition that shows in earlier versions it should fail and in
+  - Other sections can also explain decisions made in language specific ways or other details about versions/approaches.
+  - Feel free to add links and more text as necessary.
+- Verification/regression feature directories for bugs should be under `features/bugs/<lang>`.
+  - Ideally the checking of the result has a version condition that shows in earlier versions it should fail and in
     newer versions it should succeed.
-* The more languages per non-bug feature, the better. Try not to create non-bug features that use specific language
+- The more languages per non-bug feature, the better. Try not to create non-bug features that use specific language
   constructs unless that's the purpose of the feature.
-* Refactor liberally to create shortcuts and better harnesses for making features easy to read and write.
-  * This is not a library used by anyone, there are no backwards compatibility concerns. If one feature uses something
+- Refactor liberally to create shortcuts and better harnesses for making features easy to read and write.
+  - This is not a library used by anyone, there are no backwards compatibility concerns. If one feature uses something
     that has value to another, extract and put in helper/harness and have both use it.
-* History should be generated for each feature on the earliest version the feature is expected to work at without
+- History should be generated for each feature on the earliest version the feature is expected to work at without
   history changes.
 
 #### Generating History
@@ -132,15 +136,16 @@ incompatibility. Otherwise, history files should remain checked in and not regen
 
 ## TODO
 
-* Add support for replaying testing of all versions _inside_ each SDKs harness as part of the run
-* Add TypeScript support
-  * The main support is present, but there are outstanding questions on what constitutes a "version" since really
+- Add support for replaying testing of all versions _inside_ each SDKs harness as part of the run
+- Add TypeScript support
+  - The main support is present, but there are outstanding questions on what constitutes a "version" since really
     TypeScript has many versions
-* Add many more feature workflows
-* Document how to use this framework to easily write and test features even when not committing
-* Log swallowing and concurrent execution
-* Investigate support for changing runtime versions (i.e. Go, Java, and Node versions)
-* Investigate support for changing server versions
-* CI support
-  * Support using a commit hash and alternative git location for an SDK to run against
-  * Decide whether the matrix of different SDK versions and such is really part of this repo or part of CI tooling
+- Add many more feature workflows
+- Document how to use this framework to easily write and test features even when not committing
+- Log swallowing and concurrent execution
+- Investigate support for changing runtime versions (i.e. Go, Java, and Node versions)
+- Investigate support for changing server versions
+- CI support
+  - Support using a commit hash and alternative git location for an SDK to run against
+  - Decide whether the matrix of different SDK versions and such is really part of this repo or part of CI tooling
+

features/activity/cancel_try_cancel/feature.ts

@@ -1,10 +1,11 @@
 import { Context } from '@temporalio/activity';
-import { WorkflowClient } from '@temporalio/client';
-import { CancelledFailure, ActivityFailure, ApplicationFailure } from '@temporalio/common';
-import { Feature } from '@temporalio/harness';
+import { CancelledFailure } from '@temporalio/common';
+import { Feature, getWorkflowClient } from '@temporalio/harness';
+import { ActivityFailure, ApplicationFailure } from '@temporalio/common';
 import * as wf from '@temporalio/workflow';
 
-const { cancellableActivity } = wf.proxyActivities<typeof activitiesImpl>({
+// Allow 4 retries with no backoff
+const activities = wf.proxyActivities<typeof activitiesImpl>({
   startToCloseTimeout: '1 minute',
   heartbeatTimeout: '5 seconds',
   // Disable retry
@@ -18,7 +19,7 @@ export async function workflow(): Promise<void> {
   try {
     await wf.CancellationScope.cancellable(async () => {
       // Start activity
-      const actPromise = cancellableActivity();
+      const actPromise = activities.cancellableActivity();
 
       // Sleep for smallest amount of time (force task turnover)
       await wf.sleep(1);
@@ -43,14 +44,9 @@ export async function workflow(): Promise<void> {
   }
 }
 
-let client: WorkflowClient | undefined;
-
 const activitiesImpl = {
   async cancellableActivity() {
-    // Expect client to be set
-    if (!client) {
-      throw new Error('Missing client');
-    }
+    const client = getWorkflowClient();
 
     // Heartbeat every second for a minute
     let result = 'timeout';
@@ -76,13 +72,7 @@ const activitiesImpl = {
   },
 };
 
-export const feature =
-  !wf.inWorkflowContext() &&
-  new Feature({
-    workflow,
-    activities: activitiesImpl,
-    execute: async (runner) => {
-      client = runner.client;
-      return await runner.executeSingleParameterlessWorkflow();
-    },
-  });
+export const feature = new Feature({
+  workflow,
+  activities: activitiesImpl,
+});

features/activity/retry_on_error/feature.ts

@@ -28,19 +28,17 @@ const activitiesImpl = {
   },
 };
 
-export const feature =
-  !wf.inWorkflowContext() &&
-  new Feature({
-    workflow,
-    activities: activitiesImpl,
-    checkResult: async (runner, handle) => {
-      await assert.rejects(runner.waitForRunResult(handle), (err) => {
-        assert.ok(
-          err instanceof WorkflowFailedError,
-          `expected WorkflowFailedError, got ${typeof err}, message: ${(err as any).message}`
-        );
-        assert.equal(err.cause?.cause?.message, 'activity attempt 5 failed');
-        return true;
-      });
-    },
-  });
+export const feature = new Feature({
+  workflow,
+  activities: activitiesImpl,
+  checkResult: async (runner, handle) => {
+    await assert.rejects(runner.waitForRunResult(handle), (err) => {
+      assert.ok(
+        err instanceof WorkflowFailedError,
+        `expected WorkflowFailedError, got ${typeof err}, message: ${(err as any).message}`
+      );
+      assert.equal(err.cause?.cause?.message, 'activity attempt 5 failed');
+      return true;
+    });
+  },
+});

features/data_converter/binary/README.md

@@ -2,12 +2,13 @@
 
 Binary values can be converted to and from `binary/plain` Payloads.
 
-This feature: 
+Steps:
 
-- runs the binary value `101` (5) through the default Payload Converter, writes it to `payloads/binary.[lang]`, and
-  verifies it matches the other files in `payloads/`
-- decodes all files in `payloads/` with the default Payload Converter and verifies the binary value is `101`
+- run a workflow that returns binary value `0xdeadbeef`
+- verify client result is binary `0xdeadbeef`
+- get result payload of WorkflowExecutionCompleted event from workflow history
+- load JSON payload from `./payload.json` and compare it to result payload
 
 # Detailed spec
 
-`metadata.encoding = toBinary("binary/plain")`
+`metadata.encoding = toBinary("binary/plain")`

features/data_converter/binary/feature.go

@@ -0,0 +1,74 @@
+package binary
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path"
+
+	"github.com/gogo/protobuf/jsonpb"
+	common "go.temporal.io/api/common/v1"
+	historyProto "go.temporal.io/api/history/v1"
+	"go.temporal.io/sdk-features/harness/go/harness"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/workflow"
+)
+
+var EXPECTED_RESULT = []byte{0xde, 0xad, 0xbe, 0xef}
+
+var Feature = harness.Feature{
+	Workflows:   Workflow,
+	CheckResult: CheckResult,
+}
+
+// run a workflow that returns binary value `0xdeadbeef`
+func Workflow(ctx workflow.Context) ([]byte, error) {
+	return EXPECTED_RESULT, nil
+}
+
+func CheckResult(ctx context.Context, runner *harness.Runner, run client.WorkflowRun) error {
+	// verify client result is binary `0xdeadbeef`
+	result := make([]byte, 4)
+	if err := run.Get(ctx, &result); err != nil {
+		return err
+	}
+	if !bytes.Equal(result, EXPECTED_RESULT) {
+		return fmt.Errorf("invalid result: %v", result)
+	}
+	history := runner.Client.GetWorkflowHistory(ctx, run.GetID(), "", false, 0)
+
+	var attrs *historyProto.WorkflowExecutionCompletedEventAttributes
+
+	for history.HasNext() {
+		ev, err := history.Next()
+		if err != nil {
+			return err
+		}
+		// get result payload of WorkflowExecutionCompleted event from workflow history
+		attrs = ev.GetWorkflowExecutionCompletedEventAttributes()
+		if attrs != nil {
+			break
+		}
+	}
+	if attrs == nil {
+		return errors.New("could not locate WorkflowExecutionCompleted event")
+	}
+	payload := attrs.GetResult().GetPayloads()[0]
+
+	// load JSON payload from `./payload.json` and compare it to result payload
+	file, err := os.Open(path.Join(runner.Feature.AbsDir, "../../../features/data_converter/binary/payload.json"))
+	if err != nil {
+		return err
+	}
+
+	expectedPayload := &common.Payload{}
+	unmarshaler := jsonpb.Unmarshaler{}
+	err = unmarshaler.Unmarshal(file, expectedPayload)
+	if err != nil {
+		return err
+	}
+	runner.Require.Equal(expectedPayload, payload)
+	return nil
+}

features/data_converter/binary/feature.java

@@ -0,0 +1,53 @@
+package data_converter.binary;
+
+import io.temporal.api.common.v1.Payload;
+import io.temporal.sdkfeatures.Feature;
+import io.temporal.sdkfeatures.Run;
+import io.temporal.sdkfeatures.Runner;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+import com.google.protobuf.util.JsonFormat;
+
+@WorkflowInterface
+public interface feature extends Feature {
+    static byte[] deadbeef = new byte[]{(byte)0xde, (byte)0xad, (byte)0xbe, (byte)0xef};
+    
+  @WorkflowMethod
+  public byte[] workflow();
+
+  class Impl implements feature {
+    /**
+     * run a workflow that returns binary value `0xdeadbeef`
+     */
+    @Override
+    public byte[] workflow() {
+        return deadbeef;
+    }
+
+    @Override
+    public void checkResult(Runner runner, Run run) throws Exception {
+      // verify client result is binary `0xdeadbeef`
+      var result = runner.waitForRunResult(run, byte[].class);
+      assertArrayEquals(deadbeef, result);
+
+      // get result payload of WorkflowExecutionCompleted event from workflow history
+      var history = runner.getWorkflowHistory(run);
+      var event = history.getEventsList().stream().filter(e -> e.hasWorkflowExecutionCompletedEventAttributes()).findFirst();
+      var payload = event.get().getWorkflowExecutionCompletedEventAttributes().getResult().getPayloads(0);
+      
+      // load JSON payload from `./payload.json` and compare it to JSON representation of result payload
+      var content = Files.readAllBytes(Paths.get(System.getProperty("user.dir"), "..", "features", runner.featureInfo.dir, "payload.json"));
+      var builder = Payload.newBuilder();
+      JsonFormat.parser().merge(new String(content), builder);
+      var expected = builder.build();
+      assertEquals(expected, payload);
+    }
+  }
+}