mdebray/tvl-depot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(tvix/castore): simplify test_valid_unix_path_ping_pong

Browse source 
We don't need to spawn two tokio runtimes anymore, and can do the URL
parsing at once, too.

Change-Id: I38ab96978cb7f8c31ded2726262e0b1366655094
Reviewed-on: https://cl.tvl.fyi/c/depot/+/9566
Tested-by: BuildkiteCI
Reviewed-by: Connor Brewster <cbrewster@hey.com>
Autosubmit: flokli <flokli@flokli.de>
This commit is contained in:
Florian Klink 2023-10-08 13:15:30 +02:00 • committed by clbot
parent 269ab866f1
commit b196cbbc67
1 changed files with 44 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

106
tvix/castore/src/blobservice/grpc.rs
View file

@ -281,7 +281,6 @@ impl<W: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for GRPCBlobWriter<
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use std::sync::Arc; use std::sync::Arc;
use std::thread;
use tempfile::TempDir; use tempfile::TempDir;
use tokio::net::UnixListener; use tokio::net::UnixListener;
@ -350,77 +349,60 @@ mod tests {
assert!(GRPCBlobService::from_url(&url).is_err()); assert!(GRPCBlobService::from_url(&url).is_err());
} }
/// This uses the correct scheme for a unix socket, and provides a server on the other side. /// This ensures connecting via gRPC works as expected.
/// This is not a tokio::test, because spawn two separate tokio runtimes and #[tokio::test]
// want to have explicit control. async fn test_valid_unix_path_ping_pong() {
#[test]
fn test_valid_unix_path_ping_pong() {
let tmpdir = TempDir::new().unwrap(); let tmpdir = TempDir::new().unwrap();
let path = tmpdir.path().join("daemon"); let socket_path = tmpdir.path().join("daemon");
let path_clone = path.clone(); let path_clone = socket_path.clone();
// Spin up a server, in a thread far away, which spawns its own tokio runtime, // Spin up a server
// and blocks on the task. tokio::spawn(async {
thread::spawn(move || { let uds = UnixListener::bind(path_clone).unwrap();
// Create the runtime let uds_stream = UnixListenerStream::new(uds);
let rt = tokio::runtime::Runtime::new().unwrap();
let task = rt.spawn(async { // spin up a new server
let uds = UnixListener::bind(path_clone).unwrap(); let mut server = tonic::transport::Server::builder();
let uds_stream = UnixListenerStream::new(uds); let router =
server.add_service(crate::proto::blob_service_server::BlobServiceServer::new(
// spin up a new server GRPCBlobServiceWrapper::from(
let mut server = tonic::transport::Server::builder(); Arc::new(MemoryBlobService::default()) as Arc<dyn BlobService>
let router = ),
server.add_service(crate::proto::blob_service_server::BlobServiceServer::new( ));
GRPCBlobServiceWrapper::from( router.serve_with_incoming(uds_stream).await
Arc::new(MemoryBlobService::default()) as Arc<dyn BlobService>
),
));
router.serve_with_incoming(uds_stream).await
});
rt.block_on(task).unwrap().unwrap();
}); });
// Now create another tokio runtime which we'll use in the main test code. // wait for the socket to be created
let rt = tokio::runtime::Runtime::new().unwrap(); {
let mut socket_created = false;
let task = rt.spawn(async move { // TODO: exponential backoff urgently
// wait for the socket to be created for _try in 1..20 {
{ if socket_path.exists() {
let mut socket_created = false; socket_created = true;
// TODO: exponential backoff urgently break;
for _try in 1..20 {
if path.exists() {
socket_created = true;
break;
}
tokio::time::sleep(time::Duration::from_millis(20)).await;
} }
tokio::time::sleep(time::Duration::from_millis(20)).await;
assert!(
socket_created,
"expected socket path to eventually get created, but never happened"
);
} }
// prepare a client assert!(
let client = { socket_created,
let mut url = "expected socket path to eventually get created, but never happened"
url::Url::parse("grpc+unix:///path/to/somewhere").expect("must parse"); );
url.set_path(path.to_str().unwrap()); }
GRPCBlobService::from_url(&url).expect("must succeed")
};
let has = client // prepare a client
.has(&fixtures::BLOB_A_DIGEST) let grpc_client = {
.await let url = url::Url::parse(&format!("grpc+unix://{}", socket_path.display()))
.expect("must not be err"); .expect("must parse");
GRPCBlobService::from_url(&url).expect("must succeed")
};
assert!(!has); let has = grpc_client
}); .has(&fixtures::BLOB_A_DIGEST)
rt.block_on(task).unwrap() .await
.expect("must not be err");
assert!(!has);
} }
} }