refactor(tvix/castore): move gRPC directorysvc tests off tonic_mock r/6725

The tests were using the gRPC Service/Server interface, not the gRPC
clients.

Use our new gen_directorysvc_grpc_client() function to spin up a server
in a separate thread, and test via that client, rather than *just* the
server implementation.

This is more correct, because we're now really exercising the gRPC
stack, and some of the stream implementations are also only implemented
on the client.

Change-Id: I99434a2399856a44c9e6988d0b720f7a429d0ad1
Reviewed-on: https://cl.tvl.fyi/c/depot/+/9560
Reviewed-by: Connor Brewster <cbrewster@hey.com>
Tested-by: BuildkiteCI
Autosubmit: flokli <flokli@flokli.de>

1 files changed, 90 insertions, 74 deletions

diff --git a/tvix/castore/src/proto/tests/grpc_directoryservice.rs b/tvix/castore/src/proto/tests/grpc_directoryservice.rs
index 6e8cf1e4a7a4..bf60c56aad11 100644
--- a/tvix/castore/src/proto/tests/grpc_directoryservice.rs
+++ b/tvix/castore/src/proto/tests/grpc_directoryservice.rs
@@ -1,26 +1,25 @@
 use crate::fixtures::{DIRECTORY_A, DIRECTORY_B, DIRECTORY_C};
-use crate::proto::directory_service_server::DirectoryService as GRPCDirectoryService;
+use crate::proto::directory_service_client::DirectoryServiceClient;
 use crate::proto::get_directory_request::ByWhat;
+use crate::proto::GetDirectoryRequest;
 use crate::proto::{Directory, DirectoryNode, SymlinkNode};
-use crate::proto::{GRPCDirectoryServiceWrapper, GetDirectoryRequest};
-use crate::utils::gen_directory_service;
+use crate::utils::gen_directorysvc_grpc_client;
+use tempfile::TempDir;
 use tokio_stream::StreamExt;
+use tonic::transport::Channel;
 use tonic::Status;
 
-fn gen_grpc_service() -> GRPCDirectoryServiceWrapper {
-    let directory_service = gen_directory_service();
-    GRPCDirectoryServiceWrapper::from(directory_service)
-}
-
 /// Send the specified GetDirectoryRequest.
 /// Returns an error in the case of an error response, or an error in one of
 // the items in the stream, or a Vec<Directory> in the case of a successful
 /// request.
-async fn get_directories<S: GRPCDirectoryService>(
-    svc: &S,
+async fn get_directories(
+    grpc_client: &mut DirectoryServiceClient<Channel>,
     get_directory_request: GetDirectoryRequest,
 ) -> Result<Vec<Directory>, Status> {
-    let resp = svc.get(tonic::Request::new(get_directory_request)).await;
+    let resp = grpc_client
+        .get(tonic::Request::new(get_directory_request))
+        .await;
 
     // if the response is an error itself, return the error, otherwise unpack
     let stream = match resp {
@@ -38,47 +37,53 @@ async fn get_directories<S: GRPCDirectoryService>(
 /// Trying to get a non-existent Directory should return a not found error.
 #[tokio::test]
 async fn not_found() {
-    let service = gen_grpc_service();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
 
-    let resp = service
+    let resp = grpc_client
         .get(tonic::Request::new(GetDirectoryRequest {
             by_what: Some(ByWhat::Digest(DIRECTORY_A.digest().into())),
             ..Default::default()
         }))
         .await;
 
-    let mut rx = resp.expect("must succeed").into_inner().into_inner();
+    let stream = resp.expect("must succeed").into_inner();
+
+    let items: Vec<_> = stream.collect().await;
 
     // The stream should contain one element, an error with Code::NotFound.
-    let item = rx
-        .recv()
-        .await
-        .expect("must be some")
-        .expect_err("must be err");
-    assert_eq!(item.code(), tonic::Code::NotFound);
-
-    // … and nothing else
-    assert!(rx.recv().await.is_none());
+    assert_eq!(1, items.len());
+    let item = items[0].clone();
+
+    assert!(item.is_err(), "must be err");
+    assert_eq!(
+        tonic::Code::NotFound,
+        item.unwrap_err().code(),
+        "must be err"
+    );
 }
 
 /// Put a Directory into the store, get it back.
 #[tokio::test]
 async fn put_get() {
-    let service = gen_grpc_service();
-
-    let streaming_request = tonic_mock::streaming_request(vec![DIRECTORY_A.clone()]);
-    let put_resp = service
-        .put(streaming_request)
-        .await
-        .expect("must succeed")
-        .into_inner();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
+
+    // send directory A.
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::once(DIRECTORY_A.clone()))
+            .await
+            .expect("must succeed")
+            .into_inner()
+    };
 
     // the sent root_digest should match the calculated digest
     assert_eq!(put_resp.root_digest, DIRECTORY_A.digest().to_vec());
 
     // get it back
     let items = get_directories(
-        &service,
+        &mut grpc_client,
         GetDirectoryRequest {
             by_what: Some(ByWhat::Digest(DIRECTORY_A.digest().into())),
             ..Default::default()
@@ -93,33 +98,36 @@ async fn put_get() {
 /// Put multiple Directories into the store, and get them back
 #[tokio::test]
 async fn put_get_multiple() {
-    let service = gen_grpc_service();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
 
     // sending "b" (which refers to "a") without sending "a" first should fail.
-    let put_resp = service
-        .put(tonic_mock::streaming_request(vec![DIRECTORY_B.clone()]))
-        .await
-        .expect_err("must fail");
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::once(DIRECTORY_B.clone()))
+            .await
+            .expect_err("must fail")
+    };
 
     assert_eq!(tonic::Code::InvalidArgument, put_resp.code());
 
     // sending "a", then "b" should succeed, and the response should contain the digest of b.
-    let put_resp = service
-        .put(tonic_mock::streaming_request(vec![
-            DIRECTORY_A.clone(),
-            DIRECTORY_B.clone(),
-        ]))
-        .await
-        .expect("must succeed");
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::iter(vec![
+                DIRECTORY_A.clone(),
+                DIRECTORY_B.clone(),
+            ]))
+            .await
+            .expect("must succeed")
+            .into_inner()
+    };
 
-    assert_eq!(
-        DIRECTORY_B.digest().to_vec(),
-        put_resp.into_inner().root_digest
-    );
+    assert_eq!(DIRECTORY_B.digest().to_vec(), put_resp.root_digest);
 
     // now, request b, first in non-recursive mode.
     let items = get_directories(
-        &service,
+        &mut grpc_client,
         GetDirectoryRequest {
             recursive: false,
             by_what: Some(ByWhat::Digest(DIRECTORY_B.digest().into())),
@@ -133,7 +141,7 @@ async fn put_get_multiple() {
 
     // now, request b, but in recursive mode.
     let items = get_directories(
-        &service,
+        &mut grpc_client,
         GetDirectoryRequest {
             recursive: true,
             by_what: Some(ByWhat::Digest(DIRECTORY_B.digest().into())),
@@ -149,18 +157,21 @@ async fn put_get_multiple() {
 /// Put multiple Directories into the store, and omit duplicates.
 #[tokio::test]
 async fn put_get_dedup() {
-    let service = gen_grpc_service();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
 
     // Send "A", then "C", which refers to "A" two times
     // Pretend we're a dumb client sending A twice.
-    let put_resp = service
-        .put(tonic_mock::streaming_request(vec![
-            DIRECTORY_A.clone(),
-            DIRECTORY_A.clone(),
-            DIRECTORY_C.clone(),
-        ]))
-        .await
-        .expect("must succeed");
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::iter(vec![
+                DIRECTORY_A.clone(),
+                DIRECTORY_A.clone(),
+                DIRECTORY_C.clone(),
+            ]))
+            .await
+            .expect("must succeed")
+    };
 
     assert_eq!(
         DIRECTORY_C.digest().to_vec(),
@@ -169,7 +180,7 @@ async fn put_get_dedup() {
 
     // Ask for "C" recursively. We expect to only get "A" once, as there's no point sending it twice.
     let items = get_directories(
-        &service,
+        &mut grpc_client,
         GetDirectoryRequest {
             recursive: true,
             by_what: Some(ByWhat::Digest(DIRECTORY_C.digest().into())),
@@ -185,7 +196,8 @@ async fn put_get_dedup() {
 /// Trying to upload a Directory failing validation should fail.
 #[tokio::test]
 async fn put_reject_failed_validation() {
-    let service = gen_grpc_service();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
 
     // construct a broken Directory message that fails validation
     let broken_directory = Directory {
@@ -198,10 +210,12 @@ async fn put_reject_failed_validation() {
     assert!(broken_directory.validate().is_err());
 
     // send it over, it must fail
-    let put_resp = service
-        .put(tonic_mock::streaming_request(vec![broken_directory]))
-        .await
-        .expect_err("must fail");
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::once(broken_directory))
+            .await
+            .expect_err("must fail")
+    };
 
     assert_eq!(put_resp.code(), tonic::Code::InvalidArgument);
 }
@@ -209,7 +223,8 @@ async fn put_reject_failed_validation() {
 /// Trying to upload a Directory with wrong size should fail.
 #[tokio::test]
 async fn put_reject_wrong_size() {
-    let service = gen_grpc_service();
+    let tempdir = TempDir::new().expect("must succeed");
+    let mut grpc_client = gen_directorysvc_grpc_client(tempdir.path()).await;
 
     // Construct a directory referring to DIRECTORY_A, but with wrong size.
     let broken_parent_directory = Directory {
@@ -227,13 +242,14 @@ async fn put_reject_wrong_size() {
     );
 
     // now upload both (first A, then the broken parent). This must fail.
-    let put_resp = service
-        .put(tonic_mock::streaming_request(vec![
-            DIRECTORY_A.clone(),
-            broken_parent_directory,
-        ]))
-        .await
-        .expect_err("must fail");
-
+    let put_resp = {
+        grpc_client
+            .put(tokio_stream::iter(vec![
+                DIRECTORY_A.clone(),
+                broken_parent_directory,
+            ]))
+            .await
+            .expect_err("must fail")
+    };
     assert_eq!(put_resp.code(), tonic::Code::InvalidArgument);
 }