Adapt to test suite 3.0

slinkydeveloper · slinkydeveloper · commit e4a585688829 · 2025-03-04T18:00:16.000+01:00
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
@@ -105,7 +105,7 @@ jobs:
           cache-to: type=gha,mode=max,scope=${{ github.workflow }}
 
       - name: Run test tool
-        uses: restatedev/sdk-test-suite@v2.4
+        uses: restatedev/sdk-test-suite@v3.0
         with:
           restateContainerImage: ${{ inputs.restateCommit != '' && 'localhost/restatedev/restate-commit-download:latest' || (inputs.restateImage != '' && inputs.restateImage || 'ghcr.io/restatedev/restate:main') }}
           serviceContainerImage: "restatedev/rust-test-services"
diff --git a/test-services/README.md b/test-services/README.md
@@ -9,5 +9,5 @@ $ podman build -f test-services/Dockerfile -t restatedev/rust-test-services .
 To run (download the [sdk-test-suite](https://github.com/restatedev/sdk-test-suite) first):
 
 ```shell
-$ java -jar restate-sdk-test-suite.jar run restatedev/rust-test-services
+$ java -jar restate-sdk-test-suite.jar run localhost/restatedev/rust-test-services:latest
 ```
diff --git a/test-services/exclusions.yaml b/test-services/exclusions.yaml
@@ -1,14 +1,21 @@
 exclusions:
   "alwaysSuspending":
-    - "dev.restate.sdktesting.tests.AwaitTimeout"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwakeableTimeoutCommand"
+  - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable"
   "default":
-    - "dev.restate.sdktesting.tests.AwaitTimeout"
-    - "dev.restate.sdktesting.tests.RawHandler"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwakeableTimeoutCommand"
+  - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable"
   "singleThreadSinglePartition":
-    - "dev.restate.sdktesting.tests.AwaitTimeout"
-    - "dev.restate.sdktesting.tests.RawHandler"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwakeableTimeoutCommand"
+  - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable"
   "threeNodes":
-    - "dev.restate.sdktesting.tests.AwaitTimeout"
-    - "dev.restate.sdktesting.tests.RawHandler"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwakeableTimeoutCommand"
+  - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable"
   "threeNodesAlwaysSuspending":
-    - "dev.restate.sdktesting.tests.AwaitTimeout"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
+  - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwakeableTimeoutCommand"
+  - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable"
diff --git a/test-services/src/main.rs b/test-services/src/main.rs
@@ -9,6 +9,7 @@ mod map_object;
 mod non_deterministic;
 mod proxy;
 mod test_utils_service;
+mod virtual_object_command_interpreter;
 
 use restate_sdk::prelude::{Endpoint, HttpServer};
 use std::env;
@@ -76,6 +77,13 @@ async fn main() {
             test_utils_service::TestUtilsServiceImpl,
         ))
     }
+    if services == "*" || services.contains("VirtualObjectCommandInterpreter") {
+        builder = builder.bind(
+            virtual_object_command_interpreter::VirtualObjectCommandInterpreter::serve(
+                virtual_object_command_interpreter::VirtualObjectCommandInterpreterImpl,
+            ),
+        )
+    }
 
     if let Ok(key) = env::var("E2E_REQUEST_SIGNING_ENV") {
         builder = builder.identity_key(&key).unwrap()
diff --git a/test-services/src/proxy.rs b/test-services/src/proxy.rs
@@ -11,6 +11,7 @@ pub(crate) struct ProxyRequest {
     service_name: String,
     virtual_object_key: Option<String>,
     handler_name: String,
+    idempotency_key: Option<String>,
     message: Vec<u8>,
     delay_millis: Option<u64>,
 }
@@ -59,19 +60,22 @@ impl Proxy for ProxyImpl {
         ctx: Context<'_>,
         Json(req): Json<ProxyRequest>,
     ) -> HandlerResult<Json<Vec<u8>>> {
-        Ok(ctx
-            .request::<Vec<u8>, Vec<u8>>(req.to_target(), req.message)
-            .call()
-            .await?
-            .into())
+        let mut request = ctx.request::<Vec<u8>, Vec<u8>>(req.to_target(), req.message);
+        if let Some(idempotency_key) = req.idempotency_key {
+            request = request.idempotency_key(idempotency_key);
+        }
+        Ok(request.call().await?.into())
     }
 
     async fn one_way_call(
         &self,
         ctx: Context<'_>,
         Json(req): Json<ProxyRequest>,
     ) -> HandlerResult<String> {
-        let request = ctx.request::<_, ()>(req.to_target(), req.message);
+        let mut request = ctx.request::<_, ()>(req.to_target(), req.message);
+        if let Some(idempotency_key) = req.idempotency_key {
+            request = request.idempotency_key(idempotency_key);
+        }
 
         let invocation_id = if let Some(delay_millis) = req.delay_millis {
             request
@@ -93,8 +97,11 @@ impl Proxy for ProxyImpl {
         let mut futures: Vec<BoxFuture<'_, Result<Vec<u8>, TerminalError>>> = vec![];
 
         for req in requests {
-            let restate_req =
+            let mut restate_req =
                 ctx.request::<_, Vec<u8>>(req.proxy_request.to_target(), req.proxy_request.message);
+            if let Some(idempotency_key) = req.proxy_request.idempotency_key {
+                restate_req = restate_req.idempotency_key(idempotency_key);
+            }
             if req.one_way_call {
                 if let Some(delay_millis) = req.proxy_request.delay_millis {
                     restate_req.send_after(Duration::from_millis(delay_millis));
diff --git a/test-services/src/test_utils_service.rs b/test-services/src/test_utils_service.rs
@@ -1,71 +1,29 @@
-use crate::awakeable_holder;
-use crate::list_object::ListObjectClient;
 use futures::future::BoxFuture;
 use futures::FutureExt;
 use restate_sdk::prelude::*;
-use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::convert::Infallible;
 use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-#[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub(crate) struct CreateAwakeableAndAwaitItRequest {
-    awakeable_key: String,
-    await_timeout: Option<u64>,
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(tag = "type")]
-#[serde(rename_all_fields = "camelCase")]
-pub(crate) enum CreateAwakeableAndAwaitItResponse {
-    #[serde(rename = "timeout")]
-    Timeout,
-    #[serde(rename = "result")]
-    Result { value: String },
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub(crate) struct InterpretRequest {
-    list_name: String,
-    commands: Vec<InterpretCommand>,
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(tag = "type")]
-#[serde(rename_all_fields = "camelCase")]
-pub(crate) enum InterpretCommand {
-    #[serde(rename = "createAwakeableAndAwaitIt")]
-    CreateAwakeableAndAwaitIt { awakeable_key: String },
-    #[serde(rename = "getEnvVariable")]
-    GetEnvVariable { env_name: String },
-}
-
 #[restate_sdk::service]
 #[name = "TestUtilsService"]
 pub(crate) trait TestUtilsService {
     #[name = "echo"]
     async fn echo(input: String) -> HandlerResult<String>;
     #[name = "uppercaseEcho"]
     async fn uppercase_echo(input: String) -> HandlerResult<String>;
+    #[name = "rawEcho"]
+    async fn raw_echo(input: Vec<u8>) -> Result<Vec<u8>, Infallible>;
     #[name = "echoHeaders"]
     async fn echo_headers() -> HandlerResult<Json<HashMap<String, String>>>;
-    #[name = "createAwakeableAndAwaitIt"]
-    async fn create_awakeable_and_await_it(
-        req: Json<CreateAwakeableAndAwaitItRequest>,
-    ) -> HandlerResult<Json<CreateAwakeableAndAwaitItResponse>>;
     #[name = "sleepConcurrently"]
     async fn sleep_concurrently(millis_durations: Json<Vec<u64>>) -> HandlerResult<()>;
     #[name = "countExecutedSideEffects"]
     async fn count_executed_side_effects(increments: u32) -> HandlerResult<u32>;
-    #[name = "getEnvVariable"]
-    async fn get_env_variable(env: String) -> HandlerResult<String>;
     #[name = "cancelInvocation"]
     async fn cancel_invocation(invocation_id: String) -> Result<(), TerminalError>;
-    #[name = "interpretCommands"]
-    async fn interpret_commands(req: Json<InterpretRequest>) -> HandlerResult<()>;
 }
 
 pub(crate) struct TestUtilsServiceImpl;
@@ -79,6 +37,10 @@ impl TestUtilsService for TestUtilsServiceImpl {
         Ok(input.to_ascii_uppercase())
     }
 
+    async fn raw_echo(&self, _: Context<'_>, input: Vec<u8>) -> Result<Vec<u8>, Infallible> {
+        Ok(input)
+    }
+
     async fn echo_headers(
         &self,
         context: Context<'_>,
@@ -94,27 +56,6 @@ impl TestUtilsService for TestUtilsServiceImpl {
         Ok(headers.into())
     }
 
-    async fn create_awakeable_and_await_it(
-        &self,
-        context: Context<'_>,
-        Json(req): Json<CreateAwakeableAndAwaitItRequest>,
-    ) -> HandlerResult<Json<CreateAwakeableAndAwaitItResponse>> {
-        if req.await_timeout.is_some() {
-            unimplemented!("await timeout is not yet implemented");
-        }
-
-        let (awk_id, awakeable) = context.awakeable::<String>();
-
-        context
-            .object_client::<awakeable_holder::AwakeableHolderClient>(req.awakeable_key)
-            .hold(awk_id)
-            .call()
-            .await?;
-        let value = awakeable.await?;
-
-        Ok(CreateAwakeableAndAwaitItResponse::Result { value }.into())
-    }
-
     async fn sleep_concurrently(
         &self,
         context: Context<'_>,
@@ -153,10 +94,6 @@ impl TestUtilsService for TestUtilsServiceImpl {
         Ok(counter.load(Ordering::SeqCst) as u32)
     }
 
-    async fn get_env_variable(&self, _: Context<'_>, env: String) -> HandlerResult<String> {
-        Ok(std::env::var(env).ok().unwrap_or_default())
-    }
-
     async fn cancel_invocation(
         &self,
         ctx: Context<'_>,
@@ -165,34 +102,4 @@ impl TestUtilsService for TestUtilsServiceImpl {
         ctx.invocation_handle(invocation_id).cancel().await?;
         Ok(())
     }
-
-    async fn interpret_commands(
-        &self,
-        context: Context<'_>,
-        Json(req): Json<InterpretRequest>,
-    ) -> HandlerResult<()> {
-        let list_client = context.object_client::<ListObjectClient>(req.list_name);
-
-        for cmd in req.commands {
-            match cmd {
-                InterpretCommand::CreateAwakeableAndAwaitIt { awakeable_key } => {
-                    let (awk_id, awakeable) = context.awakeable::<String>();
-                    context
-                        .object_client::<awakeable_holder::AwakeableHolderClient>(awakeable_key)
-                        .hold(awk_id)
-                        .call()
-                        .await?;
-                    let value = awakeable.await?;
-                    list_client.append(value).send();
-                }
-                InterpretCommand::GetEnvVariable { env_name } => {
-                    list_client
-                        .append(std::env::var(env_name).ok().unwrap_or_default())
-                        .send();
-                }
-            }
-        }
-
-        Ok(())
-    }
 }
diff --git a/test-services/src/virtual_object_command_interpreter.rs b/test-services/src/virtual_object_command_interpreter.rs
