add integration tests

capture-server/tests/events.rs

@@ -13,8 +13,10 @@ async fn it_captures_one_event() -> Result<()> {
     setup_tracing();
     let token = random_string("token", 16);
     let distinct_id = random_string("id", 16);
-    let topic = EphemeralTopic::new().await;
-    let server = ServerHandle::for_topic(&topic).await;
+
+    let main_topic = EphemeralTopic::new().await;
+    let histo_topic = EphemeralTopic::new().await;
+    let server = ServerHandle::for_topics(&main_topic, &histo_topic).await;
 
     let event = json!({
         "token": token,
@@ -24,7 +26,7 @@ async fn it_captures_one_event() -> Result<()> {
     let res = server.capture_events(event.to_string()).await;
     assert_eq!(StatusCode::OK, res.status());
 
-    let event = topic.next_event()?;
+    let event = main_topic.next_event()?;
     assert_json_include!(
         actual: event,
         expected: json!({
@@ -37,14 +39,15 @@ async fn it_captures_one_event() -> Result<()> {
 }
 
 #[tokio::test]
-async fn it_captures_a_batch() -> Result<()> {
+async fn it_captures_a_posthogjs_array() -> Result<()> {
     setup_tracing();
     let token = random_string("token", 16);
     let distinct_id1 = random_string("id", 16);
     let distinct_id2 = random_string("id", 16);
 
-    let topic = EphemeralTopic::new().await;
-    let server = ServerHandle::for_topic(&topic).await;
+    let main_topic = EphemeralTopic::new().await;
+    let histo_topic = EphemeralTopic::new().await;
+    let server = ServerHandle::for_topics(&main_topic, &histo_topic).await;
 
     let event = json!([{
         "token": token,
@@ -59,14 +62,98 @@ async fn it_captures_a_batch() -> Result<()> {
     assert_eq!(StatusCode::OK, res.status());
 
     assert_json_include!(
-        actual: topic.next_event()?,
+        actual: main_topic.next_event()?,
         expected: json!({
             "token": token,
             "distinct_id": distinct_id1
         })
     );
     assert_json_include!(
-        actual: topic.next_event()?,
+        actual: main_topic.next_event()?,
+        expected: json!({
+            "token": token,
+            "distinct_id": distinct_id2
+        })
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn it_captures_a_batch() -> Result<()> {
+    setup_tracing();
+    let token = random_string("token", 16);
+    let distinct_id1 = random_string("id", 16);
+    let distinct_id2 = random_string("id", 16);
+
+    let main_topic = EphemeralTopic::new().await;
+    let histo_topic = EphemeralTopic::new().await;
+    let server = ServerHandle::for_topics(&main_topic, &histo_topic).await;
+
+    let event = json!({
+        "token": token,
+        "batch": [{
+            "event": "event1",
+            "distinct_id": distinct_id1
+        },{
+            "event": "event2",
+            "distinct_id": distinct_id2
+        }]
+    });
+    let res = server.capture_events(event.to_string()).await;
+    assert_eq!(StatusCode::OK, res.status());
+
+    assert_json_include!(
+        actual: main_topic.next_event()?,
+        expected: json!({
+            "token": token,
+            "distinct_id": distinct_id1
+        })
+    );
+    assert_json_include!(
+        actual: main_topic.next_event()?,
+        expected: json!({
+            "token": token,
+            "distinct_id": distinct_id2
+        })
+    );
+
+    Ok(())
+}
+#[tokio::test]
+async fn it_captures_a_historical_batch() -> Result<()> {
+    setup_tracing();
+    let token = random_string("token", 16);
+    let distinct_id1 = random_string("id", 16);
+    let distinct_id2 = random_string("id", 16);
+
+    let main_topic = EphemeralTopic::new().await;
+    let histo_topic = EphemeralTopic::new().await;
+    let server = ServerHandle::for_topics(&main_topic, &histo_topic).await;
+
+    let event = json!({
+        "token": token,
+        "historical_migration": true,
+        "batch": [{
+            "event": "event1",
+            "distinct_id": distinct_id1
+        },{
+            "event": "event2",
+            "distinct_id": distinct_id2
+        }]
+    });
+    let res = server.capture_events(event.to_string()).await;
+    assert_eq!(StatusCode::OK, res.status());
+
+    assert_json_include!(
+        actual: histo_topic.next_event()?,
+        expected: json!({
+            "token": token,
+            "distinct_id": distinct_id1
+        })
+    );
+    assert_json_include!(
+        actual: histo_topic.next_event()?,
         expected: json!({
             "token": token,
             "distinct_id": distinct_id2

capture/tests/django_compat.rs

@@ -4,7 +4,7 @@ use axum::http::StatusCode;
 use axum_test_helper::TestClient;
 use base64::engine::general_purpose;
 use base64::Engine;
-use capture::api::{CaptureError, CaptureResponse, CaptureResponseCode, ProcessedEvent};
+use capture::api::{CaptureError, CaptureResponse, CaptureResponseCode, DataType, ProcessedEvent};
 use capture::limiters::billing::BillingLimiter;
 use capture::redis::MockRedisClient;
 use capture::router::router;
@@ -146,6 +146,9 @@ async fn it_matches_django_capture_behaviour() -> anyhow::Result<()> {
         for (event_number, (message, expected)) in
             sink.events().iter().zip(case.output.iter()).enumerate()
         {
+            // Ensure the data type matches
+            assert_eq!(DataType::AnalyticsMain, message.data_type);
+
             // Normalizing the expected event to align with known django->rust inconsistencies
             let mut expected = expected.clone();
             if let Some(value) = expected.get_mut("sent_at") {