Rust dial test (#39)

Author: maximpertsov

.github/workflows/test.yml

@@ -0,0 +1,28 @@
+name: Run tests
+
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  test:
+    if: github.repository_owner == 'viamrobotics'
+    runs-on: [self-hosted, x64]
+    container:
+      image: ghcr.io/viamrobotics/canon:amd64
+    steps:
+      - name: Checkout PR/Push/Workflow Dispatch
+        uses: actions/checkout@v3
+      - name: Setup Rust toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+      - name: Run tests
+        run: |
+          apt-get -y install lsof
+          make test

.gitignore

@@ -3,3 +3,9 @@ target/
 **/*.rs.bk
 
 *.DS_Store
+
+# make tools bin
+bin/
+
+# goutils
+tests/goutils/

Cargo.toml

@@ -44,8 +44,14 @@ tower-http = { version = "0.3.3", features = ["add-extension","auth","propagate-
 tracing = {version = "0.1.34"}
 tracing-subscriber = {version = "0.3.11", features = ["env-filter"]}
 webpki-roots = "0.21.1"
-webrtc = "0.7.1"
+# TODO: We are using a commit hash to include a bug fix that has not yet been
+# released in a crate. Once the new crate is released, please use that instead
+# of the git revision below. As of this comment the latest version is `0.7.1`.
+webrtc = { git = "https://github.com/webrtc-rs/webrtc.git", rev = "5aa49c03a183a610b44fe01e9531508e4fddecb1" }
 
+[dev-dependencies]
+async-stream = "0.3.3"
+env_logger = "0.9.0"
 
 [build-dependencies]
 tonic-build = {version = "0.7.2",features = ["prost", "compression"]}

Makefile

@@ -1,10 +1,21 @@
+BUF_BIN="`pwd`/bin"
+PATH_WITH_TOOLS="${BUF_BIN}:${PATH}"
+
 all: build build-example
 build:
 	cargo build
 build-example:
 	cd examples/ && cargo build
 buf-clean:
 	find src/gen -type f \( -iname "*.rs" ! -iname "mod.rs" \) -delete
-buf:	buf-clean
-	buf generate buf.build/viamrobotics/goutils --template buf.gen.yaml
-	buf generate buf.build/googleapis/googleapis --template buf.gen.yaml --path google/rpc --path google/api
+buf-install:
+	./etc/install_buf.sh $(BUF_BIN)
+buf:	buf-install buf-clean
+	PATH=${PATH_WITH_TOOLS} buf generate buf.build/viamrobotics/goutils --template buf.gen.yaml
+	PATH=${PATH_WITH_TOOLS} buf generate buf.build/googleapis/googleapis --template buf.gen.yaml --path google/rpc --path google/api
+tests/goutils:
+	git clone --depth=1 https://github.com/viamrobotics/goutils.git tests/goutils
+test: buf tests/goutils build
+	./etc/test.sh
+test-server: tests/goutils
+	./etc/run_echo_server.sh 8081

etc/install_buf.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+# Substitute BIN for your bin directory.
+# Substitute VERSION for the current released version.
+
+if [[ -z "$1" ]]; then
+    echo "please provide an installation direction"
+    exit 1
+fi
+
+BIN="$1"
+VERSION="1.16.0"
+
+mkdir -p "$BIN"
+curl -sSL \
+    "https://github.com/bufbuild/buf/releases/download/v${VERSION}/buf-$(uname -s)-$(uname -m)" \
+    -o "${BIN}/buf"
+chmod +x "${BIN}/buf"

etc/run_echo_server.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+cd tests/goutils/rpc/examples/echo || exit 1
+
+go run server/cmd/main.go -instance-name="localhost:$1" "$1"

etc/test.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+SERVER_PORT=8081
+
+if lsof -i ":$SERVER_PORT" | grep -q LISTEN; then
+    echo "port :$SERVER_PORT is in use"
+    exit 1
+fi
+
+./etc/run_echo_server.sh "$SERVER_PORT" &
+
+while ! lsof -i ":$SERVER_PORT" | grep -q LISTEN; do
+  sleep 0.1
+done
+
+SERVER_PORT=$SERVER_PORT cargo test --test "*"
+result=$?
+
+kill "$(lsof -t -i:$SERVER_PORT)"
+
+exit $result

tests/echo_test.rs

@@ -0,0 +1,184 @@
+/// Tests unary, server, and bidi streaming with simple echo requests. To run, simply
+/// update the credentials and uri as necessary.
+use anyhow::Result;
+use std::env;
+use std::sync::{Arc, RwLock};
+use viam_rust_utils::gen::proto::rpc::examples::echo::v1::echo_service_client::EchoServiceClient;
+use viam_rust_utils::gen::proto::rpc::examples::echo::v1::{
+    EchoBiDiRequest, EchoMultipleRequest, EchoRequest,
+};
+use viam_rust_utils::rpc::dial;
+
+async fn dial_direct() -> Result<dial::ViamChannel> {
+    let port = env::var("SERVER_PORT").unwrap().to_owned();
+    let uri = ["localhost:".to_string(), port].join("");
+
+    dial::DialOptions::builder()
+        .uri(&uri)
+        .without_credentials()
+        .insecure()
+        .disable_webrtc()
+        .connect()
+        .await
+}
+
+#[tokio::test]
+async fn test_dial_direct_unary() -> Result<()> {
+    let c = dial_direct().await?;
+
+    let mut service = EchoServiceClient::new(c);
+    let echo_request = EchoRequest {
+        message: "hi".to_string(),
+    };
+    let resp = service.echo(echo_request).await?.into_inner();
+    assert_eq!(resp.message, "hi".to_string());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_dial_direct_server_stream() -> Result<()> {
+    let c = dial_direct().await?;
+
+    let mut service = EchoServiceClient::new(c);
+    let multi_echo_request = EchoMultipleRequest {
+        message: "hello?".to_string(),
+    };
+
+    let mut expected = vec!["h", "e", "l", "l", "o", "?"];
+    expected.reverse();
+
+    let mut resp = service
+        .echo_multiple(multi_echo_request)
+        .await?
+        .into_inner();
+    while let Some(resp) = resp.message().await? {
+        assert_eq!(resp.message, expected.pop().unwrap().to_string())
+    }
+    assert!(expected.is_empty());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_dial_direct_bidi() -> Result<()> {
+    let c = dial_direct().await?;
+
+    let received = Arc::new(RwLock::new(0));
+    let received_async = Arc::clone(&received);
+
+    let bidi_stream = async_stream::stream! {
+        for i in 0..3 {
+            loop {
+                // We need to wait a small amount of time between each request/response count
+                // check, otherwise we lock up the main thread.
+                let sleep_time = std::time::Duration::from_millis(10);
+                tokio::time::sleep(sleep_time).await;
+
+                // Wait until we have received one response for each request before sending the
+                // next request. This allows requests/response to be interleaved.
+                let value = received_async.read().unwrap();
+                if *value == i {
+                    break;
+                }
+            }
+
+            let request =
+            EchoBiDiRequest {
+                message: i.to_string()
+            };
+            yield request;
+        }
+    };
+
+    let mut service = EchoServiceClient::new(c);
+    let mut bidi_resp = service.echo_bi_di(bidi_stream).await?.into_inner();
+
+    for i in 0..3 {
+        let resp = bidi_resp.message().await?.unwrap();
+        assert_eq!(resp.message, i.to_string());
+
+        let mut count = received.write().unwrap();
+        *count += 1;
+        drop(count);
+    }
+
+    Ok(())
+}
+
+async fn dial_webrtc() -> Result<dial::ViamChannel> {
+    let port = env::var("SERVER_PORT").unwrap().to_owned();
+    let uri = ["localhost:".to_string(), port].join("");
+
+    dial::DialOptions::builder()
+        .uri(&uri)
+        .without_credentials()
+        .insecure()
+        .connect()
+        .await
+}
+
+#[tokio::test]
+async fn test_dial_webrtc_unary() -> Result<()> {
+    let c = dial_webrtc().await?;
+
+    let mut service = EchoServiceClient::new(c);
+    let echo_request = EchoRequest {
+        message: "hi".to_string(),
+    };
+    let resp = service.echo(echo_request).await?.into_inner();
+    assert_eq!(resp.message, "hi".to_string());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_dial_webrtc_server_stream() -> Result<()> {
+    let c = dial_webrtc().await?;
+
+    let mut service = EchoServiceClient::new(c);
+    let multi_echo_request = EchoMultipleRequest {
+        message: "hello?".to_string(),
+    };
+
+    let mut expected = vec!["h", "e", "l", "l", "o", "?"];
+    expected.reverse();
+
+    let mut resp = service
+        .echo_multiple(multi_echo_request)
+        .await?
+        .into_inner();
+    while let Some(resp) = resp.message().await? {
+        assert_eq!(resp.message, expected.pop().unwrap().to_string())
+    }
+    assert!(expected.is_empty());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_dial_webrtc_bidi() -> Result<()> {
+    let c = dial_webrtc().await?;
+
+    // TODO(RSDK-2414): ideally we should mix the timing of our requests and responses to ensure
+    // that we truly support bi-directionality.
+    let bidi_stream = async_stream::stream! {
+        for i in 0..3 {
+            let request =
+            EchoBiDiRequest {
+                message: i.to_string()
+            };
+            yield request;
+        }
+    };
+
+    let mut service = EchoServiceClient::new(c);
+    let mut bidi_resp = service.echo_bi_di(bidi_stream).await?.into_inner();
+
+    for i in 0..3 {
+        let resp = bidi_resp.message().await?.unwrap();
+        assert_eq!(resp.message, i.to_string());
+    }
+
+    Ok(())
+}