Integration test of concurrent reads and writes

This covers the stream-through-cache case.

Include a note on why we can't exercise "input error from stream" in an
integration test.

Author: cceckman-at-fastly

test-fixtures/Cargo.lock

@@ -19,3 +19,4 @@ http = "1.1.0"
 rustls-pemfile = "1.0.3"
 serde = "1.0.114"
 sha2 = "0.10.8"
+uuid = { version = "1.15.1", features = ["v4"] }

test-fixtures/Cargo.toml

@@ -3,9 +3,19 @@
 use fastly::cache::core::*;
 use std::io::Write;
 use std::time::Duration;
+use uuid::Uuid;
+
+fn main() {
+    test_non_concurrent();
+    test_concurrent();
+    // We don't have a way of testing "incomplete streaming results in an error"
+    // in a single instance. If we fail to close the (write) body handle, the underlying host object
+    // is still hanging around, ready for more writes, until the instance is done.
+    // Oh well -- that's what we have collecting_body::tests::unfinished_stream for.
+}
 
 fn test_non_concurrent() {
-    let key = CacheKey::from_static("hello".as_bytes());
+    let key = new_key();
 
     {
         let fetch = lookup(key.clone())
@@ -14,7 +24,7 @@ fn test_non_concurrent() {
         assert!(fetch.is_none());
     }
 
-    let body = "world".as_bytes();
+    let body = "hello world".as_bytes();
     {
         let mut writer = insert(key.clone(), Duration::from_secs(10))
             .known_length(body.len() as u64)
@@ -34,6 +44,40 @@ fn test_non_concurrent() {
     }
 }
 
-fn main() {
-    test_non_concurrent();
+fn test_concurrent() {
+    let key = new_key();
+
+    {
+        let fetch = lookup(key.clone())
+            .execute()
+            .expect("failed initial lookup");
+        assert!(fetch.is_none());
+    }
+
+    let mut writer = insert(key.clone(), Duration::from_secs(10))
+        .execute()
+        .unwrap();
+
+    let fetch: Found = lookup(key.clone()).execute().unwrap().unwrap();
+    let mut body = fetch.to_stream().unwrap();
+    let mut body = body.read_chunks(6);
+
+    write!(writer, "hello ").unwrap();
+    writer.flush().unwrap();
+
+    // This appears to be the only read mechanism that won't block for more.
+    let hello = body.next().unwrap().unwrap();
+    assert_eq!(hello, b"hello ");
+
+    write!(writer, "world").unwrap();
+    writer.finish().unwrap();
+
+    let cached = body.next().unwrap().unwrap();
+    assert_eq!(cached, b"world");
+
+    assert!(body.next().is_none());
+}
+
+fn new_key() -> CacheKey {
+    Uuid::new_v4().into_bytes().to_vec().into()
 }