E2E, ingest_data_to_player_scores_from_stdio(): add

kamu-data · Oct 8, 2024 · d490c1e · d490c1e
1 parent 1043e13
commit d490c1e
Show file tree

Hide file tree

Showing 3 changed files with 103 additions and 2 deletions.
diff --git a/src/e2e/app/cli/common/src/kamu_api_server_client_ext.rs b/src/e2e/app/cli/common/src/kamu_api_server_client_ext.rs
@@ -15,6 +15,7 @@ use crate::{KamuApiServerClient, RequestBody};
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+/// https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/player-scores.yaml
 pub const DATASET_ROOT_PLAYER_SCORES_SNAPSHOT_STR: &str = indoc::indoc!(
     r#"
     kind: DatasetSnapshot
@@ -42,6 +43,7 @@ pub const DATASET_ROOT_PLAYER_SCORES_SNAPSHOT_STR: &str = indoc::indoc!(
     "#
 );
 
+/// https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/leaderboard.yaml
 pub const DATASET_DERIVATIVE_LEADERBOARD_SNAPSHOT_STR: &str = indoc::indoc!(
     r#"
     kind: DatasetSnapshot
@@ -86,21 +88,45 @@ pub const DATASET_DERIVATIVE_LEADERBOARD_SNAPSHOT_STR: &str = indoc::indoc!(
 );
 
 lazy_static! {
-    // https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/player-scores.yaml
+    /// https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/player-scores.yaml
     pub static ref DATASET_ROOT_PLAYER_SCORES_SNAPSHOT: String = {
         DATASET_ROOT_PLAYER_SCORES_SNAPSHOT_STR
             .escape_default()
             .to_string()
     };
 
-    // https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/leaderboard.yaml
+    /// https://github.com/kamu-data/kamu-cli/blob/master/examples/leaderboard/leaderboard.yaml
     pub static ref DATASET_DERIVATIVE_LEADERBOARD_SNAPSHOT: String = {
         DATASET_DERIVATIVE_LEADERBOARD_SNAPSHOT_STR
             .escape_default()
             .to_string()
     };
 }
 
+/// https://raw.githubusercontent.com/kamu-data/kamu-cli/refs/heads/master/examples/leaderboard/data/1.ndjson
+pub const DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_1: &str = indoc::indoc!(
+    r#"
+    {"match_time": "2000-01-01", "match_id": 1, "player_id": "Alice", "score": 100}
+    {"match_time": "2000-01-01", "match_id": 1, "player_id": "Bob", "score": 80}
+    "#
+);
+
+/// https://raw.githubusercontent.com/kamu-data/kamu-cli/refs/heads/master/examples/leaderboard/data/2.ndjson
+pub const DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_2: &str = indoc::indoc!(
+    r#"
+    {"match_time": "2000-01-02", "match_id": 2, "player_id": "Alice", "score": 70}
+    {"match_time": "2000-01-02", "match_id": 2, "player_id": "Charlie", "score": 90}
+    "#
+);
+
+/// https://raw.githubusercontent.com/kamu-data/kamu-cli/refs/heads/master/examples/leaderboard/data/3.ndjson
+pub const DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_3: &str = indoc::indoc!(
+    r#"
+    {"match_time": "2000-01-03", "match_id": 3, "player_id": "Bob", "score": 60}
+    {"match_time": "2000-01-03", "match_id": 3, "player_id": "Charlie", "score": 110}
+    "#
+);
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 pub type AccessToken = String;

diff --git a/src/e2e/app/cli/inmem/tests/tests/commands/test_ingest_command.rs b/src/e2e/app/cli/inmem/tests/tests/commands/test_ingest_command.rs
@@ -26,3 +26,11 @@ kamu_cli_execute_command_e2e_test!(
 );
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+kamu_cli_execute_command_e2e_test!(
+    storage = inmem,
+    fixture = kamu_cli_e2e_repo_tests::test_ingest_from_stdin,
+    extra_test_groups = "engine, ingest, datafusion"
+);
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/src/e2e/app/cli/repo-tests/src/commands/test_ingest_command.rs b/src/e2e/app/cli/repo-tests/src/commands/test_ingest_command.rs
@@ -11,6 +11,12 @@ use std::path::Path;
 
 use chrono::{TimeZone, Utc};
 use indoc::indoc;
+use kamu_cli_e2e_common::{
+    DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_1,
+    DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_2,
+    DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_3,
+    DATASET_ROOT_PLAYER_SCORES_SNAPSHOT_STR,
+};
 use kamu_cli_puppet::extensions::KamuCliPuppetExt;
 use kamu_cli_puppet::KamuCliPuppet;
 use opendatafabric::*;
@@ -180,8 +186,69 @@ pub async fn test_push_ingest_from_file_snapshot_with_event_time(mut kamu: KamuC
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+pub async fn test_ingest_from_stdin(kamu: KamuCliPuppet) {
+    kamu.execute_with_input(["add", "--stdin"], DATASET_ROOT_PLAYER_SCORES_SNAPSHOT_STR)
+        .await
+        .success();
+
+    ingest_data_to_player_scores_from_stdio(
+        &kamu,
+        DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_1,
+    )
+    .await;
+
+    ingest_data_to_player_scores_from_stdio(
+        &kamu,
+        DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_2,
+    )
+    .await;
+
+    ingest_data_to_player_scores_from_stdio(
+        &kamu,
+        DATASET_ROOT_PLAYER_SCORES_INGEST_DATA_NDJSON_CHUNK_3,
+    )
+    .await;
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+// Helpers
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
 fn path(p: &Path) -> &str {
     p.as_os_str().to_str().unwrap()
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+async fn ingest_data_to_player_scores_from_stdio(kamu: &KamuCliPuppet, input: &str) {
+    // Ingest
+    {
+        let assert = kamu
+            .execute_with_input(["ingest", "player-scores", "--stdin"], input)
+            .await
+            .success();
+
+        let stderr = std::str::from_utf8(&assert.get_output().stderr).unwrap();
+
+        assert!(
+            stderr.contains("Dataset updated"),
+            "Unexpected output:\n{stderr}",
+        );
+    }
+    // Trying to ingest the same data
+    {
+        let assert = kamu
+            .execute_with_input(["ingest", "player-scores", "--stdin"], input)
+            .await
+            .success();
+
+        let stderr = std::str::from_utf8(&assert.get_output().stderr).unwrap();
+
+        assert!(
+            stderr.contains("Dataset up-to-date"),
+            "Unexpected output:\n{stderr}",
+        );
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////