test: add binary data, size overwrite, and CAS conflict tests

- test_binary_values: verify non-UTF8 binary data round-trip
- test_overwrite_different_sizes: small->large->small value overwrites
- test_concurrent_cas_conflict: two concurrent CAS updates to same key,
  verify exactly one succeeds and one gets UnexpectedVersionId
- Total: 45 integration tests + 12 unit tests + 1 doctest = 58 tests

Signed-off-by: mattisonchao <mattisonchao@gmail.com>

Author: mattisonchao

liboxia-native/tests/integration_tests.rs

@@ -1587,3 +1587,125 @@ async fn test_cas_loop() {
         .unwrap();
     client.shutdown().await.unwrap();
 }
+
+// ============================================================
+// Binary (non-UTF8) data
+// ============================================================
+
+#[tokio::test]
+async fn test_binary_values() {
+    let (_container, address) = start_oxia().await;
+    let client = new_client(&address).await;
+
+    // Store binary data that isn't valid UTF-8
+    let binary_data: Vec<u8> = vec![0x00, 0x01, 0xFF, 0xFE, 0x80, 0x90, 0xAB, 0xCD];
+    client
+        .put("bin/data".to_string(), binary_data.clone())
+        .await
+        .unwrap();
+
+    let result = client.get("bin/data".to_string()).await.unwrap();
+    assert_eq!(result.value, Some(binary_data));
+
+    client
+        .delete_range("bin/".to_string(), "bin/~".to_string())
+        .await
+        .unwrap();
+    client.shutdown().await.unwrap();
+}
+
+// ============================================================
+// Overwrite with different value sizes
+// ============================================================
+
+#[tokio::test]
+async fn test_overwrite_different_sizes() {
+    let (_container, address) = start_oxia().await;
+    let client = new_client(&address).await;
+
+    // Start with a small value
+    client
+        .put("sizes/key".to_string(), b"small".to_vec())
+        .await
+        .unwrap();
+
+    // Overwrite with a larger value
+    let large_value = vec![b'x'; 10000];
+    client
+        .put("sizes/key".to_string(), large_value.clone())
+        .await
+        .unwrap();
+
+    let result = client.get("sizes/key".to_string()).await.unwrap();
+    assert_eq!(result.value, Some(large_value));
+
+    // Overwrite back to a small value
+    client
+        .put("sizes/key".to_string(), b"tiny".to_vec())
+        .await
+        .unwrap();
+
+    let result = client.get("sizes/key".to_string()).await.unwrap();
+    assert_eq!(result.value, Some(b"tiny".to_vec()));
+
+    client
+        .delete_range("sizes/".to_string(), "sizes/~".to_string())
+        .await
+        .unwrap();
+    client.shutdown().await.unwrap();
+}
+
+// ============================================================
+// Concurrent CAS conflict detection
+// ============================================================
+
+#[tokio::test]
+async fn test_concurrent_cas_conflict() {
+    let (_container, address) = start_oxia().await;
+    let client = new_client(&address).await;
+
+    // Create a key
+    let initial = client
+        .put("conflict/key".to_string(), b"v0".to_vec())
+        .await
+        .unwrap();
+    let version = initial.version.version_id;
+
+    // Two concurrent CAS updates with the same version - one should fail
+    let c1 = client.clone();
+    let c2 = client.clone();
+
+    let h1 = tokio::spawn(async move {
+        c1.put_with_options(
+            "conflict/key".to_string(),
+            b"v1-from-c1".to_vec(),
+            vec![PutOption::ExpectVersionId(version)],
+        )
+        .await
+    });
+
+    let h2 = tokio::spawn(async move {
+        c2.put_with_options(
+            "conflict/key".to_string(),
+            b"v1-from-c2".to_vec(),
+            vec![PutOption::ExpectVersionId(version)],
+        )
+        .await
+    });
+
+    let r1 = h1.await.unwrap();
+    let r2 = h2.await.unwrap();
+
+    // Exactly one should succeed and one should fail with UnexpectedVersionId
+    let (successes, failures): (Vec<_>, Vec<_>) = vec![r1, r2].into_iter().partition(|r| r.is_ok());
+
+    assert_eq!(successes.len(), 1, "Exactly one CAS should succeed");
+    assert_eq!(failures.len(), 1, "Exactly one CAS should fail");
+    assert!(matches!(failures[0], Err(OxiaError::UnexpectedVersionId())));
+
+    client
+        .delete_range("conflict/".to_string(), "conflict/~".to_string())
+        .await
+        .unwrap();
+    client.shutdown().await.unwrap();
+}