Merge pull request bruin-data#366 from bruin-data/make-tests-independent

 seperate pipelines for integration tests

Author: terzioglub

integration-tests/.gitignore

@@ -1,2 +1,3 @@
 logs/*.log
-.bruin.yml
+.bruin.yml
+logs/runs

integration-tests/integration-test.go

@@ -238,24 +238,24 @@ func getWorkflow(binary string, currentFolder string, tempfile string) []e2e.Wor
 func getTasks(binary string, currentFolder string) []e2e.Task {
 	return []e2e.Task{
 		{
-			Name:          "happy-path",
+			Name:          "parse-whole-pipeline",
 			Command:       binary,
-			Args:          []string{"internal", "parse-pipeline", filepath.Join(currentFolder, "happy-path")},
+			Args:          []string{"internal", "parse-pipeline", filepath.Join(currentFolder, "test-pipelines/parse-whole-pipeline")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "happy-path/expectations/pipeline.yml.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/parse-whole-pipeline/expectations/pipeline.yml.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
 				e2e.AssertByOutputJSON,
 			},
 		},
 		{
-			Name:    "chess-extended",
+			Name:    "run-with-tags",
 			Command: binary,
-			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", filepath.Join(currentFolder, "chess-extended")},
+			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", filepath.Join(currentFolder, "test-pipelines/run-with-tags-pipeline")},
 			Env:     []string{},
 
 			Expected: e2e.Output{
@@ -266,9 +266,9 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:    "chess-extended-only-checks",
+			Name:    "run-with-filters",
 			Command: binary,
-			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", "--only", "checks", filepath.Join(currentFolder, "chess-extended")},
+			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", "--only", "checks", filepath.Join(currentFolder, "test-pipelines/run-with-filters-pipeline")},
 			Env:     []string{},
 
 			Expected: e2e.Output{
@@ -283,7 +283,7 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:    "format-if-fail",
 			Command: binary,
-			Args:    []string{"format", "--fail-if-changed", filepath.Join(currentFolder, "chess-extended/assets/chess_games.asset.yml")},
+			Args:    []string{"format", "--fail-if-changed", filepath.Join(currentFolder, "test-pipelines/format-if-changed-pipeline/assets/correctly-formatted.sql")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 0,
@@ -293,9 +293,9 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:    "chess-extended-only-main",
+			Name:    "run-main-with-filters",
 			Command: binary,
-			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", "--only", "main", filepath.Join(currentFolder, "chess-extended")},
+			Args:    []string{"run", "--tag", "include", "--exclude-tag", "exclude", "--only", "main", filepath.Join(currentFolder, "test-pipelines/run-main-with-filters-pipeline")},
 			Env:     []string{},
 
 			Expected: e2e.Output{
@@ -308,9 +308,9 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:    "downstream-chess-extended",
+			Name:    "run-with-downstream",
 			Command: binary,
-			Args:    []string{"run", "--downstream", filepath.Join(currentFolder, "chess-extended/assets/game_outcome_summary.sql")},
+			Args:    []string{"run", "--downstream", filepath.Join(currentFolder, "test-pipelines/run-with-downstream-pipeline/assets/game_outcome_summary.sql")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 0,
@@ -322,9 +322,9 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:    "downstream-only-main-chess-extended",
+			Name:    "run-main-with-downstream",
 			Command: binary,
-			Args:    []string{"run", "--downstream", "--only", "main", filepath.Join(currentFolder, "chess-extended/assets/game_outcome_summary.sql")},
+			Args:    []string{"run", "--downstream", "--only", "main", filepath.Join(currentFolder, "test-pipelines/run-main-with-downstream-pipeline/assets/game_outcome_summary.sql")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 0,
@@ -338,7 +338,7 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:    "push-metadata",
 			Command: binary,
-			Args:    []string{"run", "--push-metadata", "--only", "push-metadata", filepath.Join(currentFolder, "bigquery-metadata")},
+			Args:    []string{"run", "--push-metadata", "--only", "push-metadata", filepath.Join(currentFolder, "test-pipelines/push-metadata-pipeline")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 1,
@@ -352,7 +352,7 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:    "validate-happy-path",
 			Command: binary,
-			Args:    []string{"validate", filepath.Join(currentFolder, "happy-path")},
+			Args:    []string{"validate", filepath.Join(currentFolder, "test-pipelines/happy-path")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 0,
@@ -362,9 +362,9 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:    "run-use-uv-happy-path",
+			Name:    "run-use-uv",
 			Command: binary,
-			Args:    []string{"run", "--use-uv", filepath.Join(currentFolder, "happy-path")},
+			Args:    []string{"run", "--use-uv", filepath.Join(currentFolder, "test-pipelines/run-use-uv-pipeline")},
 			Env:     []string{},
 			Expected: e2e.Output{
 				ExitCode: 0,
@@ -376,12 +376,12 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:          "parse-asset-happy-path-asset-py",
 			Command:       binary,
-			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "happy-path/assets/asset.py")},
+			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "test-pipelines/happy-path/assets/asset.py")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "happy-path/expectations/asset.py.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/happy-path/expectations/asset.py.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
@@ -391,12 +391,12 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:          "parse-asset-happy-path-chess-games",
 			Command:       binary,
-			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "happy-path/assets/chess_games.asset.yml")},
+			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "test-pipelines/happy-path/assets/chess_games.asset.yml")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "happy-path/expectations/chess_games.asset.yml.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/happy-path/expectations/chess_games.asset.yml.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
@@ -406,12 +406,12 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:          "parse-asset-happy-path-chess-profiles",
 			Command:       binary,
-			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "happy-path/assets/chess_profiles.asset.yml")},
+			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "test-pipelines/happy-path/assets/chess_profiles.asset.yml")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "happy-path/expectations/chess_profiles.asset.yml.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/happy-path/expectations/chess_profiles.asset.yml.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
@@ -421,12 +421,12 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:          "parse-asset-happy-path-player-summary",
 			Command:       binary,
-			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "happy-path/assets/player_summary.sql")},
+			Args:          []string{"internal", "parse-asset", filepath.Join(currentFolder, "test-pipelines/happy-path/assets/player_summary.sql")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "happy-path/expectations/player_summary.sql.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/happy-path/expectations/player_summary.sql.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
@@ -436,7 +436,7 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:    "parse-asset-faulty-pipeline-error-sql",
 			Command: binary,
-			Args:    []string{"internal", "parse-asset", filepath.Join(currentFolder, "faulty-pipeline/assets/error.sql")},
+			Args:    []string{"internal", "parse-asset", filepath.Join(currentFolder, "test-pipelines/faulty-pipeline/assets/error.sql")},
 			Env:     []string{},
 
 			Expected: e2e.Output{
@@ -451,12 +451,12 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:          "validate-missing-upstream",
 			Command:       binary,
-			Args:          []string{"validate", "-o", "json", filepath.Join(currentFolder, "missing-upstream/assets/nonexistent.sql")},
+			Args:          []string{"validate", "-o", "json", filepath.Join(currentFolder, "test-pipelines/missing-upstream-pipeline/assets/nonexistent.sql")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "missing-upstream/expectations/missing_upstream.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/missing-upstream-pipeline/expectations/missing_upstream.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
@@ -466,7 +466,7 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 		{
 			Name:    "run-malformed-sql",
 			Command: binary,
-			Args:    []string{"run", filepath.Join(currentFolder, "malformed/assets/malformed.sql")},
+			Args:    []string{"run", filepath.Join(currentFolder, "test-pipelines/run-malformed-pipeline/assets/malformed.sql")},
 			Env:     []string{},
 
 			Expected: e2e.Output{
@@ -509,29 +509,29 @@ func getTasks(binary string, currentFolder string) []e2e.Task {
 			},
 		},
 		{
-			Name:          "parse-pipeline-lineage",
+			Name:          "parse-lineage",
 			Command:       binary,
-			Args:          []string{"internal", "parse-pipeline", "-c", filepath.Join(currentFolder, "lineage")},
+			Args:          []string{"internal", "parse-pipeline", "-c", filepath.Join(currentFolder, "test-pipelines/parse-lineage-pipeline")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "lineage/expectations/lineage.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/parse-lineage-pipeline/expectations/lineage.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,
 				e2e.AssertByOutputJSON,
 			},
 		},
 		{
-			Name:          "parse-asset-lineage-example",
+			Name:          "parse-asset-lineage",
 			Command:       binary,
-			Args:          []string{"internal", "parse-asset", "-c", filepath.Join(currentFolder, "lineage/assets/example.sql")},
+			Args:          []string{"internal", "parse-asset", "-c", filepath.Join(currentFolder, "test-pipelines/parse-asset-lineage-pipeline/assets/example.sql")},
 			Env:           []string{},
 			SkipJSONNodes: []string{"\"path\""},
 			Expected: e2e.Output{
 				ExitCode: 0,
-				Output:   helpers.ReadFile(filepath.Join(currentFolder, "lineage/expectations/lineage-asset.json")),
+				Output:   helpers.ReadFile(filepath.Join(currentFolder, "test-pipelines/parse-asset-lineage-pipeline/expectations/lineage-asset.json")),
 			},
 			Asserts: []func(*e2e.Task) error{
 				e2e.AssertByExitCode,

integration-tests/test-pipelines/bruin-pipeline/.gitignore

@@ -0,0 +1 @@
+.bruin.yml

integration-tests/test-pipelines/bruin-pipeline/README.md

@@ -0,0 +1,98 @@
+# Bruin - Sample Pipeline
+
+This pipeline is a simple example of a Bruin pipeline. It demonstrates how to use the `bruin` CLI to build and run a pipeline.
+
+The pipeline includes two sample assets already:
+- `myschema.example`: A simple SQL asset that creates a table in BigQuery.
+  - Feel free to change the type from `bq.sql` to anything.
+- `myschema.country_list`: A simple Python asset that installs a dependency and runs the logic.
+
+## Setup
+The pipeline already includes an empty `.bruin.yml` file, fill it with your connections and environments. You can read more about connections [here](https://bruin-data.github.io/bruin/connections/overview.html).
+
+Here's a sample `.bruin.yml` file:
+
+```yaml
+environments:
+  default:
+    connections:
+      google_cloud_platform:
+        - name: "gcp"
+          service_account_file: "/path/to/my/key.json"
+          project_id: "my-project-dev"
+      snowflake:
+        - name: "snowflake"
+          username: "my-user"
+          password: "my-password"
+          account: "my-account"
+          database: "my-database"
+          warehouse: "my-warehouse"
+          schema: "my-dev-schema"
+      generic:
+        - name: KEY1
+          value: value1
+  production:
+    connections:
+      google_cloud_platform:
+        - name: "gcp"
+          service_account_file: "/path/to/my/prod-key.json"
+          project_id: "my-project-prod"
+      snowflake:
+        - name: "snowflake"
+          username: "my-user"
+          password: "my-password"
+          account: "my-account"
+          database: "my-database"
+          warehouse: "my-warehouse"
+          schema: "my-prod-schema" 
+      generic:
+        - name: KEY1
+          value: value1
+```
+
+You can simply switch the environment using the `--environment` flag, e.g.:
+
+```shell
+bruin validate --environment production . 
+```
+
+## Running the pipeline
+
+bruin CLI can run the whole pipeline or any task with the downstreams:
+
+```shell
+bruin run .
+```
+
+```shell
+Starting the pipeline execution...
+
+[2023-03-16T18:25:14Z] [worker-0] Running: dashboard.bruin-test
+[2023-03-16T18:25:16Z] [worker-0] Completed: dashboard.bruin-test (1.681s)
+[2023-03-16T18:25:16Z] [worker-4] Running: hello
+[2023-03-16T18:25:16Z] [worker-4] [hello] >> Hello, world!
+[2023-03-16T18:25:16Z] [worker-4] Completed: hello (116ms)
+
+Executed 2 tasks in 1.798s
+```
+
+You can also run a single task:
+
+```shell
+bruin run assets/hello.py                            
+```
+
+```shell
+Starting the pipeline execution...
+
+[2023-03-16T18:25:59Z] [worker-0] Running: hello
+[2023-03-16T18:26:00Z] [worker-0] [hello] >> Hello, world!
+[2023-03-16T18:26:00Z] [worker-0] Completed: hello (103ms)
+
+
+Executed 1 tasks in 103ms
+```
+
+You can optionally pass a `--downstream` flag to run the task with all of its downstreams.
+
+That's it, good luck!

integration-tests/test-pipelines/bruin-pipeline/assets/product_categories.sql

@@ -0,0 +1,35 @@
+/* @bruin
+name: product_categories
+type: duckdb.sql
+materialization:
+  type: table
+
+columns:
+  - name: category_id
+    type: INTEGER
+    description: "Unique identifier for the product category"
+    primary_key: true
+    checks:
+      - name: not_null
+      - name: positive
+  - name: category_name
+    type: VARCHAR
+    description: "Name of the product category"
+    checks:
+      - name: not_null
+  - name: description
+    type: VARCHAR
+    description: "Description of the product category"
+@bruin */
+
+SELECT
+    1 AS category_id, 'Electronics' AS category_name, 'Devices like phones, laptops, and monitors' AS description
+UNION ALL
+SELECT
+    2 AS category_id, 'Accessories' AS category_name, 'Complementary items like headphones and chargers' AS description
+UNION ALL
+SELECT
+    3 AS category_id, 'Appliances' AS category_name, 'Household devices like refrigerators and microwaves' AS description
+UNION ALL
+SELECT
+    4 AS category_id, 'Furniture' AS category_name, 'Home and office furniture like desks and chairs' AS description;