Integration test for focused runs

integration-tests/chess-extended/assets/chess_games.asset.yml

@@ -0,0 +1,8 @@
+name: chess_playground.games
+type: ingestr
+parameters:
+  source_connection: chess-default
+  source_table: games
+  destination: duckdb
+tags:
+    - include

integration-tests/chess-extended/assets/chess_profiles.asset.yml

@@ -0,0 +1,8 @@
+name: chess_playground.profiles
+type: ingestr
+parameters:
+  source_connection: chess-default
+  source_table: profiles
+  destination: duckdb
+tags:
+    - include

integration-tests/chess-extended/assets/game_outcome_summary.sql

@@ -0,0 +1,31 @@
+/* @bruin
+
+name: chess_playground.game_outcome_summary
+type: duckdb.sql
+materialization:
+   type: table
+
+depends:
+   - chess_playground.games
+
+columns:
+  - name: result_type
+    type: string
+    description: "Type of game result (win, draw, etc.)"
+  - name: total_games
+    type: integer
+    description: "Total number of games with this result"
+    checks:
+      - name: positive
+tags:
+   - include
+
+@bruin */
+
+SELECT
+    g.white->>'result' AS result_type,
+    COUNT(*) AS total_games
+FROM chess_playground.games g
+WHERE g.white->>'result' IS NOT NULL
+GROUP BY g.white->>'result'
+ORDER BY total_games DESC;

integration-tests/chess-extended/assets/player_profile_summary.sql

@@ -0,0 +1,30 @@
+/* @bruin
+
+name: chess_playground.player_profile_summary
+type: duckdb.sql
+materialization:
+   type: table
+
+depends:
+   - chess_playground.profiles
+columns:
+  - name: total_players
+    type: integer
+    description: "Total number of players in the profiles table"
+    checks:
+      - name: positive
+  - name: active_players
+    type: integer
+    description: "Number of players marked as active"
+  - name: inactive_players
+    type: integer
+    description: "Number of players marked as inactive"
+tags:
+    - exclude
+@bruin */
+
+SELECT
+    COUNT(*) AS total_players,
+    COUNT(CASE WHEN p.status = 'active' THEN 1 END) AS active_players,
+    COUNT(CASE WHEN p.status = 'inactive' THEN 1 END) AS inactive_players
+FROM chess_playground.profiles p;

integration-tests/chess-extended/assets/player_summary.sql

@@ -0,0 +1,62 @@
+/* @bruin
+
+name: chess_playground.player_summary
+type: duckdb.sql
+materialization:
+   type: table
+
+depends:
+   - chess_playground.game_outcome_summary
+   - chess_playground.player_profile_summary
+   - chess_playground.games
+   - chess_playground.profiles
+
+columns:
+  - name: username
+    type: string
+    description: "Username of the player"
+  - name: total_games
+    type: integer
+    description: "Total games played by the player"
+    checks:
+      - name: non_negative
+  - name: total_wins
+    type: integer
+    description: "Total games won by the player"
+  - name: win_rate
+    type: float
+    description: "Win rate of the player"
+tags:
+    - include
+    - exclude
+@bruin */
+
+WITH game_results AS (
+    SELECT
+        CASE
+            WHEN g.white->>'result' = 'win' THEN g.white->>'@id'
+            WHEN g.black->>'result' = 'win' THEN g.black->>'@id'
+            ELSE NULL
+            END AS winner_aid,
+        g.white->>'@id' AS white_aid,
+    g.black->>'@id' AS black_aid
+FROM chess_playground.games g
+    )
+
+SELECT
+    p.username,
+    p.aid,
+    COUNT(*) AS total_games,
+    COUNT(CASE WHEN g.white_aid = p.aid AND g.winner_aid = p.aid THEN 1 END) +
+    COUNT(CASE WHEN g.black_aid = p.aid AND g.winner_aid = p.aid THEN 1 END) AS total_wins,
+    ROUND(
+            (COUNT(CASE WHEN g.white_aid = p.aid AND g.winner_aid = p.aid THEN 1 END) +
+             COUNT(CASE WHEN g.black_aid = p.aid AND g.winner_aid = p.aid THEN 1 END)) * 100.0 /
+            NULLIF(COUNT(*), 0),
+            2
+    ) AS win_rate
+FROM chess_playground.profiles p
+         LEFT JOIN game_results g
+                   ON p.aid IN (g.white_aid, g.black_aid)
+GROUP BY p.username, p.aid
+ORDER BY total_games DESC;

integration-tests/chess-extended/expectations/tables_summary.json

@@ -0,0 +1 @@
+{"columns":[{"name":"table_name"}],"rows":[["games"],["game_outcome_summary"],["profiles"],["_dlt_loads"],["_dlt_pipeline_state"],["_dlt_version"]]}

integration-tests/chess-extended/pipeline.yml

@@ -0,0 +1 @@
+name: chess_extended_duckdb

integration-tests/integration-test.go

@@ -31,11 +31,23 @@ func main() {
 			panic(err)
 		}
 	}
-
+	expectJSONOutput("internal parse-pipeline happy-path", "happy-path/expectations/pipeline.yml.json")
+	expectExitCode("run --tag include --exclude-tag exclude chess-extended", 0)
+	expectExitCode("run --tag include --exclude-tag exclude chess-extended/expectations/chess_games.asset.yml", 1)
+	expectOutputIncludes(
+		"run --tag include --exclude-tag exclude --only checks chess-extended",
+		0,
+		[]string{"Executed 1 tasks", "total_games:positive"},
+	)
+	expectQueryResult(
+		"duckdb-default",
+		"SELECT table_name FROM information_schema.tables WHERE table_schema = 'chess_playground';",
+		"json",
+		"chess-extended/expectations/tables_summary.json",
+	)
 	expectExitCode("validate happy-path", 0)
 	expectExitCode("run --use-uv happy-path", 0)
 	// expectExitCode("run happy-path", 0)
-	expectJSONOutput("internal parse-pipeline happy-path", "happy-path/expectations/pipeline.yml.json")
 	expectJSONOutput(
 		"internal parse-asset happy-path/assets/asset.py",
 		"happy-path/expectations/asset.py.json",
@@ -200,3 +212,47 @@ func runCommandWithExitCode(command string) (string, int, error) {
 	}
 	return output, 0, nil
 }
+func expectQueryResult(connection, query, outputFile, expectationPath string) {
+	args := []string{"run", "../main.go", "query",
+		"--connection", connection,
+		"--query", query,
+		"--output", outputFile,
+	}
+
+	cmd := exec.Command("go", args...)
+	cmd.Dir = currentFolder
+	outputBytes, err := cmd.CombinedOutput()
+	output := string(outputBytes)
+	if err != nil {
+		fmt.Println("Command failed with error:", err)
+		os.Exit(1)
+	}
+	if _, err := os.Stat(outputFile); os.IsNotExist(err) {
+		err = os.WriteFile(outputFile, []byte(output), 0600)
+		if err != nil {
+			os.Exit(1)
+		}
+	}
+	expectationPathFull := filepath.Join(currentFolder, expectationPath)
+	expectation, err := jd.ReadJsonFile(expectationPathFull)
+	if err != nil {
+		fmt.Println("Failed to read expectation file:", err)
+		os.Exit(1)
+	}
+
+	parsedOutput, err := jd.ReadJsonString(strings.ReplaceAll(output, "\\r\\n", "\\n"))
+	if err != nil {
+		fmt.Println("Failed to parse JSON output:", err)
+		os.Exit(1)
+	}
+
+	diff := expectation.Diff(parsedOutput)
+	if len(diff) != 0 {
+		fmt.Println("Mismatch detected:")
+		for _, d := range diff {
+			fmt.Printf("Path: %v, Expected: %v, Actual: %v\n", d.Path, d.NewValues, d.OldValues)
+		}
+		os.Exit(1)
+	}
+	fmt.Println("Passed")
+}