# Ploosh — Full Content Bundle

> Ploosh is an open source YAML-based framework used to automate the testing process in data projects.
> This document concatenates the canonical content of https://ploosh.io for ingestion by Large Language Models.
> See also: https://ploosh.io/llms.txt

---

## Get Started

Source: https://ploosh.io/

### Steps

1. Install the ploosh package
2. Setup a connection file
3. Setup the test cases
4. Run tests
5. Get results

### Installation

Install the ploosh package from [PyPi](https://pypi.org/project/ploosh/) package manager with the following command

```bash
pip install ploosh
```

### Setup a connection file

Add a yaml file with name "connections.yml" and following content:

``` yaml
mssql_getstarted:
  type: mysql
  hostname: my_server_name.database.windows.net
  database: my_database_name
  username: my_user_name
  # using a parameter is highly recommended
  password: $var.my_sql_server_password
```

### Setup the test cases

Add a folder "test_cases" with a yaml file with any name. In this example "example.yaml". Add the following content:

``` yaml
Test aggregated data:
  options:
    sort:
      - gender
      - domain
  source:
    connection: mysql_demo
    type: mysql
    query: |
      select gender, right(email, length(email) - position("@" in email)) as domain, count(*) as count
        from users
        group by gender, domain
  expected:
    type: csv
    path: ./data/test_target_agg.csv
```

### Run tests

``` bash
ploosh --connections "connections.yml" --cases "test_cases" --export "JSON" --p_my_sql_server_password "mypassword"
```
![](/assets/img/ploosh-run.png)


### Test results
``` json
[
  {
    "name": "Test aggregated data",
    "state": "passed",
    "source": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.0032982
    },
    "expected": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.0012451
    }
  }
]
```

### Run with spark

It's possible to run the tests with spark. To do that, you need to install the spark package or use a platform that already has it installed like Databricks or Microsoft Fabric.

See the [Spark connector](/docs/configuration/spark) for more information.

---

## Documentation

### Api

#### Execute Cases

Source: https://ploosh.io/docs/api/execute-cases

#### Python API — execute_cases()

The `execute_cases()` function is the main entry point for running Ploosh programmatically from Python, typically in notebooks (Microsoft Fabric, Databricks) or custom scripts.

##### Import

``` python
from ploosh import execute_cases
```

##### Signature

``` python
execute_cases(
    cases=None,
    connections=None,
    spark=None,
    spark_session=None,
    filter=None,
    path_output=None,
    variables=None,
    workers=1,
)
```

##### Parameters

| Parameter | Type | Default | Description |
|-----------|------|:-------:|-------------|
| `cases` | string | | Path to the folder containing test case YAML files |
| `connections` | string | | Path to the connections YAML file |
| `spark` | string | | Set to `"true"` to enable Spark mode |
| `spark_session` | SparkSession | | Existing PySpark SparkSession to use |
| `filter` | string | | Glob pattern to filter test case files (e.g. `"*.yaml"`) |
| `path_output` | string | | Path to the output folder for results |
| `variables` | dict | | Dictionary of custom parameter values |
| `workers` | int | `1` | Number of parallel workers used to process test cases (`1` = sequential) |

##### Examples

###### Basic usage

``` python
from ploosh import execute_cases

execute_cases(
    cases="test_cases",
    connections="connections.yml"
)
```

###### Microsoft Fabric

``` python
from ploosh import execute_cases

execute_cases(
    cases="/lakehouse/default/Files/ploosh_cases",
    connections="/lakehouse/default/Files/ploosh_connections.yaml",
    spark_session=spark,
    path_output="/lakehouse/default/Files/ploosh_outputs"
)
```

###### Databricks

``` python
from ploosh import execute_cases

execute_cases(
    cases="/Workspace/Shared/cases",
    connections="/Workspace/Shared/connections.yaml",
    spark_session=spark,
    path_output="/Workspace/Shared/output"
)
```

###### With variables

``` python
from ploosh import execute_cases

execute_cases(
    cases="test_cases",
    connections="connections.yml",
    variables={
        "db_password": "my_secret",
        "schema": "production"
    }
)
```

###### With filter

``` python
from ploosh import execute_cases

execute_cases(
    cases="test_cases",
    connections="connections.yml",
    filter="quality_*.yaml"
)
```

###### Parallel execution

``` python
from ploosh import execute_cases

execute_cases(
    cases="test_cases",
    connections="connections.yml",
    workers=4
)
```

##### Behavior

- If `spark_session` is provided, Ploosh uses it for Spark connectors
- If `spark="true"` and no `spark_session` is given, Ploosh creates a local SparkSession
- With `workers > 1`, test cases are processed concurrently in a thread pool; `workers=1` (default) runs them sequentially
- Results are exported to `{path_output}/{format}/` (e.g. `output/json/test_results.json`)
- The function prints status to stdout in real-time
- By default, raises `SystemExit(1)` if any test fails. Use `--failure false` from CLI or handle the exit in your code

### Configuration

#### Command Line

Source: https://ploosh.io/docs/configuration/command-line

#### Command line

Ploosh can be executed from the command line using the `ploosh` command.

##### Usage

``` shell
ploosh --connections <path> --cases <path> [options]
```

##### Arguments

| Argument | Mandatory | Default | Description |
|----------|:---------:|:-------:|-------------|
| `--connections` | no | | Path to the connections YAML file |
| `--cases` | no | `./cases` | Path to the folder containing test case YAML files |
| `--filter` | no | `*.yml` | Glob pattern to filter test case files |
| `--output` | no | `./output` | Path to the output folder for results |
| `--export` | no | `JSON` | Export format: `JSON`, `CSV`, or `TRX` |
| `--spark` | no | `false` | Enable Spark mode (creates a local SparkSession) |
| `--failure` | no | `true` | Exit with code 1 if any test fails or errors |
| `--workers` | no | `1` | Number of parallel workers used to process test cases (`1` = sequential) |
| `--p_<name>` | no | | Custom parameter value (see [Custom parameters](/docs/configuration/custom-parameters)) |

##### Examples

###### Basic execution

``` shell
ploosh --connections "connections.yml" --cases "test_cases"
```

###### With export format and parameters

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --export "TRX" --p_db_password "my_password"
```

###### With filter and custom output

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --filter "*.yaml" --output "./results"
```

###### Disable failure exit code

Useful in CI/CD pipelines where you want to publish results even when tests fail:

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --failure false
```

###### Spark mode

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --spark true
```

> When `--spark true` is set and no spark session is provided programmatically, Ploosh creates a local SparkSession. For Fabric or Databricks, use the Python API instead. See [Spark mode overview](/docs/spark/overview).

###### Parallel execution

Process test cases concurrently using multiple workers to speed up large test suites:

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --workers 4
```

> `--workers` controls how many test cases are processed at the same time. The default value `1` runs test cases sequentially. Increasing the number of workers can significantly reduce total execution time when test cases are I/O-bound (waiting on databases, files, or remote services). Choose a value based on your available resources and the limits of the systems you query.

##### Python API

When running inside a notebook (Fabric, Databricks), use the `execute_cases()` function instead:

``` python
from ploosh import execute_cases

execute_cases(
    cases="/path/to/cases",
    connections="/path/to/connections.yaml",
    spark_session=spark
)
```

See [Python API reference](/docs/api/execute-cases) for all parameters.

#### Custom Parameters

Source: https://ploosh.io/docs/configuration/custom-parameters

#### Custom parameters

Custom parameters allow you to avoid hardcoding sensitive information like passwords or environment-specific values in your YAML configuration files.

##### Syntax

Use the `$var.<parameter_name>` syntax in your YAML files to reference a custom parameter.

##### Passing values

###### Command line

Pass the parameter value using `--p_<parameter_name>`:

``` shell
ploosh --connections connections.yml --cases test_cases --p_db_password "my_secret" --p_environment "production"
```

###### Python API

Pass variables as a dictionary:

``` python
from ploosh import execute_cases

execute_cases(
    cases="test_cases",
    connections="connections.yml",
    variables={
        "db_password": "my_secret",
        "environment": "production"
    }
)
```

##### Usage in connections

``` yaml
connections:
  my_database:
    type: mysql
    hostname: my-server.database.windows.net
    database: my_database
    username: my_user
    password: $var.db_password
```

##### Usage in test cases

``` yaml
Test with parameter:
  source:
    type: mysql
    connection: my_database
    query: |
      SELECT *
      FROM $var.environment.employees
  expected:
    type: empty
```

##### CI/CD usage

In CI/CD pipelines, pass parameters from secure variable groups or secrets:

###### Azure DevOps

``` yaml
- task: CmdLine@2
  inputs:
    script: ploosh --connections connections.yml --cases test_cases --p_db_password "$(db_password)"
```

###### GitHub Actions

``` yaml
- run: ploosh --connections connections.yml --cases test_cases --p_db_password "${{ secrets.DB_PASSWORD }}"
```

##### Security

- Never commit sensitive values (passwords, tokens) in YAML files
- Always use `$var` references and pass values at runtime
- In CI/CD, store secrets in variable groups or secret stores

#### Options

Source: https://ploosh.io/docs/configuration/options

#### Test case options

Test cases allow defining options to control the comparison behavior. Options are set in the `options` section of the test case YAML.

##### All options

| Option | Type | Default | Available in | Description |
|--------|------|:-------:|:------------:|-------------|
| `compare_mode` | string | `order` | Native, Spark | Comparison mode: `order` or `join` |
| `join_keys` | list | `[]` | Spark only | Columns to join on (for `join` mode) |
| `sort` | list | | Native, Spark | Columns to sort before comparison |
| `ignore` | list | | Native, Spark | Columns to exclude from comparison |
| `cast` | list | `[]` | Native, Spark | Columns to cast to a specific type |
| `pass_rate` | decimal | `1` | Native, Spark | Minimum success rate (0.0 to 1.0) |
| `tolerance` | decimal | `0` | Native, Spark | Numeric tolerance for float comparisons |
| `trim` | boolean | `false` | Native, Spark | Trim whitespace from string columns |
| `case_insensitive` | boolean | `false` | Native, Spark | Convert strings to lowercase before comparing |
| `allow_no_rows` | boolean | `true` | Native, Spark | Allow both datasets to be empty without error |
| `disabled` | boolean | `false` | Native, Spark | Skip this test case entirely |

---

##### compare_mode

Defines how rows are matched between source and expected datasets.

| Mode | Description | Available in |
|------|-------------|:------------:|
| `order` | Rows are matched by position (row 1 vs row 1, row 2 vs row 2, etc.) | Native, Spark |
| `join` | Rows are matched by joining on `join_keys` columns | Spark only |

###### Example (order mode — default)

``` yaml
Example:
  options:
    compare_mode: order
    sort:
      - employee_id
  source:
    type: sql_spark
    query: SELECT * FROM employees ORDER BY employee_id
  expected:
    type: csv_spark
    path: ./expected.csv
```

###### Example (join mode — Spark only)

``` yaml
Example:
  options:
    compare_mode: join
    join_keys:
      - employee_id
  source:
    type: sql_spark
    query: SELECT * FROM employees
  expected:
    type: csv_spark
    path: ./expected.csv
```

> The `join` mode is useful when row ordering is non-deterministic or when matching by business keys is more appropriate.

---

##### sort

Sort both datasets before comparison. Specify a list of column names to sort by, or `["*"]` to sort by all columns.

``` yaml
Example:
  options:
    sort:
      - department
      - name
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

> **Best practice**: Sort in your SQL queries (`ORDER BY`) for better performance and deterministic results.

---

##### ignore

Exclude specific columns from the comparison. Useful for audit columns, timestamps, or auto-generated fields.

``` yaml
Example:
  options:
    ignore:
      - created_at
      - updated_at
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

---

##### cast

Cast column types before comparison. Useful when source and expected have different types for the same data.

Allowed types: `int`, `float`, `string`, `datetime`

``` yaml
Example:
  options:
    cast:
      - name: salary
        type: float
      - name: hire_date
        type: datetime
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

---

##### pass_rate

Define the minimum percentage of matching rows required for the test to pass. Value between `0.0` and `1.0` (default: `1.0` = all rows must match).

``` yaml
Example:
  options:
    pass_rate: 0.95
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

> A `pass_rate` of `0.95` means the test passes if at least 95% of rows match.

---

##### tolerance

Allow small numeric differences when comparing float/decimal columns. Set to `0` (default) for exact matching.

``` yaml
Example:
  options:
    tolerance: 0.01
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM financial_data
  expected:
    type: csv
    path: ./expected.csv
```

> Useful for floating-point precision differences between systems (e.g. migration from one database to another).

---

##### trim

Trim leading and trailing whitespace from all string columns before comparison.

``` yaml
Example:
  options:
    trim: true
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

---

##### case_insensitive

Convert all string values to lowercase before comparison.

``` yaml
Example:
  options:
    case_insensitive: true
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees
  expected:
    type: csv
    path: ./expected.csv
```

---

##### allow_no_rows

When set to `true` (default), the test passes if both source and expected datasets are empty. When `false`, empty datasets cause the test to fail.

``` yaml
Example:
  options:
    allow_no_rows: false
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM employees WHERE department = 'NonExistent'
  expected:
    type: empty
```

---

##### disabled

Skip a test case entirely. The test will appear as `notExecuted` in the results.

``` yaml
Temporarily disabled test:
  disabled: true
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM problematic_table
  expected:
    type: empty
```

### Connectors

#### Native - Analysis Services

Source: https://ploosh.io/docs/connectors/native/analysis-services

#### Analysis Services

This connector is used to query Analysis Services models using DAX queries via the ADOMD.NET library.

> This connector requires Windows and the ADOMD.NET library (`pyadomd` + `pythonnet`).

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `oauth` | Authentication mode: `oauth`, `pbix` |
| server | yes | | Analysis Services server address |
| dataset_id | yes | | Dataset ID (in DAX Studio: right-click model name → "Copy Database ID") |
| token | no | | Access token (for `token` mode) |
| username | no | | Username (for `credentials` mode) |
| password | no | | Password (for `credentials` mode) |
| tenant_id | no | | Azure AD tenant ID (for `spn` mode) |
| client_id | no | | Azure AD application client ID (for `spn` mode) |
| client_secret | no | | Azure AD application client secret (for `spn` mode) |
| scope | no | `https://analysis.windows.net/powerbi/api/.default` | OAuth scope |

###### Authentication modes

| Mode | Description |
|------|-------------|
| `oauth` | Opens a browser login page. Auto-connects for local AS instances |
| `pbix` | Same as `oauth`, used for local Power BI Desktop models |

##### Connection example

``` yaml
connections:
  my_analysis_services:
    type: analysis_services
    mode: oauth
    server: powerbi://api.powerbi.com/v1.0/myorg/MyWorkspace
    dataset_id: my-dataset-id
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | DAX query to execute |

##### Test case example

``` yaml
Test DAX measure:
  source:
    type: analysis_services
    connection: my_analysis_services
    query: |
      EVALUATE
      SUMMARIZECOLUMNS(
        DimProduct[Category],
        "Total Sales", [Total Sales]
      )
  expected:
    type: csv
    path: ./expected/sales_by_category.csv
```

##### Requirements

- Windows OS
- ADOMD.NET library installed (typically at `C:\Program Files\Microsoft.NET\ADOMD.NET\`)
- `pip install pyadomd pythonnet`
- For `spn` mode: `pip install azure-identity`

#### Native - Big Query

Source: https://ploosh.io/docs/connectors/native/big-query

#### BigQuery

This connector is used to query Google BigQuery using SQL.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| credentials_type | no | `service_account` | Authentication type: `service_account` or `current_user` |
| credentials | no | | Google [keyfile](https://googleapis.dev/python/google-api-core/latest/auth.html) encoded in base64 (for `service_account` mode) |
| project_id | no | | Google Cloud project ID (for `current_user` mode) |

When `credentials_type` is `service_account`, the `credentials` parameter must contain a base64-encoded Google service account keyfile.

When `credentials_type` is `current_user`, the connector uses `pandas_gbq` with the default credentials from the environment (e.g. gcloud CLI or the `GOOGLE_APPLICATION_CREDENTIALS` environment variable).

> ⚠️ It is highly recommended to use a [custom parameter](/docs/configuration/custom-parameters) to pass the credentials value.

###### Example

``` yaml
connections:
  bigquery_example:
    type: bigquery
    credentials_type: service_account
    credentials: $var.gbq_credentials
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example BigQuery:
  source:
    type: bigquery
    connection: bigquery_example
    query: |
      SELECT *
      FROM `rh.employees`
      WHERE hire_date < "2000-01-01"
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- `pip install pandas-gbq` (included in `ploosh` full installation)
- For `service_account` mode: a valid Google Cloud service account keyfile
```

#### Native - CSV

Source: https://ploosh.io/docs/connectors/native/csv

#### CSV

This connector is used to read local CSV files.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the CSV file |
| delimiter | no | `,` | Delimiter used in the CSV file |
| infer | no | `true` | Infer column names from the first row |
| names | no | | List of column labels to apply |
| schema | no | | Dictionary of column names and types (`int`, `float`, `string`, `bool`, `datetime`) |
| usecols | no | | List of columns to select |
| skiprows | no | | Number of rows to skip at the start, or list of row indices to skip (0-indexed) |
| skipfooter | no | `0` | Number of rows to skip at the end of the file |
| nrows | no | | Number of rows to read |
| lineterminator | no | | Character used to denote a line break |
| quotechar | no | `"` | Character used to denote the start and end of a quoted item |
| encoding | no | `utf-8` | Encoding to use when reading the file |
| engine | no | | Parser engine to use (`c` or `python`) |

###### Example

``` yaml
Example CSV:
  source:
    type: csv
    path: ./data/employees.csv
    delimiter: ";"
    encoding: utf-8
  expected:
    type: csv
    path: ./data/expected_employees.csv
```

###### Example with schema

``` yaml
Example CSV with schema:
  source:
    type: csv
    path: ./data/employees.csv
    schema:
      id: int
      name: string
      salary: float
      hire_date: datetime
  expected:
    type: empty
```

###### Example without header

``` yaml
Example CSV without header:
  source:
    type: csv
    path: ./data/no_header.csv
    infer: false
    names:
      - id
      - name
      - department
  expected:
    type: empty
```

#### Native - Databricks

Source: https://ploosh.io/docs/connectors/native/databricks

#### Databricks

This connector is used to query a Databricks SQL warehouse or cluster using SQL.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| token | yes | | Personal Access Token (PAT) for authentication |
| hostname | yes | | Databricks instance hostname (e.g. `adb-1234567890.1.azuredatabricks.net`) |
| database | yes | | Database name |
| http_path | yes | | HTTP path for the SQL warehouse or cluster (found in JDBC/ODBC settings) |
| port | no | `443` | Port number |

###### Example

``` yaml
connections:
  databricks_connection:
    type: databricks
    token: $var.databricks_token
    hostname: adb-1234567890.1.azuredatabricks.net
    database: my_database
    http_path: /sql/1.0/warehouses/abc123
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example Databricks:
  source:
    type: databricks
    connection: databricks_connection
    query: |
      SELECT *
      FROM employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- `pip install databricks-sqlalchemy` (included in `ploosh` full installation)

#### Native - Delta

Source: https://ploosh.io/docs/connectors/native/delta

#### Delta

This connector is used to read local Delta tables using the `deltalake` library.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the Delta table directory |

###### Example

``` yaml
Example Delta:
  source:
    type: delta
    path: ./data/employees_delta
  expected:
    type: csv
    path: ./data/expected_employees.csv
```

#### Native - Empty

Source: https://ploosh.io/docs/connectors/native/empty

#### Empty

This connector returns an empty DataFrame (0 rows, 0 columns). It is used as the expected side of a test case when the source query should return no data.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

No configuration is required by this connector.

###### Example

``` yaml
Test no invalid records:
  source:
    connection: my_database
    type: mssql
    query: |
      SELECT *
      FROM fact_orders
      WHERE amount < 0
  expected:
    type: empty
```

##### Use cases

- Verify absence of anomalies or invalid data
- Check for duplicates
- Validate referential integrity
- Ensure no NULL values in mandatory columns

See [Testing approaches](/docs/use-cases/testing-approaches) for more examples.

#### Native - Excel

Source: https://ploosh.io/docs/connectors/native/excel

#### Excel

This connector is used to read local Excel files (.xlsx, .xls).

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the Excel file |
| sheet_name | yes | | Sheet name or sheet index (0-based) |
| skiprows | no | `0` | Number of rows to skip at the start of the sheet |

###### Example

``` yaml
Example Excel:
  source:
    type: excel
    path: ./data/employees.xlsx
    sheet_name: Sheet1
  expected:
    type: csv
    path: ./data/expected_employees.csv
```

###### Example with sheet index

``` yaml
Example Excel by index:
  source:
    type: excel
    path: ./data/report.xlsx
    sheet_name: 0
    skiprows: 2
  expected:
    type: empty
```

##### Requirements

- `openpyxl` (included in `ploosh` installation)

#### Native - JSON

Source: https://ploosh.io/docs/connectors/native/json

#### JSON

This connector is used to read local JSON files.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the JSON file |
| lines | no | `false` | Whether to treat the file as line-delimited JSON (one JSON object per line) |
| nrows | no | | Number of lines to read (line-delimited mode only) |
| encoding | no | `utf-8` | Encoding to use when reading the file |

###### Example

``` yaml
Example JSON:
  source:
    type: json
    path: ./data/employees.json
  expected:
    type: csv
    path: ./data/expected_employees.csv
```

###### Example with line-delimited JSON

``` yaml
Example JSONL:
  source:
    type: json
    path: ./data/events.jsonl
    lines: true
    nrows: 1000
  expected:
    type: empty
```

#### Native - MySQL

Source: https://ploosh.io/docs/connectors/native/mysql

#### MySQL

This connector is used to query a MySQL database using SQL.

##### Connection configuration

Two connection modes are available: `password` and `connection_string`.

###### Password mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `password` | Connection mode: `password` or `connection_string` |
| hostname | yes | | Server hostname or IP address |
| database | yes | | Database name |
| username | yes | | Username |
| password | yes | | Password |
| port | no | `3306` | Port number |
| require_secure_transport | no | `false` | Enable SSL/TLS transport |

###### Connection string mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | yes | | Must be `connection_string` |
| connection_string | yes | | SQLAlchemy connection string |

###### Example (password mode)

``` yaml
connections:
  mysql_connection:
    type: mysql
    hostname: my-server.database.windows.net
    database: my_database
    username: my_user
    password: $var.mysql_password
    port: 3306
```

###### Example (connection string mode)

``` yaml
connections:
  mysql_connection:
    type: mysql
    mode: connection_string
    connection_string: mysql+pymysql://user:password@host:3306/database
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example MySQL:
  source:
    type: mysql
    connection: mysql_connection
    query: |
      SELECT *
      FROM employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- `pip install pymysql` (included in `ploosh` full installation)

#### Native - ODBC

Source: https://ploosh.io/docs/connectors/native/odbc

#### ODBC

This connector is used to query any database accessible via an ODBC driver.

##### Connection configuration

Two connection modes are available: `DSN` and `connection_string`.

###### DSN mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `DSN` | Connection mode: `DSN` or `connection_string` |
| DSN | yes | | Data Source Name configured on the system |
| auto_commit | no | `true` | Enable auto-commit |
| use_credentials | no | `false` | Whether to pass username/password |
| user | no | | Username (when `use_credentials` is `true`) |
| password | no | | Password (when `use_credentials` is `true`) |
| encoding | no | `UTF-8` | Encoding for the connection |

###### Connection string mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | yes | | Must be `connection_string` |
| connection_string | yes | | ODBC connection string |
| auto_commit | no | `true` | Enable auto-commit |
| encoding | no | `UTF-8` | Encoding for the connection |

###### Example (DSN mode)

``` yaml
connections:
  odbc_connection:
    type: odbc
    DSN: my_data_source
    use_credentials: true
    user: my_user
    password: $var.odbc_password
```

###### Example (connection string mode)

``` yaml
connections:
  odbc_connection:
    type: odbc
    mode: connection_string
    connection_string: "Driver={ODBC Driver 18 for SQL Server};Server=myserver;Database=mydb;Uid=myuser;Pwd=mypassword;"
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example ODBC:
  source:
    type: odbc
    connection: odbc_connection
    query: |
      SELECT *
      FROM employees
      WHERE department = 'Engineering'
  expected:
    type: csv
    path: data/expected_engineers.csv
```

##### Requirements

- `pip install pyodbc` (included in `ploosh` full installation)
- An ODBC driver installed and a DSN configured on the system

#### Native - Parquet

Source: https://ploosh.io/docs/connectors/native/parquet

#### Parquet

This connector is used to read local Parquet files.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the Parquet file |
| columns | no | | List of columns to load (subset) |
| engine | no | `auto` | Parquet engine: `auto`, `pyarrow`, or `fastparquet` |
| filters | no | | List of row group filters to apply |
| filters.column | yes | | Column name to filter on |
| filters.operator | yes | | Comparison operator: `==`, `=`, `>`, `>=`, `<`, `<=`, `!=` |
| filters.value | yes | | Value to filter by |

###### Example

``` yaml
Example Parquet:
  source:
    type: parquet
    path: ./data/employees.parquet
  expected:
    type: csv
    path: ./data/expected_employees.csv
```

###### Example with column selection

``` yaml
Example Parquet with columns:
  source:
    type: parquet
    path: ./data/employees.parquet
    columns:
      - id
      - name
      - department
  expected:
    type: empty
```

###### Example with filters

``` yaml
Example Parquet with filters:
  source:
    type: parquet
    path: ./data/employees.parquet
    filters:
      - column: department_id
        operator: "=="
        value: 5
  expected:
    type: csv
    path: ./data/expected_dept5.csv
```

#### Native - PostgreSQL

Source: https://ploosh.io/docs/connectors/native/postgresql

#### PostgreSQL

This connector is used to query a PostgreSQL database using SQL.

##### Connection configuration

Two connection modes are available: `password` and `connection_string`.

###### Password mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `password` | Connection mode: `password` or `connection_string` |
| hostname | yes | | Server hostname or IP address |
| database | yes | | Database name |
| username | yes | | Username |
| password | yes | | Password |
| port | no | `5432` | Port number |
| ssl_context | no | `false` | Enable SSL connection |

###### Connection string mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | yes | | Must be `connection_string` |
| connection_string | yes | | SQLAlchemy connection string |

###### Example (password mode)

``` yaml
connections:
  postgresql_connection:
    type: postgresql
    hostname: my-server.postgres.database.azure.com
    database: my_database
    username: my_user
    password: $var.pg_password
    port: 5432
    ssl_context: true
```

###### Example (connection string mode)

``` yaml
connections:
  postgresql_connection:
    type: postgresql
    mode: connection_string
    connection_string: postgresql+pg8000://user:password@host:5432/database
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example PostgreSQL:
  source:
    type: postgresql
    connection: postgresql_connection
    query: |
      SELECT *
      FROM employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- `pip install pg8000` (included in `ploosh` full installation)

#### Native - Semantic Model Xmla

Source: https://ploosh.io/docs/connectors/native/semantic-model-xmla

#### Semantic Model (XMLA)

This connector is used to query Microsoft Fabric Semantic Models (Power BI datasets) using the XMLA endpoint REST API with DAX queries.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `oauth` | Authentication mode: `oauth` |
| dataset_id | yes | | Power BI dataset ID |
| token | no | | Access token (for `token` mode) |
| tenant_id | no | | Azure AD tenant ID (for `spn` mode) |
| client_id | no | | Azure AD application client ID (for `spn` mode) |
| client_secret | no | | Azure AD application client secret (for `spn` mode) |

###### Authentication modes

| Mode | Description |
|------|-------------|
| `oauth` | Opens a browser login page for interactive authentication |

##### Connection example

``` yaml
connections:
  my_semantic_model:
    type: semantic_model
    mode: oauth
    dataset_id: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | DAX query to execute |
| body | no | | Custom request body (overrides default) |

##### Test case example

``` yaml
Test semantic model measure:
  source:
    type: semantic_model
    connection: my_semantic_model
    query: |
      EVALUATE
      SUMMARIZECOLUMNS(
        DimDate[Year],
        "Revenue", [Total Revenue]
      )
  expected:
    type: csv
    path: ./expected/revenue_by_year.csv
```

##### How it works

The connector sends a POST request to the Power BI REST API:

```
POST https://api.powerbi.com/v1.0/myorg/datasets/{dataset_id}/executeQueries
```

The DAX query is wrapped in a JSON body with `includeNulls: true` and the results are parsed into a DataFrame.

##### Requirements

- `pip install azure-identity requests`
- Power BI Premium or Fabric capacity (XMLA endpoint must be enabled)
- Appropriate permissions on the dataset

#### Native - Snowflake

Source: https://ploosh.io/docs/connectors/native/snowflake

#### Snowflake

This connector is used to query a Snowflake data warehouse using SQL.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| account_identifier | yes | | Snowflake account identifier (e.g. `xy12345.us-east-1`) |
| username | yes | | Username |
| password | yes | | Password |
| database | no | | Database name |
| schema | no | | Schema name |
| warehouse | no | | Warehouse name |
| role | no | | Role name |

###### Example

``` yaml
connections:
  snowflake_connection:
    type: snowflake
    account_identifier: xy12345.us-east-1
    username: my_user
    password: $var.snowflake_password
    database: my_database
    schema: public
    warehouse: my_warehouse
    role: my_role
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example Snowflake:
  source:
    type: snowflake
    connection: snowflake_connection
    query: |
      SELECT *
      FROM employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- `pip install snowflake-sqlalchemy` (included in `ploosh` full installation)

#### Native - SQL Server

Source: https://ploosh.io/docs/connectors/native/sqlserver

#### SQL Server

This connector is used to query a Microsoft SQL Server database using SQL.

> **Requirement**: ODBC Driver 18 for SQL Server must be installed on the system.
>
> Installation guides: [Windows](https://learn.microsoft.com/en-us/sql/connect/odbc/download-odbc-driver-for-sql-server) | [Linux](https://learn.microsoft.com/en-us/sql/connect/odbc/linux-mac/installing-the-microsoft-odbc-driver-for-sql-server) | [macOS](https://learn.microsoft.com/en-us/sql/connect/odbc/linux-mac/install-microsoft-odbc-driver-sql-server-macos)

##### Connection configuration

Two connection modes are available: `password` and `connection_string`.

###### Password mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | no | `password` | Connection mode: `password` or `connection_string` |
| hostname | yes | | Server hostname or IP address |
| database | yes | | Database name |
| username | yes | | Username |
| password | yes | | Password |
| port | no | `1433` | Port number |
| encrypt | no | `true` | Enable connection encryption |
| trust_server_certificate | no | `false` | Trust the server certificate without validation |
| driver | no | `ODBC Driver 18 for SQL Server` | ODBC driver name |

###### Connection string mode

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| mode | yes | | Must be `connection_string` |
| connection_string | yes | | SQLAlchemy connection string |

###### Example (password mode)

``` yaml
connections:
  mssql_connection:
    type: mssql
    hostname: my-server.database.windows.net
    database: my_database
    username: my_user
    password: $var.mssql_password
    port: 1433
    encrypt: true
    trust_server_certificate: false
```

###### Example (connection string mode)

``` yaml
connections:
  mssql_connection:
    type: mssql
    mode: connection_string
    connection_string: mssql+pyodbc:///?odbc_connect=Driver={ODBC Driver 18 for SQL Server};Server=myserver;Database=mydb;Uid=myuser;Pwd=mypassword;Encrypt=yes;
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute |

###### Example

``` yaml
Example SQL Server:
  source:
    type: mssql
    connection: mssql_connection
    query: |
      SELECT *
      FROM employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: csv
    path: data/employees_before_2000.csv
```

##### Requirements

- [ODBC Driver 18 for SQL Server](https://learn.microsoft.com/en-us/sql/connect/odbc/download-odbc-driver-for-sql-server)
- `pip install pyodbc` (included in `ploosh` full installation)

#### Spark - CSV

Source: https://ploosh.io/docs/connectors/spark/csv

#### CSV (Spark)

This connector is used to read CSV files using Spark.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the CSV file (supports wildcards, e.g. `*.csv`) |
| delimiter | no | `,` | Delimiter used in the CSV file |
| header | no | `true` | Whether the CSV file has a header row |
| inferSchema | no | `false` | Automatically infer column types from data |
| multiline | no | `false` | Enable parsing of records spanning multiple lines |
| quote | no | `"` | Character used to denote the start and end of a quoted item |
| encoding | no | `UTF-8` | Encoding to use when reading the file |
| lineSep | no | `\n` | Character used to denote a line break |

###### Example

``` yaml
Example CSV Spark:
  source:
    type: csv_spark
    path: /lakehouse/default/Files/data/employees/*.csv
    header: true
    inferSchema: true
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
```

#### Spark - Delta

Source: https://ploosh.io/docs/connectors/spark/delta

#### Delta (Spark)

This connector is used to read Delta tables using Spark.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the Delta table directory |

###### Example

``` yaml
Example Delta Spark:
  source:
    type: delta_spark
    path: /lakehouse/default/Tables/employees
  expected:
    type: csv_spark
    path: /lakehouse/default/Files/expected/employees.csv
    header: true
    inferSchema: true
```

#### Spark - Dremio

Source: https://ploosh.io/docs/connectors/spark/dremio

#### Dremio (Spark)

This connector is used to query Dremio using Spark via the Arrow Flight SQL JDBC driver.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| host | yes | | Dremio server hostname or IP address |
| port | no | `32010` | Arrow Flight SQL port |
| use_encryption | no | `true` | Enable TLS encryption for the connection |
| disable_certificate_verification | no | `false` | Disable TLS certificate verification (not recommended in production) |
| username | yes | | Dremio username |
| password | yes | | Dremio password or PAT |

###### Example

``` yaml
connections:
  dremio_connection:
    type: dremio_spark
    host: my-dremio-server.example.com
    port: 32010
    use_encryption: true
    disable_certificate_verification: false
    username: my_user
    password: $var.dremio_password
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute against Dremio |

###### Example

``` yaml
Example Dremio Spark:
  source:
    type: dremio_spark
    connection: dremio_connection
    query: |
      SELECT *
      FROM my_schema.employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
      WHERE hire_date < '2000-01-01'
```

##### Requirements

- Apache Arrow Flight SQL JDBC driver (`org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver`)
- Dremio instance accessible from the Spark cluster

#### Spark - Empty

Source: https://ploosh.io/docs/connectors/spark/empty

#### Empty (Spark)

This connector returns an empty Spark DataFrame (0 rows, 0 columns). It is used as the expected side of a test case when the source query should return no data.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

No configuration is required by this connector.

###### Example

``` yaml
Test no anomalies:
  source:
    type: sql_spark
    query: |
      SELECT *
      FROM lakehouse.fact_orders
      WHERE amount < 0
  expected:
    type: empty_spark
```

##### Use cases

- Verify absence of anomalies or invalid data in Lakehouse tables
- Check for duplicates in distributed datasets
- Validate KQL query returns no critical events

See [Testing approaches](/docs/use-cases/testing-approaches) for more examples.

#### Spark - Fabric Kql

Source: https://ploosh.io/docs/connectors/spark/fabric-kql

#### Fabric KQL (Spark)

This connector is used to query Microsoft Fabric KQL databases using Spark and the Kusto Spark connector.

It supports two execution modes:
- `native` (Spark Kusto connector)
- `api` (Fabric KQL REST API)

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| connection_mode | no | api | Execution mode: `native` or `api` |
| kusto_uri | yes | | Kusto cluster URI (e.g. `https://mycluster.kusto.windows.net`) |
| database_id | yes (native mode) | | KQL database ID used by Spark native connector |
| database_name | yes (api mode) | | KQL database name used by REST API mode |

###### Example (native mode)

``` yaml
connections:
  kql_connection:
    type: fabric_kql_spark
    connection_mode: native
    kusto_uri: https://mycluster.kusto.windows.net
    database_id: my_kql_database
```

###### Example (api mode)

``` yaml
connections:
  kql_connection:
    type: fabric_kql_spark
    connection_mode: api
    kusto_uri: https://mycluster.kusto.windows.net
    database_name: my_kql_database
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | KQL query to execute |

###### Example

``` yaml
Example Fabric KQL:
  source:
    type: fabric_kql_spark
    connection: kql_connection
    query: |
      Employees
      | where HireDate < datetime(2000-01-01)
      | project EmployeeId, Name, Department, HireDate
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
      WHERE hire_date < '2000-01-01'
```

##### Authentication

This connector uses the Microsoft Fabric authentication token obtained via `mssparkutils.credentials.getToken("kusto")`. The Spark environment must have access to Microsoft Fabric resources.

##### Runtime behavior

- In `api` mode, the connector expects a `PrimaryResult` table in the API response.
- If the API call fails or no `PrimaryResult` table is returned, the connector raises a clear error.
- If performance issues are observed with `native` mode, prefer `api` mode.

##### Requirements

- Microsoft Kusto Spark connector (`com.microsoft.kusto.spark.datasource`)
- Microsoft Fabric workspace with appropriate permissions
- Spark environment with `mssparkutils` available (typically in Fabric notebooks)

#### Spark - Fabric Semantic Model

Source: https://ploosh.io/docs/connectors/spark/fabric-semantic-model

#### Fabric Semantic Model Spark connector

This connector is used to query a Microsoft Fabric Semantic Model and return a Spark DataFrame.

Warning: a Spark connector can be used only with another Spark connector. It is not possible to use a Spark connector with a non-Spark connector.

See [Spark documentation](Spark) for more information.

#### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| semantic_model_name | yes |  | Semantic model (dataset) name in Fabric |
| workspace_name | no | `null` | Fabric workspace name. If omitted, default workspace context is used |

#### Configuration

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| method | yes |  | Query method. Allowed values: `DAX Query`, `Table`, `Measure` |
| dax_query | no | `null` | DAX query text. Used when `method: DAX Query` |
| table_to_query | no | `null` | Table name to read. Used when `method: Table` |
| measure_to_query | no | `null` | Measure name to evaluate. Used when `method: Measure` |
| group_by | no | `null` | List of grouping columns for measure evaluation. Used when `method: Measure` |
| filters_to_apply | no | `null` | Dictionary of filters for measure evaluation. Used when `method: Measure` |
| column_names | yes |  | List of output column names. List size must match returned column count |


##### Examples

```yaml
Example Fabric Semantic Model DAX:
  source:
    type: fabric_semantic_model_spark
    connection: test_fabric_semantic_model_spark
    method: DAX Query
    dax_query: |
      EVALUATE SUMMARIZECOLUMNS(
        'CountrySales'[Country],
        'CountrySales'[Sales]
      )
    column_names:
      - Country
      - Sales
  expected:
    type: empty_spark
```

```yaml
Example Fabric Semantic Model Measure:
  source:
    type: fabric_semantic_model_spark
    connection: test_fabric_semantic_model_spark
    method: Measure
    measure_to_query: Sales with tax
    group_by:
      - "'CountrySales'[Country]"
    filters_to_apply:
      {"'CountrySales'[Country]" : ["France", "Germany"]}
  expected:
    type: empty_spark
```

```yaml
Example Fabric Semantic Model Table:
  source:
    type: fabric_semantic_model_spark
    connection: test_fabric_semantic_model_spark
    method: Table
    table_to_query: CountrySales
    column_names:
      - Country
      - Sales
  expected:
    type: empty_spark
```

#### Spark - Fabric Warehouse

Source: https://ploosh.io/docs/connectors/spark/fabric-warehouse

#### Fabric Warehouse (Spark)

This connector is used to query a Microsoft Fabric Warehouse using Spark via the `sempy_labs` library.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| warehouse_name | yes | | Name (or ID) of the Fabric Warehouse to connect to |
| workspace_name | no | `None` | Name (or ID) of the Fabric workspace containing the warehouse (if not specified, uses current workspace) |

###### Example

``` yaml
connections:
  fabric_warehouse_connection:
    type: fabric_warehouse_spark
    warehouse_name: my_warehouse
    workspace_name: my_workspace
```

###### Example (current workspace)

When `workspace_name` is omitted, the connector uses the current workspace:

``` yaml
connections:
  fabric_warehouse_connection:
    type: fabric_warehouse_spark
    warehouse_name: my_warehouse
```

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | SQL query to execute against the Fabric Warehouse |

###### Example

``` yaml
Example Fabric Warehouse Spark:
  source:
    type: fabric_warehouse_spark
    connection: fabric_warehouse_connection
    query: |
      SELECT *
      FROM dbo.employees
      WHERE hire_date < '2000-01-01'
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
      WHERE hire_date < '2000-01-01'
```

###### Example with cross-workspace warehouse

Query a warehouse in a different workspace:

``` yaml
Cross-workspace Warehouse Test:
  source:
    type: fabric_warehouse_spark
    connection: fabric_warehouse_connection
    query: |
      SELECT 
        department,
        COUNT(*) AS employee_count,
        AVG(salary) AS avg_salary
      FROM dbo.employees
      GROUP BY department
  expected:
    type: csv_spark
    path: /lakehouse/default/Files/expected/department_summary.csv
    header: true
    inferSchema: true
```

##### Requirements

- Microsoft Fabric environment with active Spark session
- `sempy-labs` package (automatically available in Fabric notebooks)
- Read permissions on the target Fabric Warehouse
- Query permissions on the warehouse tables

##### Notes

- The connector operates within an active Spark session in a Fabric environment
- Authentication is automatic when running in a Fabric notebook
- The query executes against the Fabric Warehouse and returns a Spark DataFrame

#### Spark - JSON

Source: https://ploosh.io/docs/connectors/spark/json

#### JSON (Spark)

This connector is used to read JSON files using Spark.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the JSON file |
| multiline | no | `true` | Enable reading of multi-line JSON files |
| encoding | no | `UTF-8` | Encoding to use when reading the file |
| lineSep | no | `\n` | Character used to denote a line break |

###### Example

``` yaml
Example JSON Spark:
  source:
    type: json_spark
    path: /lakehouse/default/Files/data/employees.json
    multiline: true
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
```

#### Spark - Parquet

Source: https://ploosh.io/docs/connectors/spark/parquet

#### Parquet (Spark)

This connector is used to read Parquet files using Spark.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| path | yes | | Path to the Parquet file |

###### Example

``` yaml
Example Parquet Spark:
  source:
    type: parquet_spark
    path: /lakehouse/default/Files/data/employees.parquet
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM expected_employees
```

#### Spark - SQL

Source: https://ploosh.io/docs/connectors/spark/sql

#### SQL (Spark)

This connector is used to execute Spark SQL queries. It is the primary connector for querying Lakehouse tables in Microsoft Fabric or registered tables in Databricks.

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

See [Spark mode overview](/docs/spark/overview) for more information.

##### Connection configuration

No connection is required by this connector.

##### Test case configuration

| Name | Mandatory | Default | Description |
|------|:---------:|:-------:|-------------|
| query | yes | | Spark SQL query to execute |

###### Example

``` yaml
Example SQL Spark:
  source:
    type: sql_spark
    query: |
      SELECT department, COUNT(*) AS count
      FROM lakehouse.employees
      GROUP BY department
  expected:
    type: csv_spark
    path: /lakehouse/default/Files/expected/department_counts.csv
    header: true
    inferSchema: true
```

###### Example with Fabric shortcuts

When using shortcuts in Microsoft Fabric, remote Lakehouse tables are queryable as local tables:

``` yaml
Test cross-workspace data:
  source:
    type: sql_spark
    query: |
      SELECT *
      FROM dw_lakehouse.fact_sales
      WHERE sale_date >= '2024-01-01'
  expected:
    type: sql_spark
    query: |
      SELECT *
      FROM reporting_lakehouse.fact_sales_report
      WHERE sale_date >= '2024-01-01'
```

See [Fabric shortcuts strategy](/docs/spark/fabric-shortcuts) for more details.

### Developers

#### Add Connector

Source: https://ploosh.io/docs/developers/add-connector

#### Add a new connector

This guide explains how to create a new connector for Ploosh. A connector is a Python class that fetches data from a source and returns it as a DataFrame.

##### Architecture overview

Ploosh discovers connectors **automatically** at startup. The `__init__.py` file in `src/ploosh/connectors/` scans all files matching the pattern `connector_*.py`, imports them, and registers every class whose name starts with `Connector`.

There is **no registry to update** and **no configuration to change**: simply creating a correctly named file is enough.

```
src/ploosh/connectors/
├── connector.py               # Base class
├── connector_csv.py           # Example: native connector
├── connector_csv_spark.py     # Example: Spark connector
├── connector_<your_name>.py   # ← Your new connector
└── __init__.py                # Auto-discovery logic
```

##### Step 1 — Create the connector file

Create a new file in `src/ploosh/connectors/` with the naming convention:

| Type | File name |
|------|-----------|
| Native (Pandas) | `connector_<name>.py` |
| Spark | `connector_<name>_spark.py` |

##### Step 2 — Implement the class

Every connector extends the `Connector` base class and must:

1. Set `name` — the identifier used in YAML files (`type: <NAME>`)
2. Set `connection_definition` — parameters declared in `connections.yml`
3. Set `configuration_definition` — parameters declared in test case YAML
4. Implement `get_data()` — returns a `pandas.DataFrame` (native) or a Spark `DataFrame` (Spark)
5. Set `executed_action` — the query, file path or command that was executed (used in logs)

###### Base class reference

```python
class Connector:
    name = None
    connection_definition = None
    configuration_definition = None
    is_spark = False
    spark = None
    executed_action = None

    def get_data(self, configuration: dict, connection: dict):
        return None

    def get_executed_action(self):
        return self.executed_action
```

###### Minimal native connector

```python
"""Connector to read data from FooBar"""

import pandas as pd
from connectors.connector import Connector


class ConnectorFooBar(Connector):
    """Connector to read data from FooBar"""

    def __init__(self):
        self.name = "FOOBAR"
        self.connection_definition = []
        self.configuration_definition = [
            {"name": "path"},
        ]

    def get_data(self, configuration: dict, connection: dict):
        self.executed_action = configuration["path"]

        df = pd.read_csv(configuration["path"])
        return df
```

###### Minimal Spark connector

```python
"""Connector to read data from FooBar with Spark"""

from connectors.connector import Connector


class ConnectorFooBarSpark(Connector):
    """Connector to read data from FooBar with Spark"""

    def __init__(self):
        self.name = "FOOBAR_SPARK"
        self.is_spark = True
        self.connection_definition = []
        self.configuration_definition = [
            {"name": "path"},
        ]

    def get_data(self, configuration: dict, connection: dict):
        self.executed_action = configuration["path"]

        df = self.spark.read.format("foobar").load(configuration["path"])
        return df
```

> For Spark connectors, the Spark session is automatically injected into `self.spark` by the framework at startup.

##### Step 3 — Define parameters

Parameters are validated and resolved at runtime by [pyjeb](https://github.com/CSharplie/pyjeb) via the `control_and_setup` function. This library handles default values, type casting, required field validation and `validset` enforcement. You only need to declare the parameter definitions — Ploosh and pyjeb take care of the rest.

Each parameter is a dictionary with the following keys:

| Key | Required | Description |
|-----|----------|-------------|
| `name` | Yes | Parameter name, used as key in YAML |
| `default` | No | Default value. If absent, the parameter is **required** |
| `type` | No | Type cast: `string`, `integer`, `decimal`, `boolean`, `list`, `dict` |
| `validset` | No | List of allowed values |

###### connection_definition

These parameters are declared in `connections.yml` and shared across all test cases using this connection.

```python
self.connection_definition = [
    {
        "name": "mode",
        "default": "password",
        "validset": ["password", "connection_string"],
    },
    {"name": "hostname", "default": None},
    {"name": "database", "default": None},
    {"name": "username", "default": None},
    {"name": "password", "default": None},
    {"name": "port", "default": 3306, "type": "integer"},
    {"name": "connection_string", "default": None},
]
```

> Parameters with `"default": None` are optional. Parameters without `default` are required.

###### configuration_definition

These parameters are declared in the test case YAML, under `source` or `expected`.

```python
self.configuration_definition = [
    {"name": "query"},                           # Required
    {"name": "connection"},                      # Required
    {"name": "timeout", "type": "integer", "default": 30},  # Optional
]
```

##### Step 4 — Implement get_data()

The `get_data` method receives two dictionaries:

| Parameter | Description |
|-----------|-------------|
| `configuration` | Resolved test case parameters (from `configuration_definition`) |
| `connection` | Resolved connection parameters (from `connection_definition`), or `None` if no connection is needed |

The method must:

1. Set `self.executed_action` with a meaningful description (query, file path, etc.)
2. Return a DataFrame — `pandas.DataFrame` for native connectors, Spark `DataFrame` for Spark connectors

Parameter values are already validated and defaults are applied by the framework using `pyjeb.control_and_setup` before `get_data()` is called.

###### Example with connection

```python
def get_data(self, configuration: dict, connection: dict):
    hostname = connection["hostname"]
    database = connection["database"]
    query = configuration["query"]

    self.executed_action = query

    engine = create_engine(f"foobar://{hostname}/{database}")
    df = pd.read_sql(query, engine)
    return df
```

##### Step 5 — Add dependencies

If your connector requires an external Python package, you need to declare it in `pyproject.toml`.

###### 1. Core dependency

If your dependency is a **core library** needed by the framework itself (e.g. pandas, pyjeb), add it with a pinned version to the `dependencies` list under `[project]`:

```toml
#### pyproject.toml
[project]
dependencies = [
    # ... existing dependencies
    "my-package==1.2.3",
]
```

Core dependencies are always installed with `pip install ploosh`.

###### 2. Connector-specific dependency (extra)

If your dependency is **specific to your connector** (e.g. a database driver), add it as an optional dependency (extra) under `[project.optional-dependencies]`. Create a dedicated extra named after your connector, and also add it to the `full` extra:

```toml
#### pyproject.toml
[project.optional-dependencies]
my-connector = ["my-package==1.2.3"]

full = [
    # ... existing connector dependencies
    "my-package==1.2.3",
]
```

Users can then install only what they need:

```bash
pip install ploosh[my-connector]   # core + your connector
pip install ploosh[full]           # core + all connectors
```

> Import the optional dependency **inside** the connector's `get_data` method (not at module level), so that the connector module can still be loaded when the extra is not installed.

##### Step 6 — Add unit tests

Create a test file in `tests/connectors/` named `test_<name>.py`.

The tests should:

1. Instantiate the connector
2. Prepare `configuration` and `connection` dictionaries
3. Call `control_and_setup` to apply defaults (same behavior as the framework)
4. Call `get_data` and validate the result

```python
import pandas as pd
import pytest
from pyjeb import control_and_setup
from ploosh.connectors.connector_foobar import ConnectorFooBar


@pytest.fixture
def connector():
    return ConnectorFooBar()


def test_get_data(connector):
    configuration = {"path": "./tests/.data/sample.csv"}
    connection = {}

    configuration = control_and_setup(configuration, connector.configuration_definition)
    connection = control_and_setup(connection, connector.connection_definition)

    df = connector.get_data(configuration, connection)

    assert not df.empty
    assert connector.executed_action == "./tests/.data/sample.csv"
```

Run the tests with:

```shell
pytest tests/connectors/test_foobar.py -v
```

##### Step 7 — Add documentation

Create a documentation page in `docs/connectors/native/` or `docs/connectors/spark/` following the existing format. Each connector page should include:

- A short description
- The list of connection parameters (if any) with types and defaults
- The list of configuration parameters with types and defaults
- A complete YAML example showing `connections.yml` and a test case

##### YAML usage after creation

Once the file is created, users can immediately use the connector in their test cases:

```yaml
#### connections.yml (only if connection_definition is not empty)
my_foobar:
  type: foobar
  hostname: localhost
  database: mydb
```

```yaml
#### test_case.yml
Test FooBar data:
  source:
    type: foobar
    connection: my_foobar
    query: "SELECT * FROM my_table"
  expected:
    type: csv
    path: ./expected/my_table.csv
```

> The `type` value in YAML is case-insensitive and maps to the `name` attribute of the connector class.

##### Checklist

- [ ] File created as `connector_<name>.py` in `src/ploosh/connectors/`
- [ ] Class extends `Connector` and name starts with `Connector`
- [ ] `name` is set (uppercase, unique)
- [ ] `connection_definition` is set (empty list `[]` if no connection needed)
- [ ] `configuration_definition` is set
- [ ] `is_spark = True` if it's a Spark connector
- [ ] `get_data()` returns a DataFrame
- [ ] `executed_action` is set in `get_data()`
- [ ] Unit tests added in `tests/connectors/`
- [ ] Documentation added in `docs/connectors/`

### Exporters

#### CSV

Source: https://ploosh.io/docs/exporters/csv

#### CSV exporter

The CSV exporter generates a flat CSV file with one row per test case and detailed Excel files for failed tests.

##### Output structure

```
{output_path}/csv/
├── test_results.csv              → Summary of all test cases
└── test_results/
    ├── Test case 1.xlsx          → Gap details (only for failed tests)
    └── Test case 2.xlsx
```

##### CSV columns

| Column | Description |
|--------|-------------|
| `execution_id` | Unique identifier for the test run |
| `name` | Test case name |
| `state` | Result: `passed`, `failed`, `error`, `notExecuted` |
| `source_start` | Source data loading start time |
| `source_end` | Source data loading end time |
| `source_duration` | Source loading duration in seconds |
| `source_count` | Number of rows in source dataset |
| `source_executed_action` | Query or path executed for source |
| `expected_start` | Expected data loading start time |
| `expected_end` | Expected data loading end time |
| `expected_duration` | Expected loading duration in seconds |
| `expected_count` | Number of rows in expected dataset |
| `expected_executed_action` | Query or path executed for expected |
| `compare_start` | Comparison start time |
| `compare_end` | Comparison end time |
| `compare_duration` | Comparison duration in seconds |
| `success_rate` | Percentage of matching rows (0.0 to 1.0) |
| `error_type` | Error category: `headers`, `count`, `data`, `compare` |
| `error_message` | Error description |
| `error_detail_file_path` | Path to the XLSX gap analysis file |

##### Usage

###### Command line

``` shell
ploosh --connections connections.yml --cases test_cases --export CSV
```

###### Python API

``` python
from ploosh import execute_cases

execute_cases(cases="test_cases", connections="connections.yml", path_output="./output")
#### Then manually set export format — CSV export is available via CLI --export flag
```

##### Gap analysis Excel files

For each failed test case, an XLSX file is generated showing the differences between source and expected datasets with side-by-side comparison of values.

#### JSON

Source: https://ploosh.io/docs/exporters/json

#### JSON exporter

The JSON exporter generates a structured JSON file with detailed results for each test case, plus Excel files for failed tests.

This is the default export format.

##### Output structure

```
{output_path}/json/
├── test_results.json             → Structured results for all test cases
└── test_results/
    ├── Test case 1.xlsx          → Gap details (only for failed tests)
    └── Test case 2.xlsx
```

##### JSON structure

``` json
[
  {
    "execution_id": "a1b2c3d4-...",
    "name": "Test case name",
    "state": "passed",
    "source": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.003298,
      "count": 150,
      "executed_action": "SELECT * FROM employees"
    },
    "expected": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.000061,
      "count": 150,
      "executed_action": "./data/expected.csv"
    },
    "compare": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.000465,
      "success_rate": 1.0
    },
    "error": {
      "type": null,
      "message": null,
      "detail_file_path": null
    }
  }
]
```

##### JSON properties

| Property | Description |
|----------|-------------|
| `execution_id` | Unique identifier for the test run |
| `name` | Test case name |
| `state` | Result: `passed`, `failed`, `error`, `notExecuted` |
| `source.start` / `end` / `duration` | Source loading timing |
| `source.count` | Number of rows in source dataset |
| `source.executed_action` | Query or path executed |
| `expected.start` / `end` / `duration` | Expected loading timing |
| `expected.count` | Number of rows in expected dataset |
| `expected.executed_action` | Query or path executed |
| `compare.start` / `end` / `duration` | Comparison timing |
| `compare.success_rate` | Percentage of matching rows |
| `error.type` | Error category: `headers`, `count`, `data`, `compare` |
| `error.message` | Error description |
| `error.detail_file_path` | Path to XLSX gap analysis file |

##### Usage

###### Command line

``` shell
ploosh --connections connections.yml --cases test_cases --export JSON
```

JSON is the default format, so `--export` can be omitted.

##### Gap analysis Excel files

For each failed test case, an XLSX file is generated showing the differences between source and expected datasets with side-by-side comparison of values.

#### TRX

Source: https://ploosh.io/docs/exporters/trx

#### TRX exporter

The TRX exporter generates test results in the Visual Studio Test Results (TRX) XML format. This format is compatible with Azure DevOps Test Plans and Visual Studio.

##### Output structure

```
{output_path}/trx/
├── test_results.xml                          → TRX results file
└── test_results/
    └── In/
        └── {execution_id}/
            ├── Test case 1.xlsx              → Gap details (only for failed tests)
            └── Test case 2.xlsx
```

##### TRX format

The XML file follows the Visual Studio TestRun schema and includes:

- **TestSettings**: Execution settings
- **ResultSummary**: Counters for total, executed, passed, failed, error, and notExecuted tests
- **TestDefinitions**: One entry per test case
- **Results**: Outcome, duration, and error details per test
- **TestEntries**: Links between definitions and results

##### Usage

###### Command line

``` shell
ploosh --connections connections.yml --cases test_cases --export TRX
```

###### Azure DevOps integration

The TRX format integrates directly with Azure DevOps Test Plans using the `PublishTestResults` task:

``` yaml
- task: PublishTestResults@2
  inputs:
    testResultsFormat: 'VSTest'
    testResultsFiles: '*.xml'
    searchFolder: 'output/trx/'
    mergeTestResults: true
    testRunTitle: '$(Build.DefinitionName)'
```

See [Azure DevOps pipeline](/docs/pipelines/azure-devops) for a complete pipeline example.

##### Gap analysis Excel files

For each failed test case, an XLSX file is generated in the `test_results/In/{execution_id}/` folder, showing a side-by-side comparison of differing values between source and expected datasets.

### Getting Started

#### Concepts

Source: https://ploosh.io/docs/getting-started/concepts

#### Core concepts

##### Test case

A test case is defined in a YAML file and consists of:

- A **source**: the data to validate (query result, file content, etc.)
- An **expected**: the reference data to compare against
- **Options** (optional): comparison settings (sort, cast, ignore, tolerance, etc.)

``` yaml
My test case:
  disabled: false
  options:
    sort:
      - column1
  source:
    type: mysql
    connection: my_connection
    query: SELECT * FROM my_table
  expected:
    type: csv
    path: ./expected_data.csv
```

##### Connectors

A connector defines how to read data from a source. Each connector has:

- A **type** identifier (e.g. `mysql`, `csv`, `parquet_spark`)
- Optional **connection parameters** (hostname, credentials, etc.) defined in a connections file
- **Configuration parameters** (query, path, etc.) defined in the test case

There are two families of connectors:

| Family | Engine | Connectors |
|--------|--------|------------|
| **Native** | Pandas | CSV, JSON, Parquet, Delta, Excel, Empty, MySQL, PostgreSQL, SQL Server, Snowflake, BigQuery, Databricks, ODBC, Analysis Services, Semantic Model |
| **Spark** | PySpark | CSV Spark, JSON Spark, Parquet Spark, Delta Spark, Empty Spark, SQL Spark, Fabric KQL Spark, Dremio Spark |

> ⚠️ Both source and expected must use the same family. You cannot mix a native connector with a Spark connector in the same test case.

##### Connections

Connections are defined in a separate YAML file and contain the parameters needed to connect to data sources (hostnames, credentials, ports, etc.). They are referenced by name in the test case configuration.

``` yaml
my_connection:
  type: mysql
  hostname: server.database.windows.net
  database: my_db
  username: user
  password: $var.db_password
```

##### Compare engine

The compare engine validates that source and expected datasets match. The comparison process follows three steps:

1. **Structure check**: Verify that both datasets have the same columns (case-insensitive, trimmed).
2. **Row count check**: Verify that both datasets have the same number of rows.
3. **Data comparison**: Compare data row by row, applying preprocessing options (trim, case_insensitive, tolerance).

###### Compare modes

| Mode | Description | Available in |
|------|-------------|--------------|
| **order** | Row-by-row comparison based on row position or sort order | Native, Spark |
| **join** | Match rows by join keys instead of position | Spark only |

##### Test case states

| State | Description |
|-------|-------------|
| `passed` | Source and expected datasets match |
| `failed` | Differences found between datasets |
| `error` | An exception occurred during execution |
| `notExecuted` | Test case is disabled |

##### Error types

When a test case fails or errors, an error type indicates the stage where the issue occurred:

| Error type | Description |
|------------|-------------|
| `headers` | Column structure mismatch |
| `count` | Row count mismatch |
| `data` | Data values differ between source and expected |
| `compare` | Exception during comparison |

##### Exporters

Exporters save test results to files. All exporters also generate **Excel files (`.xlsx`)** with detailed gap analysis for failed test cases.

| Format | Output file | Description |
|--------|-------------|-------------|
| JSON | `test_results.json` | Structured results with nested objects |
| CSV | `test_results.csv` | Flattened results, one row per test |
| TRX | `test_results.xml` | Visual Studio Test Results format, compatible with Azure DevOps |

For each failed test, an `.xlsx` file is generated containing only the rows and columns that differ, with `{column}_source` and `{column}_expected` side by side. This makes it easy to identify exactly where source and expected data diverge.

##### Custom parameters

Variables can be used in YAML files with the `$var.<name>` syntax and passed at runtime via `--p_<name>` arguments. This avoids hardcoding sensitive information like passwords.

``` yaml
#### In connections.yml
my_connection:
  type: mysql
  password: $var.db_password
```

``` shell
#### At runtime
ploosh --connections connections.yml --cases test_cases --p_db_password "secret"
```

#### Installation

Source: https://ploosh.io/docs/getting-started/installation

#### Installation

##### Requirements

- Python 3.11, 3.12 or 3.13

##### Install from PyPi

Ploosh is available on [PyPi](https://pypi.org/project/ploosh/) and can be installed using pip.

###### Core installation

By default, `ploosh` installs a lightweight core with the native (pandas-based) engine and file connectors only:

``` shell
pip install ploosh
```

###### Optional connectors (extras)

Connector-specific dependencies are provided as optional extras. Install only what you need:

``` shell
pip install "ploosh[mysql]"              # MySQL
pip install "ploosh[postgresql]"         # PostgreSQL
pip install "ploosh[snowflake,bigquery]" # multiple extras at once
```

Available extras:

| Extra | Connector / Feature | Main dependency |
|-------|---------------------|-----------------|
| `spark` | Spark engine and Spark connectors | `pyspark`, `delta-spark` |
| `mysql` | MySQL | `pymysql` |
| `postgresql` | PostgreSQL | `pg8000` |
| `sqlserver` | SQL Server | `pyodbc` |
| `odbc` | ODBC | `pyodbc` |
| `snowflake` | Snowflake | `snowflake-sqlalchemy` |
| `databricks` | Databricks | `databricks-sql-connector` |
| `bigquery` | BigQuery | `pandas-gbq`, `sqlalchemy-bigquery` |
| `xmla` | Analysis Services, Semantic Model (XMLA) | `pyadomd`, `azure-identity` |
| `fabric` | Fabric semantic model / warehouse | `semantic-link-labs` |

###### Full installation

Includes every connector (all extras above):

``` shell
pip install "ploosh[full]"
```

##### Install for Spark mode

The Spark engine is **not** included in the core installation. Install the `spark` extra to enable it:

``` shell
pip install "ploosh[spark]"
```

When running Ploosh inside a managed Spark environment (Microsoft Fabric, Databricks) that already provides PySpark, install the package directly in the notebook:

``` python
%pip install ploosh
```

See [Spark documentation](/docs/spark/overview) for detailed setup instructions.

#### Quick Start

Source: https://ploosh.io/docs/getting-started/quick-start

#### Quick start

This guide walks you through your first test case with Ploosh in 5 steps.

##### 1. Install Ploosh

``` shell
pip install ploosh
```

##### 2. Setup connection file

Create a file `connections.yml` with your database connections:

``` yaml
my_database:
  type: mysql
  hostname: my_server.database.windows.net
  database: my_database_name
  username: my_user_name
  password: $var.db_password
```

> Using `$var.db_password` instead of a hardcoded password allows you to pass it securely via the command line at runtime.

##### 3. Create test cases

Create a folder `test_cases/` and add a YAML file (e.g. `tests.yml`) with your test definitions:

``` yaml
Test aggregated data:
  options:
    sort:
      - gender
      - domain
  source:
    connection: my_database
    type: mysql
    query: |
      SELECT gender,
             RIGHT(email, LENGTH(email) - POSITION("@" IN email)) AS domain,
             COUNT(*) AS count
      FROM users
      GROUP BY gender, domain
  expected:
    type: csv
    path: ./data/expected_aggregation.csv

Test no invalid emails:
  source:
    connection: my_database
    type: mysql
    query: |
      SELECT *
      FROM users
      WHERE email NOT LIKE '%@%.%'
  expected:
    type: empty
```

##### 4. Run tests

``` shell
ploosh --connections "connections.yml" --cases "test_cases" --export "JSON" --p_db_password "my_secret_password"
```

During execution, the status of each test is displayed in real-time:

```
 ____  _                 _
|  _ \| | ___   ___  ___| |__
| |_) | |/ _ \ / _ \/ __| '_ \
|  __/| | (_) | (_) \__ \ | | |
|_|   |_|\___/ \___/|___/_| |_|

Initialization[...]
Start processing tests cases[...]
Test aggregated data [...] (1/2) - Started
Test aggregated data [...] (1/2) - Passed
Test no invalid emails [...] (2/2) - Started
Test no invalid emails [...] (2/2) - Passed
Summary[...]
Total: 2 | Passed: 2 | Failed: 0 | Error: 0
```

##### 5. Review results

A `test_results.json` file is generated in the `output/json/` folder:

``` json
[
  {
    "name": "Test aggregated data",
    "state": "passed",
    "source": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.003298
    },
    "expected": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.000061
    },
    "compare": {
      "start": "2024-02-05T17:08:36Z",
      "end": "2024-02-05T17:08:36Z",
      "duration": 0.000465
    }
  }
]
```

###### Gap analysis Excel files

When a test fails, an Excel file (`.xlsx`) is automatically generated in `output/json/test_results/` with a detailed gap analysis. The file contains a side-by-side comparison of the differing values:

| Column | Description |
|--------|-------------|
| `{column}_source` | Value from the source dataset |
| `{column}_expected` | Value from the expected dataset |

Only rows and columns with differences are included, making it easy to pinpoint exactly where the data diverges.

##### Next steps

- [Command line options](/docs/configuration/command-line) — All CLI arguments
- [Test case options](/docs/configuration/options) — Sort, cast, ignore, tolerance, etc.
- [Custom parameters](/docs/configuration/custom-parameters) — Secure variable substitution
- [Spark mode](/docs/spark/overview) — Run Ploosh on Microsoft Fabric or Databricks

#### What Is Ploosh

Source: https://ploosh.io/docs/getting-started/what-is-ploosh

#### What is Ploosh?

Ploosh is an automated testing framework designed for data projects. Based on YAML configuration files, it enables teams to quickly and simply define, execute, and report on data validation tests.

##### Why Ploosh?

Testing tools for application development are not necessarily suited for data and BI projects. Data systems are often chains of complex workflows with multiple dependencies, making it difficult to test the entire process. In the traditional development world, we are used to the sequence: CI (build & tests) / deployment / execution. In data projects, this becomes: CI (build only) / deployment / execution / tests.

Ploosh fills this gap by providing a dedicated framework for data testing.

##### Key benefits

- **Reduce testing effort**: With industrialized tests, teams can focus on development or creating complex and high-value test cases.
- **Reduce regression risks**: By continuously running tests, regressions can be detected quickly and fixed before they impact production.
- **Increase test quality**: When a new bug is detected, new test cases can be added to the framework to prevent recurrence.
- **Improve project quality**: With fewer regression bugs and a more efficient team, the product's quality improves.

##### How it works

A test case consists of two parts: a **source** (the data to validate) and an **expected** (the reference data). For a test to pass, the source must match the expected.

The framework offers three main components:

1. **Connectors**: Query data sources (databases, files, APIs) and store the result in a homogeneous format (DataFrame).
2. **Compare engine**: Compare, for each test case, the source data with the expected data through three successive steps: row count comparison, structural equality check, and row-by-row data comparison.
3. **Exporters**: Export test results in different formats (JSON, CSV, TRX) for integration with reporting tools or CI/CD pipelines.

##### Two execution modes

Ploosh provides two execution modes to adapt to different environments:

| Mode | Engine | Best for |
|------|--------|----------|
| **Native** | Pandas | Local execution, CI/CD agents, small to medium datasets |
| **Spark** | PySpark | Microsoft Fabric, Databricks, large distributed datasets |

> ⚠️ A Spark connector can only be used with another Spark connector. It is not possible to mix Spark and native connectors in the same test case.

##### Supported connectors

| Type | Native connectors | Spark connectors |
|-----------|-------------------|------------------|
| Databases | BigQuery, Databricks, Snowflake, SQL Server, PostgreSQL, MySQL, ODBC | SQL Spark, Dremio |
| Files | CSV, Excel, JSON, Parquet, Delta | CSV, JSON, Parquet, Delta |
| BI Tools | Analysis Services, Semantic Model (XMLA) | Fabric KQL |
| Utilities | Empty | Empty |

##### Supported export formats

| Format | Description |
|--------|-------------|
| JSON | JSON file with detailed results |
| CSV | CSV file with flattened results |
| TRX | Visual Studio Test Results XML format, compatible with Azure DevOps Test Plans |

All export formats also generate Excel files (XLSX) with detailed gap analysis for failed test cases.

### Pipelines

#### Azure Devops

Source: https://ploosh.io/docs/pipelines/azure-devops

Ploosh is easy to use and can be integrated with any CI/CD pipeline. 
The following steps are required to run Ploosh tests in Azure DevOps and publish the results into Azure DevOps Test Plans.

#### Exemple of pipeline

1. Install ODBC driver for SQL Server if SQL Server connector is used
2. Install Ploosh package from PyPi
3. Execute Ploosh
    1. Provide the connections file
    2. Provide the test cases folder
    3. Provide the export format (TRX for Azure DevOps Test Plans)
    4. Disable the failure flag to avoid the pipeline to fail if a test fails
    5. Provide the passwords as parameters from the variables group
4. Publish test results

```yaml
trigger:
- main

variables:
- group: demo
stages:
  - stage: 
    displayName: Build
    jobs:
      - job: 
        steps:
          - checkout: self
          - task: CmdLine@2
            displayName: Install ODBC driver for SQL Server
            inputs:
              script: |
                curl https://packages.microsoft.com/keys/microsoft.asc | sudo tee /etc/apt/trusted.gpg.d/microsoft.asc
                curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
                sudo apt-get update
                sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18
          - task: CmdLine@2
            displayName: Install ploosh
            inputs:
              script: |
                pip install "ploosh[mysql,sqlserver,postgresql]"
          - task: CmdLine@2
            displayName: Execute ploosh
            inputs:
              script: ploosh  --connections "connections.yml" --cases "test_cases" --export "TRX" --failure False --p_mysql_password_db "$(mysql_password)" --p_mssql_password_db "$(mssql_password)" --p_postgresql_password_db "$(postgresql_password)"
          - task: PublishTestResults@2
            inputs:
              testResultsFormat: 'VSTest'
              testResultsFiles: '*.xml'
              searchFolder: 'output/trx/'
              mergeTestResults: true
              testRunTitle: '$(Build.DefinitionName)'
```

#### Fabric Pipeline

Source: https://ploosh.io/docs/pipelines/fabric-pipeline

#### Fabric Pipeline

Ploosh can be orchestrated directly from a Microsoft Fabric Pipeline using a Notebook activity.

##### Pipeline setup

###### Step 1: Create the orchestration notebook

Follow the [Fabric notebook guide](/docs/spark/fabric-notebook) to create a notebook with parameterized execution.

###### Step 2: Create the pipeline

1. In your Fabric workspace, click **New** → **Data pipeline**
2. Add a **Notebook** activity
3. Configure the activity:
   - **Notebook**: Select the Ploosh orchestration notebook
   - **Base parameters**: Override `cases_sub_folder` and `cases_filter` as needed

###### Step 3: Parameterize the pipeline

You can pass parameters to the notebook to target specific test suites:

| Parameter | Type | Example | Description |
|-----------|------|---------|-------------|
| `cases_sub_folder` | string | `/daily_checks` | Subfolder within `ploosh_cases/` |
| `cases_filter` | string | `*.yaml` | Glob pattern for test case files |

###### Step 4: Add upstream activities

Chain the Ploosh notebook after your data pipeline activities:

```
[Ingest Data] → [Transform Data] → [Run Ploosh Tests] → [Send Notification]
```

##### Example pipeline structure

``` json
{
  "activities": [
    {
      "name": "Run Data Pipeline",
      "type": "Pipeline",
      "dependsOn": []
    },
    {
      "name": "Run Ploosh Tests",
      "type": "Notebook",
      "dependsOn": [
        {
          "activity": "Run Data Pipeline",
          "dependencyConditions": ["Succeeded"]
        }
      ],
      "parameters": {
        "cases_sub_folder": "/",
        "cases_filter": "*.yaml"
      }
    }
  ]
}
```

##### Scheduling

Configure the pipeline trigger:

- **Schedule**: Daily, hourly, or custom cron
- **Event-based**: After another pipeline completes
- **Manual**: On-demand execution

##### Monitoring

After execution:

1. Check the pipeline run status in Fabric
2. Review detailed results in the `ploosh_results` Delta table
3. Open the Power BI report for visual quality dashboard

See [Fabric reporting](/docs/spark/fabric-reporting) for dashboard setup.

#### Github Actions

Source: https://ploosh.io/docs/pipelines/github-actions

#### GitHub Actions

Ploosh can be integrated into GitHub Actions workflows for automated data testing.

##### Example workflow

``` yaml
name: Data Tests

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install ODBC Driver (if using SQL Server)
        run: |
          curl https://packages.microsoft.com/keys/microsoft.asc | sudo tee /etc/apt/trusted.gpg.d/microsoft.asc
          curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
          sudo apt-get update
          sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18

      - name: Install Ploosh
        run: pip install "ploosh[sqlserver]"

      - name: Run tests
        run: |
          ploosh \
            --connections "connections.yml" \
            --cases "test_cases" \
            --export "JSON" \
            --failure false \
            --p_db_password "${{ secrets.DB_PASSWORD }}"

      - name: Upload test results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: test-results
          path: output/
```

##### Using TRX format

If you want to integrate with test result viewers:

``` yaml
      - name: Run tests (TRX)
        run: |
          ploosh \
            --connections "connections.yml" \
            --cases "test_cases" \
            --export "TRX" \
            --failure false \
            --p_db_password "${{ secrets.DB_PASSWORD }}"

      - name: Upload TRX results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: trx-results
          path: output/trx/
```

##### Secrets management

Store sensitive values in GitHub repository secrets:

1. Go to **Settings** → **Secrets and variables** → **Actions**
2. Add secrets (e.g. `DB_PASSWORD`, `SNOWFLAKE_PASSWORD`)
3. Reference them in the workflow as `${{ secrets.SECRET_NAME }}`

##### Tips

- Set `--failure false` to ensure the workflow completes and uploads results even when tests fail
- Use the `if: always()` condition on the upload step to capture results regardless of test outcomes
- Store connection files in the repository (with `$var` references for secrets)

### Spark

#### Databricks

Source: https://ploosh.io/docs/spark/databricks

#### Databricks setup

This guide explains how to run Ploosh on Databricks using Spark mode.

##### Step 1: Install Ploosh

In the first cell of your Databricks notebook:

``` python
%pip install ploosh
```

##### Step 2: Restart Python

Databricks requires a Python restart after installing new packages:

``` python
dbutils.library.restartPython()
```

##### Step 3: Execute Ploosh

``` python
from ploosh import execute_cases

root_folder = "/Workspace/Shared"

execute_cases(
    cases=f"{root_folder}/cases",
    connections=f"{root_folder}/connections.yaml",
    spark_session=spark,
    path_output=f"{root_folder}/output"
)
```

##### File organization

Organize your files in the Databricks workspace:

```
/Workspace/Shared/
├── connections.yaml        → Connection definitions
├── cases/                  → Test case YAML files
│   ├── validation.yaml
│   └── quality_checks.yaml
└── output/                 → Generated results
    └── json/
        ├── test_results.json
        └── test_results/
            └── *.xlsx
```

##### Using Unity Catalog

When using Databricks Unity Catalog, you can query tables directly with the `sql_spark` connector:

``` yaml
Test catalog data:
  source:
    type: sql_spark
    query: |
      SELECT *
      FROM my_catalog.my_schema.employees
      WHERE department = 'Engineering'
  expected:
    type: csv_spark
    path: /Workspace/Shared/expected/employees.csv
```

##### Using variables

Pass variables to Ploosh for dynamic configuration:

``` python
execute_cases(
    cases=f"{root_folder}/cases",
    connections=f"{root_folder}/connections.yaml",
    spark_session=spark,
    variables={
        "env": "production",
        "db_password": dbutils.secrets.get("my-scope", "db-password")
    }
)
```

##### Scheduling

Use Databricks **Jobs** to schedule Ploosh execution:

1. Create a new Job
2. Add a Notebook task pointing to your Ploosh notebook
3. Configure schedule (cron, trigger, or manual)
4. Optionally add parameters to override notebook widgets

#### Fabric Notebook

Source: https://ploosh.io/docs/spark/fabric-notebook

#### Fabric notebook orchestration

This page provides a complete notebook implementation for running Ploosh in Microsoft Fabric and tracking results over time.

##### Notebook structure

The notebook is composed of 4 cells:

1. **Parameters** — Input variables for flexible execution
2. **Imports** — Load the required libraries
3. **Ploosh execution** — Run the test cases
4. **Results history** — Persist results into a Delta table

##### Cell 1: Parameters

``` python
#### Parameters (can be overridden by Fabric Pipeline)
cases_sub_folder = "/"
cases_filter = "*.yaml"
```

These parameters enable flexible execution:
- `cases_sub_folder`: Target a specific subfolder within `ploosh_cases/`
- `cases_filter`: Filter which YAML files to process

##### Cell 2: Imports

``` python
from ploosh import execute_cases
from pyspark.sql.functions import col, lit, to_date
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    LongType, TimestampType
)
```

##### Cell 3: Ploosh execution

``` python
output_folder = "ploosh_outputs"

cases_folder_path = f"/lakehouse/default/Files/ploosh_cases{cases_sub_folder}"
connections_file = f"/lakehouse/default/Files/ploosh_connections.yaml"
output_path = f"/lakehouse/default/Files/{output_folder}"

execute_cases(
    cases=cases_folder_path,
    connections=connections_file,
    spark_session=spark,
    filter=cases_filter,
    path_output=output_path
)
```

> The `spark` variable is automatically available in Fabric notebooks.

##### Cell 4: Results history

This cell reads the JSON output generated by Ploosh and appends it to a Delta table for historical tracking.

``` python
spark_output_path = f"Files/{output_folder}/json/test_results.json"

schema = StructType([
    StructField("execution_id", StringType(), True),
    StructField("name", StringType(), True),
    StructField("state", StringType(), True),
    StructField("source", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("count", LongType(), True),
        StructField("executed_action", StringType(), True)
    ]), True),
    StructField("expected", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("count", LongType(), True),
        StructField("executed_action", StringType(), True)
    ]), True),
    StructField("compare", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("success_rate", DoubleType(), True)
    ]), True),
    StructField("error", StructType([
        StructField("type", StringType(), True),
        StructField("message", StringType(), True),
        StructField("detail_file_path", StringType(), True)
    ]), True)
])

df = spark.read.schema(schema).option("multiline", "true").json(spark_output_path)

flatten_cols = [
    col("execution_id"),
    col("name"),
    col("state"),
    to_date(col("source.start")).alias("execution_date"),

    col("source.start").alias("source_start"),
    col("source.end").alias("source_end"),
    col("source.duration").alias("source_duration"),
    col("source.count").cast("integer").alias("source_count"),
    col("source.executed_action").alias("source_executed_action"),

    col("expected.start").alias("expected_start"),
    col("expected.end").alias("expected_end"),
    col("expected.duration").alias("expected_duration"),
    col("expected.count").cast("integer").alias("expected_count"),
    col("expected.executed_action").alias("expected_executed_action"),

    col("compare.start").alias("compare_start"),
    col("compare.end").alias("compare_end"),
    col("compare.duration").alias("compare_duration"),
    col("compare.success_rate").alias("compare_success_rate"),

    col("error.type").alias("error_type"),
    col("error.message").alias("error_message"),
    col("error.detail_file_path").alias("error_detail_file_path"),
]

df_flat = df.select(*flatten_cols)
df_flat.write.mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("ploosh_results")
```

##### Results table schema

The `ploosh_results` Delta table contains:

| Column | Type | Description |
|--------|------|-------------|
| `execution_id` | string | Unique identifier for the test run |
| `name` | string | Test case name |
| `state` | string | Result: passed, failed, error |
| `execution_date` | date | Date of execution |
| `source_start` | timestamp | Source data loading start time |
| `source_end` | timestamp | Source data loading end time |
| `source_duration` | double | Source loading duration in seconds |
| `source_count` | integer | Number of rows in source dataset |
| `source_executed_action` | string | Query or path executed for source |
| `expected_start` | timestamp | Expected data loading start time |
| `expected_end` | timestamp | Expected data loading end time |
| `expected_duration` | double | Expected loading duration in seconds |
| `expected_count` | integer | Number of rows in expected dataset |
| `expected_executed_action` | string | Query or path executed for expected |
| `compare_start` | timestamp | Comparison start time |
| `compare_end` | timestamp | Comparison end time |
| `compare_duration` | double | Comparison duration in seconds |
| `compare_success_rate` | double | Percentage of matching rows (0.0 to 1.0) |
| `error_type` | string | Error category (headers, count, data, compare) |
| `error_message` | string | Error description |
| `error_detail_file_path` | string | Path to XLSX gap analysis file |

##### Pipeline integration

This notebook can be called from a Fabric Pipeline using a **Notebook activity**:

1. Create a new Pipeline
2. Add a **Notebook** activity
3. Point it to the orchestration notebook
4. Override parameters (`cases_sub_folder`, `cases_filter`) as needed
5. Schedule the pipeline or trigger it after upstream pipelines complete

#### Fabric Reporting

Source: https://ploosh.io/docs/spark/fabric-reporting

#### Fabric reporting

By persisting Ploosh test results into a Delta table (see [Fabric notebook](/docs/spark/fabric-notebook)), you can build Power BI reports to monitor data quality over time.

##### Semantic model

Create a Semantic Model on top of the `ploosh_results` Delta table in your Lakehouse:

1. In your Ploosh workspace, click **New** → **Semantic model**
2. Select the `ploosh_results` table from the Lakehouse
3. Define a date hierarchy on `execution_date` for time-based analysis
4. Publish the model

##### Suggested measures

| Measure | DAX formula |
|---------|-------------|
| Total tests | `COUNTROWS('ploosh_results')` |
| Passed tests | `CALCULATE(COUNTROWS('ploosh_results'), 'ploosh_results'[state] = "passed")` |
| Failed tests | `CALCULATE(COUNTROWS('ploosh_results'), 'ploosh_results'[state] = "failed")` |
| Error tests | `CALCULATE(COUNTROWS('ploosh_results'), 'ploosh_results'[state] = "error")` |
| Pass rate | `DIVIDE([Passed tests], [Total tests], 0)` |
| Avg source duration | `AVERAGE('ploosh_results'[source_duration])` |
| Avg compare duration | `AVERAGE('ploosh_results'[compare_duration])` |

##### Dashboard layout

A typical data quality dashboard includes:

###### Overview page
- **KPI cards**: Total tests, pass rate, failed count, error count
- **Trend chart**: Pass rate over time (by `execution_date`)
- **Table**: Latest execution results with state, duration, and error details

###### Detail page
- **Filter by test name**: Drill into a specific test case history
- **Duration chart**: Source/expected/compare durations over time
- **Error breakdown**: Error types distribution (headers, count, data, compare)

###### Operational page
- **Execution timeline**: Gantt-style view of test execution times
- **Success rate heatmap**: Success rates by test case and date
- **Alert list**: Tests with success rate below threshold

##### Alerting

Use Power BI data alerts or Fabric Data Activator to trigger notifications when:

- A test case fails for the first time
- The overall pass rate drops below a threshold
- A test case duration exceeds a limit (potential performance regression)

#### Fabric Setup

Source: https://ploosh.io/docs/spark/fabric-setup

#### Microsoft Fabric setup

This guide details how to set up Ploosh in a Microsoft Fabric environment to validate your data platform workloads.

##### Architecture overview

The recommended architecture is based on a **dedicated Fabric workspace** structured as follows:

```
Ploosh Workspace
├── Python Environment     → Ploosh package pre-installed
├── Lakehouse
│   ├── Files/
│   │   ├── ploosh_cases/         → Test case definitions (YAML)
│   │   ├── ploosh_connections.yaml → Connection definitions
│   │   ├── ploosh_resources/     → Reference datasets (CSV, Parquet, etc.)
│   │   └── ploosh_outputs/       → Test results (JSON, XLSX)
│   ├── Tables/
│   │   └── ploosh_results        → Results history (Delta table)
│   └── Shortcuts/                → Links to other workspace Lakehouses
├── Notebook                → Orchestration notebook
├── Semantic Model          → Results exposure for analysis
└── Power BI Report         → Quality dashboard
```

##### Step 1: Create the Python environment

1. In your Fabric workspace, create a new **Environment**
2. In the environment settings, add `ploosh` as a pip package
3. Save and publish the environment

This ensures Ploosh is available by default in all notebooks using this environment.

##### Step 2: Create the Lakehouse

Create a Lakehouse named (e.g. `ploosh_lakehouse`) and organize the `Files/` folder:

| Folder | Purpose |
|--------|---------|
| `ploosh_cases/` | Test case YAML files |
| `ploosh_resources/` | Reference data files (CSV, JSON, Parquet) used in expected tests |
| `ploosh_outputs/` | Output folder for test results |

Upload your `ploosh_connections.yaml` file to `Files/`.

##### Step 3: Configure shortcuts

To access data located in other Fabric workspaces, use **shortcuts**:

1. In the Lakehouse, go to the **Tables** section
2. Click **New shortcut**
3. Select the source (OneLake, Azure Data Lake, etc.)
4. Map the target Lakehouse tables from other workspaces

This makes remote tables queryable via Spark SQL as if they were local.

See [Shortcuts strategy](/docs/spark/fabric-shortcuts) for more details.

##### Step 4: Create the connections file

Create a `ploosh_connections.yaml` file for your Fabric data sources:

``` yaml
#### For KQL databases
kql_connection:
  type: fabric_kql_spark
  connection_mode: native
  kusto_uri: https://mycluster.kusto.windows.net
  database_id: my_kql_database

#### No connection needed for Spark SQL (uses shortcuts)
```

> Spark SQL queries against Lakehouse tables via shortcuts do not require a connection definition. Use the `sql_spark` connector directly.

##### Step 5: Create test cases

Create YAML files in `ploosh_cases/`:

``` yaml
Test employee count:
  source:
    type: sql_spark
    query: |
      SELECT department, COUNT(*) AS employee_count
      FROM hr_lakehouse.employees
      GROUP BY department
  expected:
    type: sql_spark
    query: |
      SELECT department, expected_count AS employee_count
      FROM ploosh_resources.expected_employee_counts

Test no KQL anomalies:
  source:
    type: fabric_kql_spark
    connection: kql_connection
    query: |
      AnomalyEvents
      | where Timestamp > ago(1d)
      | where Severity == "Critical"
  expected:
    type: empty_spark
```

##### Step 6: Create the orchestration notebook

See [Fabric notebook orchestration](/docs/spark/fabric-notebook) for a complete notebook implementation.

##### Step 7: Schedule execution

You can automate test execution through:

- **Fabric Pipeline**: Add a Notebook activity pointing to the orchestration notebook
- **Schedule**: Configure a recurring schedule on the notebook directly
- **Event trigger**: Trigger tests after upstream pipeline completion

See [Fabric pipeline integration](/docs/pipelines/fabric-pipeline) for pipeline examples.

#### Fabric Shortcuts

Source: https://ploosh.io/docs/spark/fabric-shortcuts

#### Fabric shortcuts strategy

When running Ploosh in Microsoft Fabric, your test cases often need to query data located in **multiple workspaces**. Fabric shortcuts provide a mechanism to make remote data accessible locally without copying it.

##### The problem

In a typical Fabric environment, data is distributed across multiple workspaces:

- **Workspace A**: Raw data Lakehouse (ingestion)
- **Workspace B**: Data warehouse / transformed data
- **Workspace C**: Reporting / datamart

Ploosh needs to access tables from all these workspaces to run cross-layer validations.

##### The solution: shortcuts

Shortcuts create virtual links to data in other locations, making it queryable via Spark SQL from the Ploosh Lakehouse.

###### Types of shortcuts

| Source | Description |
|--------|-------------|
| **OneLake** | Link to another Fabric Lakehouse in the same or different workspace |
| **Azure Data Lake Storage** | Link to ADLS Gen2 storage |
| **Amazon S3** | Link to S3 buckets |

###### Creating a shortcut

1. Open your Ploosh Lakehouse
2. In the **Tables** section, click **New shortcut**
3. Select the source type (e.g. OneLake)
4. Navigate to the target workspace and Lakehouse
5. Select the tables to link
6. The tables appear as local tables in your Lakehouse

###### Querying shortcut data

Once shortcuts are created, the remote tables are queryable via Spark SQL using the `sql_spark` connector:

``` yaml
Test data warehouse employees:
  source:
    type: sql_spark
    query: |
      SELECT department, COUNT(*) AS count
      FROM dw_lakehouse.employees
      GROUP BY department
  expected:
    type: sql_spark
    query: |
      SELECT department, count
      FROM reporting_lakehouse.employee_summary
```

> No connection configuration is required for `sql_spark` — Spark SQL resolves tables through the Lakehouse metadata.

##### Combining shortcuts with KQL

For workloads that write events to KQL databases, use the `fabric_kql_spark` connector alongside `sql_spark`:

``` yaml
connections:
  kql_events:
    type: fabric_kql_spark
    connection_mode: native
    kusto_uri: https://mycluster.kusto.windows.net
    database_id: events_db
```

``` yaml
Test event completeness:
  source:
    type: fabric_kql_spark
    connection: kql_events
    query: |
      ProcessingEvents
      | where Timestamp > ago(1d)
      | summarize event_count = count() by Pipeline
  expected:
    type: sql_spark
    query: |
      SELECT pipeline AS Pipeline, expected_count AS event_count
      FROM ploosh_resources.expected_event_counts
```

##### Best practices

- **Organize shortcuts by source workspace**: Create a naming convention (e.g. `dw_lakehouse`, `raw_lakehouse`) to identify origins
- **Use reference tables**: Store expected data as tables or files in the Ploosh Lakehouse for tests that don't compare two live sources
- **Minimize shortcut count**: Only link tables that are actually tested to reduce metadata overhead

#### Local Spark

Source: https://ploosh.io/docs/spark/local-spark

#### Local Spark setup

You can run Ploosh with a local Spark session for development and testing.

##### Prerequisites

- Python 3.9+
- Ploosh installed with the Spark extra: `pip install "ploosh[spark]"`

##### Usage

``` python
from pyspark.sql import SparkSession
from ploosh import execute_cases

#### Initialize a local Spark session
spark = SparkSession.builder \
    .appName("Ploosh") \
    .master("local[*]") \
    .getOrCreate()

#### Execute test cases
execute_cases(
    cases="test_cases",
    connections="connections.yml",
    spark_session=spark
)
```

##### When to use local Spark

- **Developing and debugging** Spark test cases before deploying to Fabric or Databricks
- **Testing Spark-specific features** like join mode comparison or Spark SQL queries
- **Working with local files** (CSV, JSON, Parquet, Delta) that you want to test with the Spark engine

##### Example test case

``` yaml
Test local delta:
  source:
    type: delta_spark
    path: ./data/employees_delta
  expected:
    type: csv_spark
    path: ./data/expected_employees.csv
    header: true
    inferSchema: true
```

##### Command line

You can also use Spark mode from the command line:

``` shell
ploosh --cases test_cases --connections connections.yml --spark true
```

When `--spark true` is set and no `spark_session` is provided, Ploosh automatically creates a local SparkSession.

#### Overview

Source: https://ploosh.io/docs/spark/overview

#### Spark mode overview

Ploosh supports two execution modes: **Native** (Pandas) and **Spark** (PySpark). Spark mode is designed to run within distributed environments like **Microsoft Fabric**, **Databricks**, or a **local Spark session**, enabling validation of large-scale datasets.

##### Why Spark mode?

| Benefit | Description |
|---------|-------------|
| **Distributed processing** | Leverage Spark clusters to validate large volumes of data |
| **Native platform access** | Query Lakehouse tables, KQL databases, and Delta files directly |
| **No data movement** | Data stays within the platform, avoiding costly exports |
| **Integrated execution** | Run tests directly from notebooks alongside your data pipelines |

##### When to use Spark mode?

| Scenario | Recommended mode |
|----------|-----------------|
| CI/CD pipeline on a build agent | Native |
| Local development with small datasets | Native |
| Microsoft Fabric notebooks | **Spark** |
| Databricks notebooks | **Spark** |
| Large datasets (millions of rows) | **Spark** |
| Querying Lakehouse/KQL/Delta files on a cluster | **Spark** |

##### Spark connectors

Spark mode uses dedicated connectors. You **cannot** mix Spark and native connectors in the same test case.

| Connector | Type | Description |
|-----------|------|-------------|
| `csv_spark` | File | Read CSV files via Spark |
| `json_spark` | File | Read JSON files via Spark |
| `parquet_spark` | File | Read Parquet files via Spark |
| `delta_spark` | File | Read Delta tables via Spark |
| `sql_spark` | Query | Execute Spark SQL queries |
| `fabric_kql_spark` | Database | Query Fabric KQL databases |
| `dremio_spark` | Database | Query Dremio via Arrow Flight SQL |
| `empty_spark` | Utility | Return an empty DataFrame |

##### Spark comparison engine

The Spark compare engine supports two comparison modes:

| Mode | Description |
|------|-------------|
| **order** (default) | Rows are matched by position using a `row_number()` window function |
| **join** | Rows are matched by specified `join_keys` columns (Spark only) |

The **join** mode is particularly useful when row ordering is not deterministic or when matching by business keys is more appropriate.

``` yaml
My test case:
  options:
    compare_mode: join
    join_keys:
      - employee_id
  source:
    type: sql_spark
    query: SELECT * FROM lakehouse.employees
  expected:
    type: csv_spark
    path: /lakehouse/default/Files/expected/employees.csv
```

##### Calling Ploosh from Python

In Spark mode, Ploosh is called programmatically from Python using the `execute_cases()` function:

``` python
from ploosh import execute_cases

execute_cases(
    cases="/path/to/cases",
    connections="/path/to/connections.yaml",
    spark_session=spark,
    filter="*.yaml",
    path_output="/path/to/output"
)
```

See the [Python API reference](/docs/api/execute-cases) for full details.

##### Platform-specific guides

- [Microsoft Fabric setup](/docs/spark/fabric-setup) — Complete guide for Fabric
- [Fabric notebook orchestration](/docs/spark/fabric-notebook) — Notebook implementation
- [Fabric shortcuts strategy](/docs/spark/fabric-shortcuts) — Cross-workspace data access
- [Fabric reporting](/docs/spark/fabric-reporting) — Power BI dashboards on test results
- [Databricks setup](/docs/spark/databricks) — Running Ploosh on Databricks
- [Local Spark](/docs/spark/local-spark) — Running Ploosh with a local SparkSession

### Use Cases

#### Data Quality

Source: https://ploosh.io/docs/use-cases/data-quality

#### Data quality checks

Ploosh can be used for operational data quality monitoring, ensuring data consistency, completeness, and correctness across your data platform.

##### Approaches

###### Absence of anomalies (empty approach)

Use the `empty` connector to verify that no problematic data exists:

``` yaml
Test no NULL emails:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT *
      FROM dim_customer
      WHERE email IS NULL AND is_active = 1
  expected:
    type: empty

Test no duplicate customer IDs:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT customer_id, COUNT(*) AS cnt
      FROM dim_customer
      GROUP BY customer_id
      HAVING COUNT(*) > 1
  expected:
    type: empty

Test no future dates:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT *
      FROM fact_orders
      WHERE order_date > GETDATE()
  expected:
    type: empty
```

###### Completeness checks

Verify that record counts match across layers:

``` yaml
Test row count consistency:
  source:
    connection: raw
    type: mssql
    query: |
      SELECT source_table, COUNT(*) AS row_count
      FROM raw.load_log
      WHERE load_date = CAST(GETDATE() AS DATE)
      GROUP BY source_table
      ORDER BY source_table
  expected:
    connection: dwh
    type: mssql
    query: |
      SELECT source_table, COUNT(*) AS row_count
      FROM dwh.load_log
      WHERE load_date = CAST(GETDATE() AS DATE)
      GROUP BY source_table
      ORDER BY source_table
```

###### Referential integrity

Verify that foreign key relationships are respected:

``` yaml
Test no orphan orders:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT f.order_id, f.product_id
      FROM fact_orders f
      LEFT JOIN dim_product p ON f.product_id = p.product_id
      WHERE p.product_id IS NULL
  expected:
    type: empty
```

###### Range validation

Verify that values fall within expected ranges:

``` yaml
Test valid percentages:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT *
      FROM fact_metrics
      WHERE success_rate < 0 OR success_rate > 1
  expected:
    type: empty
```

##### Spark mode for Fabric

The same quality checks work in Microsoft Fabric with Spark connectors:

``` yaml
Test no NULL customer IDs (Fabric):
  source:
    type: sql_spark
    query: |
      SELECT *
      FROM lakehouse.dim_customer
      WHERE customer_id IS NULL
  expected:
    type: empty_spark
```

See [Testing a Fabric data platform](/docs/use-cases/fabric-data-platform) for a complete Fabric use case.

##### Scheduling quality checks

For operational monitoring, schedule quality checks to run regularly:

- **Daily**: After nightly batch processing completes
- **After each pipeline run**: As a post-processing step
- **On demand**: When investigating data issues

Use the `disabled` option to temporarily skip checks that are under investigation:

``` yaml
Temporarily disabled check:
  disabled: true
  source:
    connection: dwh
    type: mssql
    query: SELECT * FROM problematic_table
  expected:
    type: empty
```

#### Fabric Data Platform

Source: https://ploosh.io/docs/use-cases/fabric-data-platform

#### Testing a Fabric data platform

This guide demonstrates a complete use case of Ploosh in Microsoft Fabric: validating workloads that write data to Lakehouse tables and events to KQL databases, distributed across multiple workspaces.

##### Architecture

```
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Ingestion WS   │     │  Transform WS   │     │  Reporting WS   │
│  ┌───────────┐  │     │  ┌───────────┐  │     │  ┌───────────┐  │
│  │ Raw       │  │     │  │ DW        │  │     │  │ Datamart   │  │
│  │ Lakehouse │  │     │  │ Lakehouse │  │     │  │ Lakehouse  │  │
│  └───────────┘  │     │  └───────────┘  │     │  └───────────┘  │
└────────┬────────┘     └────────┬────────┘     └────────┬────────┘
         │     shortcuts         │     shortcuts         │
         └───────────────┬───────┴───────────────────────┘
                         │
              ┌──────────▼──────────┐
              │    Ploosh WS        │
              │  ┌───────────────┐  │
              │  │ Ploosh        │  │
              │  │ Lakehouse     │  │
              │  │  ├ cases/     │  │
              │  │  ├ outputs/   │  │
              │  │  └ shortcuts/ │  │
              │  ├───────────────┤  │
              │  │ Notebook      │  │
              │  │ Semantic Model│  │
              │  │ PBI Report    │  │
              │  └───────────────┘  │
              └─────────────────────┘
```

##### Test scenarios

###### 1. Cross-layer validation

Verify that the transformation layer correctly aggregates raw data:

``` yaml
Test employee aggregation:
  options:
    sort:
      - department
  source:
    type: sql_spark
    query: |
      SELECT department, COUNT(*) AS employee_count
      FROM raw_lakehouse.employees
      GROUP BY department
  expected:
    type: sql_spark
    query: |
      SELECT department, employee_count
      FROM dw_lakehouse.dim_department_summary
```

###### 2. KQL event completeness

Verify that pipeline events are correctly logged:

``` yaml
Test pipeline events logged:
  source:
    type: fabric_kql_spark
    connection: kql_events
    query: |
      PipelineEvents
      | where Timestamp > ago(1d)
      | summarize event_count = count() by PipelineName
      | order by PipelineName asc
  expected:
    type: sql_spark
    query: |
      SELECT pipeline_name AS PipelineName, expected_count AS event_count
      FROM ploosh_resources.expected_daily_events
      ORDER BY PipelineName ASC
```

###### 3. Data quality checks

Detect anomalies using the empty connector:

``` yaml
Test no NULL customer IDs:
  source:
    type: sql_spark
    query: |
      SELECT *
      FROM dw_lakehouse.fact_orders
      WHERE customer_id IS NULL
  expected:
    type: empty_spark

Test no duplicate orders:
  source:
    type: sql_spark
    query: |
      SELECT order_id, COUNT(*) AS cnt
      FROM dw_lakehouse.fact_orders
      GROUP BY order_id
      HAVING cnt > 1
  expected:
    type: empty_spark
```

###### 4. Reference data validation

Compare Lakehouse data against known reference files:

``` yaml
Test product categories:
  options:
    sort:
      - category_id
  source:
    type: sql_spark
    query: |
      SELECT category_id, category_name, is_active
      FROM dw_lakehouse.dim_product_category
  expected:
    type: csv_spark
    path: /lakehouse/default/Files/ploosh_resources/expected_categories.csv
    header: true
    inferSchema: true
```

##### Connections file

``` yaml
kql_events:
  type: fabric_kql_spark
  connection_mode: native
  kusto_uri: https://mycluster.kusto.windows.net
  database_id: events_database
```

> Spark SQL queries against Lakehouse tables (via shortcuts) do not require a connection definition.

##### Results exploitation

After execution, the results are:
1. **Exported as JSON** for immediate review
2. **Persisted to a Delta table** (`ploosh_results`) for historical tracking
3. **Visualized in Power BI** through a Semantic Model

See [Fabric reporting](/docs/spark/fabric-reporting) for dashboard setup details.

##### End-to-end automation

1. **Upstream pipeline** completes data processing
2. **Fabric Pipeline** triggers the Ploosh notebook
3. **Ploosh** executes all test cases and exports results
4. **History tracking** appends results to the Delta table
5. **Power BI** refreshes and shows updated quality metrics
6. **Alerts** notify the team if tests fail

#### Migration Testing

Source: https://ploosh.io/docs/use-cases/migration-testing

#### Migration testing

Ploosh simplifies testing during data migration projects by allowing you to compare data between legacy and target systems.

##### Migration context

In a data migration project (e.g. on-premise to cloud), the key challenge is validating that data feeds produce the **same results** on both platforms. This involves:

- Migrating hundreds of processes/data feeds
- Handling gigabytes of data across many tables
- Ensuring complex calculations and business rules are preserved

##### Testing strategy

###### Workflow

1. **Code migration**: Develop new processing code for the target environment
2. **Deployment**: Install migrated code on the target platform
3. **Define testing scope**: Identify tables impacted by each feed
4. **Sampling**: Define representative data samples
5. **Write test cases**: Create Ploosh test cases comparing both environments
6. **Execute processes**: Run data feeds on both legacy and target environments
7. **Run tests**: Execute Ploosh to compare samples from both databases
8. **Analysis**: Review results — if a test passes, the feed produces identical results; otherwise, send back for correction

###### Connections setup

``` yaml
target_connection:
  type: mysql
  hostname: target-server.database.windows.net
  database: target_db
  username: user
  password: $var.target_password

legacy_connection:
  type: bigquery
  credentials: $var.bq_credentials
  credentials_type: service_account
```

##### Sampling strategies

Sampling is crucial for efficient testing. Two main approaches:

###### Precise sampling

Select rows using technical or business keys:

``` yaml
Sales migration test:
  options:
    sort:
      - sale_id
  source:
    connection: target_connection
    type: mysql
    query: |
      SELECT *
      FROM fact_sales
      WHERE sale_id IN ('E9EYKDIW6C', 'R5QUFFYXF0', 'FF0YIVG63B',
                         'DZWG6FJQWO', '7Y8G0EQ3JD', '9PF09Z3A6O')
      ORDER BY sale_id
  expected:
    connection: legacy_connection
    type: bigquery
    query: |
      SELECT *
      FROM fact_sales
      WHERE sale_id IN ('E9EYKDIW6C', 'R5QUFFYXF0', 'FF0YIVG63B',
                         'DZWG6FJQWO', '7Y8G0EQ3JD', '9PF09Z3A6O')
      ORDER BY sale_id
```

> Include a large number of values to cover more edge cases.

###### Batch sampling

Test a broader set using functional criteria:

``` yaml
Sales migration test:
  options:
    sort:
      - sale_id
  source:
    connection: target_connection
    type: mysql
    query: |
      SELECT *
      FROM fact_sales
      WHERE country = 'france'
      ORDER BY sale_id
  expected:
    connection: legacy_connection
    type: bigquery
    query: |
      SELECT *
      FROM fact_sales
      WHERE country = 'france'
      ORDER BY sale_id
```

> Always add `ORDER BY` clauses for deterministic comparison.

##### Useful options for migration

| Option | Use case |
|--------|----------|
| `sort` | Ensure consistent ordering between systems |
| `cast` | Handle type differences between platforms (e.g. INT vs BIGINT) |
| `tolerance` | Allow small numeric differences from floating-point precision |
| `case_insensitive` | Handle case differences in string data |
| `trim` | Handle whitespace differences between systems |
| `ignore` | Exclude columns that are expected to differ (timestamps, audit columns) |

##### Tips

- **Sort in queries**: Always add `ORDER BY` in your SQL queries for better performance and deterministic behavior
- **Iterate**: Start with simple tests, then add complexity as you find issues
- **Reuse test cases**: Once a fix is applied, re-run the same test to validate the correction
- **Parameterize**: Use `$var` parameters to switch between environments without modifying test files

#### Regression Testing

Source: https://ploosh.io/docs/use-cases/regression-testing

#### Regression testing

Ploosh can be used as a regression testing framework to ensure that changes to data pipelines do not break existing functionality.

##### Context

In data projects, every modification to the ETL chain introduces a risk of regression:

- New feature development
- Bug fixes
- Infrastructure changes
- Dependency updates
- Schema modifications

##### Strategy

###### Build a test suite incrementally

1. **Start with critical tests**: Focus on the most important tables and business rules
2. **Add tests on bug discovery**: When a bug is found, create a test case that would catch it
3. **Cover all layers**: Test across the entire data chain (raw → warehouse → datamart)

###### Example test suite

``` yaml
#### Regression test: employee count by department
Test employee count:
  options:
    sort:
      - department
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT department, COUNT(*) AS count
      FROM dwh.dim_employee
      WHERE is_active = 1
      GROUP BY department
  expected:
    connection: dmt
    type: mssql
    query: |
      SELECT department, employee_count AS count
      FROM dmt.department_summary
```

``` yaml
#### Regression test: no orphan records
Test no orphan orders:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT o.order_id
      FROM dwh.fact_orders o
      LEFT JOIN dwh.dim_customer c ON o.customer_id = c.customer_id
      WHERE c.customer_id IS NULL
  expected:
    type: empty
```

##### CI/CD integration

Run regression tests automatically after every deployment:

1. Deploy new code to the data platform
2. Execute data pipelines
3. Run Ploosh regression suite
4. Publish results to Azure DevOps / GitHub

See [Azure DevOps pipeline](/docs/pipelines/azure-devops) and [GitHub Actions](/docs/pipelines/github-actions) for integration guides.

##### Best practices

- **Version control test cases**: Store YAML files alongside pipeline code in Git
- **Run on every deployment**: Automate execution in CI/CD pipelines
- **Collaborative maintenance**: Tests can be written by developers, testers, and business analysts
- **Use pass_rate for tolerance**: Allow minor acceptable differences instead of strict matching
- **Disable flaky tests**: Use the `disabled` option to temporarily skip unstable tests while investigating

#### Testing Approaches

Source: https://ploosh.io/docs/use-cases/testing-approaches

#### Testing approaches

Ploosh is a flexible framework, but the intelligence of the tests lies in the people who write them. This guide presents three key approaches to structuring your data tests.

##### By recalculation

This approach validates the results produced by ETL processes by **recalculating the same operations** using independent SQL queries.

###### Principle

Write an SQL query that reproduces the calculations and transformations performed by the ETL on a given dataset. Then compare the results of this query to the ETL output. If both match, the test passes.

###### Example

Suppose an ETL aggregates sales data by category and region. To verify the results:

``` yaml
Test sales aggregation:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT category, region, SUM(sales_amount) AS total_sales
      FROM dwh.fact_sales
      GROUP BY category, region

  expected:
    connection: dmt
    type: mssql
    query: |
      SELECT category, region, total_sales
      FROM dmt.fact_sales_aggregated
```

###### Use cases

- Verification of complex aggregations (monthly sales by product and region)
- Calculation of ratios or KPIs derived from multiple sources
- Validation of business rules at each stage of data processing

###### Trade-offs

The main effort is writing SQL queries that replicate complex transformations. However, it provides fine control and pinpoints exactly where errors occur.

---

##### By empty

This approach relies on the **empty** connector for the expected part. The source query should return **no data**. If it does, the test fails.

###### Principle

Define a source query whose criteria identify **incorrect or undesirable data**. If the query returns results, it means anomalies exist. If no data is returned, the test passes.

###### Example

Check for rejected records inserted in the last 24 hours:

``` yaml
Test no recent rejects:
  source:
    connection: dwh
    type: mssql
    query: |
      SELECT *
      FROM dwh.fact_sales_rejects
      WHERE date_insert BETWEEN DATEADD(day, -1, GETDATE()) AND GETDATE()
  expected:
    type: empty
```

###### Use cases

- **Data quality**: Ensure invalid data is not present in target tables
- **Duplicate detection**: Verify no duplicate records exist
- **Consistency checks**: Ensure relationships and business rules are respected
- **NULL checks**: Verify no NULL values in mandatory columns
- **Range validation**: Ensure values fall within expected ranges

###### Advantages

- **Simplicity**: Only a source query is needed, no expected query
- **Fast implementation**: Rapid creation of tests for critical checks
- **Flexibility**: Adaptable to many data validation scenarios

---

##### By test data

This approach uses **predefined test data** inserted into source systems to simulate specific scenarios.

###### Principle

1. Insert known test data into the data sources
2. Run the ETL processes
3. Compare the output against pre-calculated expected results

###### Example

Using a CSV file with pre-calculated expected results:

``` yaml
Test transformation rules:
  options:
    sort:
      - employee_id
  source:
    connection: dmt
    type: mysql
    query: |
      SELECT employee_id, full_name, department, is_active
      FROM dmt.dim_employee
      WHERE employee_id IN (9001, 9002, 9003)
  expected:
    type: csv
    path: ./test_data/expected_employees.csv
```

###### Use cases

- Validating complex business rules with specific edge cases
- Testing data type conversions and format transformations
- Verifying calculations with known inputs and outputs

###### Trade-offs

Requires functional understanding of the system and effort to maintain test datasets. However, it provides the most deterministic and reproducible tests.

---

##### Choosing an approach

| Approach | Best for | Complexity | Maintenance |
|----------|----------|------------|-------------|
| **Recalculation** | Aggregations, KPIs, business rules | Medium | Medium |
| **Empty** | Quality checks, duplicates, anomalies | Low | Low |
| **Test data** | Edge cases, transformations, rules | High | High |

In practice, combine all three approaches for comprehensive test coverage.

---

## Blog

### Ploosh + Microsoft Fabric: The Easy Way to Test Your Data Platform

Date: 2026-01-30 — Author: Charlie Collier

Source: https://ploosh.io/blog/ploosh-microsoft-fabric-the-easy-way-to-test-your-data-platform

##### Introduction

This article details the implementation of the **Ploosh** automated testing framework within the **Microsoft Fabric** ecosystem. We will rely on a concrete use case: validating the results of workloads writing data to **Lakehouse** tables and events to **Kusto** tables, distributed across multiple workspaces.

**Ploosh** positions itself as a validation framework for data projects. It allows you to define tests declaratively and execute them automatically to generate quality reports, while alerting teams in case of non-compliance or anomalies.

The framework offers two execution modes:

- **Native mode**: Ideal for local execution or via a classic CI/CD agent.
- **Spark mode**: Designed to run within a Spark cluster, enabling distributed processing of large data volumes.

This use case specifically leverages **Spark mode** in **Microsoft Fabric**, thereby taking advantage of the platform's distributed computing power to efficiently validate data pipelines.

##### Architecture and Implementation

![](/assets/img/blog/ploosh-microsoft-fabric-the-easy-way-to-test-your-data-platform/schema-1.png)

The architecture is based on a dedicated **Fabric** workspace, structured as follows:

- **Python environment**: The **Ploosh** package is pre-installed and configured by default.
- **Central Lakehouse** hosting:
    1. Shortcuts to access data (**Lakehouse** tables) located in other workspaces.
    2. The definition of test cases (YAML files).
    3. Reference datasets.
    4. Raw test results (JSON files generated after each execution).
    5. A structured table for results history tracking.
- **Notebook**: Orchestrates test execution and history tracking.
- **Semantic model**: Exposes results for analysis.
- **Power BI reports**: Provide clear visualization of data quality.

###### The Execution Environment
The **Fabric** environment is configured with the **Ploosh** package (installed via `pip`). This configuration allows invoking **Ploosh** directly from a **Fabric** notebook using **Spark** as the execution engine. This is a major advantage for processing large volumes distributed across multiple workspaces. Using a dedicated environment ensures code portability and immediate availability of dependencies during execution.

###### The Central Role of Lakehouse
The **Lakehouse** is the cornerstone of this architecture. It acts as a single repository for tests (YAML definitions, reference data) and results. Most importantly, it centralizes access to heterogeneous data via the shortcuts mechanism.

![](/assets/img/blog/ploosh-microsoft-fabric-the-easy-way-to-test-your-data-platform/schema-2.png)

###### The Shortcuts Strategy
**Ploosh** has native connectors to interact with various sources.

- For **Kusto** (KQL Database): **Ploosh** uses the **Kusto Spark** connector, authenticating via cluster information (URI, database, table, credentials).
- For **Lakehouse** (Spark SQL): To execute **Spark SQL** queries on data located in other workspaces, they must be made visible locally. This is where shortcuts come in: all source **Lakehouses** are linked to the "**Ploosh**" workspace via shortcuts, making their tables queryable as if they were local.

###### File Organization
The folder structure within the **Lakehouse** is standardized:

- `ploosh_cases`: Contains test definitions (YAML).
- `ploosh_resources`: Hosts reference data needed for validations.
- `ploosh_outputs`: Receives output artifacts (statistics, statuses, Excel variance files).

###### Results History Tracking
A **Delta** table `ploosh_results` ensures results persistence. It captures key metrics such as test name, final status (Success/Failure), variance details, execution timestamp, and SQL queries executed. It transforms the ephemeral content of output JSON files into an exploitable history for monitoring.

###### The Orchestration Notebook

The **Fabric** notebook serves as the entry point for automation. Its logic is sequential:

1. **Loading**: It reads test definitions from the **Lakehouse**.
2. **Execution**: It launches the **Ploosh** engine in **Spark** mode. It accepts the target folder and file pattern as parameters, offering flexibility to execute a complete suite or a unit test.
3. **History tracking**: It ingests raw JSON results, transforms them to match the target schema, and inserts them into the history table.

Here is an example of a typical implementation of this type of notebook:

Cell 1: The parameters
```python
cases_sub_folder = "/"
cases_filter = "*.yaml"
```

Cell 2: The imports
```python
from ploosh import execute_cases
from pyspark.sql.functions import col, lit, to_date
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType, ArrayType, TimestampType, DateType
```

Cell 3: The ploosh execution
```python
output_folder = "ploosh_results"

cases_folder_path = f"/lakehouse/default/Files/ploosh_cases{cases_sub_folder}"
connections_file = f"/lakehouse/default/Files/ploosh_connections.yaml"
output_path = f"/lakehouse/default/Files/{output_folder}"

execute_cases(cases = cases_folder_path, connections = connections_file, spark_session = spark, filter=cases_filter, path_output=output_path)
```

Cell 4: The results history tracking
```python
spark_ouput_path = f"Files/{output_folder}/json/test_results.json"

schema = StructType([
    StructField("execution_id", StringType(), True),
    StructField("name", StringType(), True),
    StructField("state", StringType(), True),
    StructField("source", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("count", LongType(), True),
        StructField("executed_action", StringType(), True)
    ]), True),
    StructField("expected", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("count", LongType(), True),
        StructField("executed_action", StringType(), True)
    ]), True),
    StructField("compare", StructType([
        StructField("start", TimestampType(), True),
        StructField("end", TimestampType(), True),
        StructField("duration", DoubleType(), True),
        StructField("success_rate", DoubleType(), True)
    ]), True),
    StructField("error", StructType([
        StructField("type", StringType(), True),
        StructField("message", StringType(), True),
        StructField("detail_file_path", StringType(), True)
    ]), True)
])

df = spark.read.schema(schema).option("multiline", "true").json(spark_ouput_path)

name_parts = split(col("name"), "/")
flatten_cols_to_select = [
    # global metadata 
    col("execution_id").cast("string"),
    col("name").cast("string"),
    col("state").cast("string"),
    to_date(col("source.start")).alias("execution_date"),

    # source statistics
    col("source.start").cast("timestamp").alias("source_start"),
    col("source.end").cast("timestamp").alias("source_end"),
    col("source.duration").cast("double").alias("source_duration"),
    col("source.count").cast("integer").alias("source_count"),
    col("source.executed_action").cast("string").alias("source_executed_action"),

    # expected statistics
    col("expected.start").cast("timestamp").alias("expected_start"),
    col("expected.end").cast("timestamp").alias("expected_end"),
    col("expected.duration").cast("double").alias("expected_duration"),
    col("expected.count").cast("integer").alias("expected_count"),
    col("expected.executed_action").cast("string").alias("expected_executed_action"),

    # compare statistics
    col("compare.start").cast("timestamp").alias("compare_start"),
    col("compare.end").cast("timestamp").alias("compare_end"),
    col("compare.duration").cast("double").alias("compare_duration"),
    col("compare.success_rate").cast("double").alias("compare_success_rate"),

    # errors columns
    col("error.type").cast("string").alias("error_type"),
    col("error.message").cast("string").alias("error_message"),
    col("error.detail_file_path").cast("string").alias("error_detail_file_path"),
]

### Save results to table
df_flat = df.select(*flatten_cols_to_select)
df_flat.write.mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("ploosh_results")
```

###### Industrialization and DevOps
To fully integrate these tests into the software development lifecycle (SDLC), several improvement axes are possible:

- **Versioning (Git)**: Connect the workspace to a source control manager (**Azure DevOps**, **GitHub**) to version **YAML** tests and reference data. This ensures traceability and facilitates collaboration.
- **CI/CD**: Implement automatic deployment pipelines. Each modification in the repository would trigger an update of test files in the **Lakehouse**, guaranteeing constant alignment between code and tests.
- **Alerting**: Configure automatic notifications (**Teams**, **Outlook**) triggered by the notebook or a **Data Activator** in case of critical failure, to reduce response time (MTTR).
- **Advanced analytics**: Enrich **Power BI** reports with trend analyses to detect progressive data quality degradation before it becomes critical.

###### Conclusion

The integration of **Ploosh** in **Microsoft Fabric** is a robust response to data quality challenges at scale. By combining **Ploosh**'s declarative flexibility with the distributed computing power of **Spark** on **Fabric**, data teams have a powerful tool to ensure the reliability of their deliverables, while paving the way for a true **DataOps** approach.

Once the setup is in place, it is now a matter of applying the most relevant testing approaches. As I detailed in a previous article, [the 3 key approaches to automating tests with ploosh](https://ploosh.io/blog/ploosh-three-key-approaches-to-automating-tests-in-data-projects) are perfectly suited to **Microsoft Fabric** projects.

### ploosh: three key approaches to automating tests in data projects

Date: 2024-10-01 — Author: Charlie Collier

Source: https://ploosh.io/blog/ploosh-three-key-approaches-to-automating-tests-in-data-projects

##### Introduction

In previous articles, we introduced Ploosh as an automated testing framework, highlighting its role in preventing regressions and improving the quality of deliveries in complex data projects. We also demonstrated its effectiveness in a data migration context, where it was used to test data flows between a legacy system and a cloud platform.

In this article, we will explore new approaches to using Ploosh in more traditional data projects, focusing on regression testing, data quality checks, and validation through test datasets.

##### Context

![](/assets/img/blog/ploosh-three-key-approaches-to-automating-tests-in-data-projects/schema-1.png)

In this traditional data project, Ploosh was used to manage complex data flows feeding multiple layers of data, including sources in the form of files, a data warehouse, and a data mart. Each modification to this ETL chain, composed of complex calculations and specific business rules, introduced a risk of regression.

Initially, Ploosh was used to minimize these regression risks, ensuring optimal quality in the delivery of new features and fixes. Later, its use was extended to more operational checks, such as managing data rejections and verifying data completeness across the different layers.

Ploosh was fully integrated into Azure DevOps, with tests planned and executed automatically in the pipelines. The test cases, stored in a Git repository, were designed and maintained by the entire team (developers, testers, and business analysts), promoting seamless collaboration.

##### Approaches

Ploosh is a tool designed to simplify test creation, but the person writing the tests remains central to the process. Although Ploosh allows for the creation of any type of test, the method of implementation depends on the approach chosen. In this project, three main approaches were used, although other options are also possible.

###### **By recalculation**

This approach involves validating the results produced by the ETL processes by recalculating the same operations using independent SQL queries. The idea is to replicate in SQL the transformations performed by the ETL to ensure that the final results are accurate and meet expectations. This ensures that the business rules and complex calculations applied during the transformations are correctly implemented.

###### Principle

The recalculation approach involves writing an SQL query that reproduces the calculations and transformations performed by the ETL on a given dataset. Then, the results obtained by this SQL query are compared to the results from the ETL process. If the results are identical, the test is validated; otherwise, it fails, indicating a potential error in the ETL process.

This method is particularly useful in cases where complex calculations or aggregations are performed, as it ensures that the ETL follows the business rules and that the data is handled correctly at each stage.

###### Exemple

Let’s suppose an ETL process aggregates sales data and groups it by category and region. To verify the accuracy of the results, you can recalculate these aggregations with a simple SQL query:

``` yaml
Test sales aggregation:
  source:
    connection: demo_sql
    type: MSSQL
    query: |
      SELECT category, region, SUM(sales_amount) AS total_sales
      FROM dwh.fact_sales
      GROUP BY category, region

  expected:
    connection: demo_sql
    type: MSSQL
    query: |
      SELECT category, region, total_sales
      FROM dmt.fact_sales_aggregated
```


In this example, the source query and the expected query are identical, as they simulate the aggregation that the ETL process would perform. The results of the two queries are then compared to ensure that the ETL correctly applied the business rules during the aggregation of sales data.

###### **Use cases**

This approach can be used in several types of scenarios, such as:

*   **Verification of complex aggregations**: For example, calculating monthly sales by product and by region.
*   **Calculation of ratios or indicators**: Checking the accuracy of KPIs (Key Performance Indicators) derived from multiple data sources.
*   **Validation of business rules**: Ensuring that business rules are properly applied at each stage of data processing.

###### Disadvantages

The main disadvantage of this approach is the time and effort required to write SQL queries that replicate complex transformations. However, it provides fine control and allows you to identify exactly where errors may occur in the ETL processes.

##### By empty

This approach stands out for its simplicity and effectiveness, as it only requires half of a typical test case. Instead of defining both a source query and an expected query, this method relies on the use of the **“empty”** connector for the expected part of the test. The idea is to define a source query that should return no data, meaning that the data is considered incorrect. If Ploosh receives data in response to this query, the test is considered to have failed.

##### Principle

The test is based on the assumption that the data source should not contain any results that meet the query’s criteria. If the query returns data, this indicates that an anomaly has been detected, causing the test to fail. Conversely, if no data is returned, the test is validated.

This method is particularly suitable for checks such as:

*   Detecting **incorrect or invalid data** (e.g., values outside the allowed ranges),
*   Verifying the **absence of duplicates**,
*   Ensuring **data consistency** (absence of conflicts or inconsistencies in the datasets).

###### Example

Let’s take an example to verify the absence of rejected records in a table called `fact_sales_rejects`, which contains entries rejected during the ETL process.

``` yaml
Test sales rejects:
  source:
    connection: demo_sql
    type: MSSQL
    query: |
      SELECT * 
      FROM dwh.fact_sales_rejects
      WHERE date_insert BETWEEN DATEADD(day, -1, GETDATE()) AND GETDATE()

  expected:
    type: empty
```


In this example, the source query checks the `fact_sales_rejects` table to retrieve data inserted in the last 24 hours. The `empty` connector is used for the expected part of the test, meaning Ploosh expects no data to be returned. If any data is found, this signals a problem (abnormal or unresolved rejections), and the test fails. If no data is returned, the test passes.

###### **Use cases**

This approach is extremely flexible and can be applied to various control scenarios:

*   **Data quality checks**: Ensuring that invalid data is not present in target tables.
*   **Duplicate verification**: Ensuring that no duplicate records exist in a given table.
*   **Consistency checks**: Ensuring that relationships and business rules are respected in the datasets.

###### **Advantages**

*   **Simplicity**: You only need to define a source query without creating an expected query. The `empty` connector significantly simplifies the test definition.
*   **Fast implementation**: This method allows for rapid creation of tests to check for the absence of undesirable or inconsistent data.
*   **Flexibility**: It can be used in various contexts to quickly verify critical aspects of the data system.

###### Variants

You can adapt this approach by slightly modifying the criteria in the source query to test different aspects of the data. For example, it can be used to check for the absence of `NULL` values in specific columns or to ensure that certain thresholds or business rules are not violated.

##### **By test data**

This approach is particularly effective for validating the accuracy of calculations and business rules in data projects. However, it requires a solid functional understanding of the system being tested. The idea is to insert predefined **test data** into the project’s data sources to simulate specific scenarios. These data are then compared to expected results, which have already been calculated, to ensure that the processes comply with the established business rules.

One major advantage of this method is that it allows for the control of specific cases while ensuring that the sampling of test data remains consistent and reliable. By doing so, potential regressions can be quickly identified. The test data are injected permanently into the data flow (for example, via a Y-flow) without altering the real production data.

###### **Implementation steps**

1.  **Set up a Y-flow** to inject the test source files, ensuring that these data do not modify the production sources.
2.  **Create test source files** that contain the datasets necessary to populate the relevant tables (e.g., the `fact_sales` table).
3.  **Develop the expected results files**, which will be used to compare the test output with known and validated data.
4.  **Write the Ploosh test case**, which will compare the data generated from the test files with the expected results.

![](/assets/img/blog/ploosh-three-key-approaches-to-automating-tests-in-data-projects/schema-2.png)

###### Example

Suppose the `fact_sales` table in a data warehouse is populated by two source files (`sales_headers.csv` and `sales_content.csv`), which are ingested via ETL processes. These processes apply complex business rules to transform the data.

Here’s how a test can be implemented to verify the integrity of this process:

``` yaml
Test sales:
  source:
    connection: demo_sql
    type: MSSQL
    query: | 
      SELECT * 
      FROM dwh.fact_sales
      WHERE customer = 'Test Customer'

  expected:
    type: csv
    path: ./data/dwh/facts/fact_sales.csv
```


In this example, the SQL query retrieves data for a specific test customer (`Test Customer`) from the `fact_sales` table in the data warehouse. The results obtained are then compared to a pre-existing CSV file, which contains the expected results. This file has already been validated, ensuring that any discrepancies detected during test execution indicate a potential regression in the ETL processes.

###### **Advantages**

*   **Quick regression detection**: Since the input data (test files) and expected results (output files) are fixed, any unexpected variation in the results can be quickly identified and corrected.
*   **Controlled sampling**: Using predefined test datasets ensures that the tests cover a wide range of scenarios, including edge cases, exceptions, or special cases.
*   **Reliability**: Because the test datasets and expected results are constant, tests can be run with each code modification to ensure system stability, without needing to rewrite the tests at each iteration.

##### **Conclusion**

Ploosh is a versatile tool, extremely useful for securing ETL workflow changes and ensuring data integrity at each stage of processing. The various approaches presented here demonstrate how Ploosh can adapt to a wide range of needs, whether for regression testing, data quality control, or specific validations. Its agnosticism and flexibility make it a valuable asset for technical teams and business analysts, who can collaborate in the creation and execution of tests.

As Ploosh evolves, new features and connectors will further enhance its capabilities. In the meantime, you can install it (see the guide on the wiki) or check out the source code available on [GitHub](https://github.com/CSharplie/ploosh/).

* * *

See original post on LinkedIn (in french) : [Ploosh : comment faciliter ses tests de migration ?](https://www.linkedin.com/pulse/ploosh-comment-faciliter-ses-tests-de-migration-charlie-collier-3cmpe/)

### ploosh: how to simplify your migration testing?

Date: 2024-09-17 — Author: Charlie Collier

Source: https://ploosh.io/blog/ploosh-how-to-simplify-your-migration-testing

##### Introduction

**In a previous article, I introduced Ploosh, a tool I developed to facilitate testing in the data domain. Today, I will show you a use case where Ploosh was used to improve efficiency during testing phases. This article presents how Ploosh was implemented in a migration project. In a future article, I will demonstrate how it is used in more traditional BI and data projects.**

##### Migration Context

We used Ploosh as part of a cloud migration, a classic use case since we had to manage two heterogeneous data systems: a legacy on-premise system and a target system in SaaS.

Beyond platform and security aspects, when we think of data migration, we generally distinguish between two main areas: data migration and the migration of data feeds. I will elaborate on the second point, although the method can also be applied to data migration.

The scope of the migration included hundreds of processes (or data feeds) handling gigabytes of data across nearly a thousand tables. We used Ploosh to reduce the heavy testing load required to validate the proper functioning of the new data feeds on the target platform.

##### Testing Strategy

To successfully complete the testing phase, we defined a strategy to cover as many cases as possible while optimizing performance to test a large number of elements.

![](/assets/img/blog/ploosh-three-key-approaches-to-automating-tests-in-data-projects/schema-1.png)

1.  **Code migration:** Development of the new processing code ready to be executed on the target environment. The first tests aim to validate the technical functionality. Some basic tests are also carried out at this stage. It is worth noting that the testing phase can begin in parallel with development to optimize the timeline.
2.  **Deployment:** Installation of the migrated code on the target environment so it is ready to be executed and tested.
3.  **Defining the testing scope:** Identification of the elements to be tested. Most feeds write or update tables. This step identifies the tables impacted by the feed being tested, so the tests can focus on the right elements. This is purely a technical step, based on reading and understanding the code to be migrated.
4.  **Sampling:** Defining a representative sample of the data processed by the feed. The idea is to define a sample large enough to cover as many cases as possible, while small enough to ensure the tests are executed in a reasonable amount of time. This step requires a good understanding of the feed to ensure that as many cases as possible are covered. Filters can be very specific (customer IDs, order numbers) or broader (cities, services).
5.  **Writing test cases:** Developing the test case with Ploosh. The objective is to create one test case per table to be tested, using the same query and sample to run it on both the legacy and target environments.
6.  **Executing processes:** Running the data feeds on both the legacy and target environments to align the two environments in terms of data.
7.  **Running the tests:** Executing the test cases to compare the samples from both databases and validate or invalidate the test results.
8.  **Analysis:** Analyzing the test case results with Ploosh. If the test case passes, it means the feed produces the same result on the target environment as on the legacy one. Otherwise, the feed is sent back for correction and retested with the same test case.

##### Sampling Example

As mentioned earlier, sampling is a crucial step in the testing strategy. Two main approaches can be followed:

**Precise Sampling**

This approach allows for quickly creating test cases by selecting rows using their technical or business keys. Here is a fictitious example with a small sample. In practice, it’s recommended to include a larger number of values to cover more cases

``` yaml
Sales test:
  source:
    connection: target_connection
    type: mysql
    query: |
      select * 
      from fact_sales
      where sale_id in ('E9EYKDIW6C', 'R5QUFFYXF0', 'FF0YIVG63B', 'DZWG6FJQWO', '7Y8G0EQ3JD', '9PF09Z3A6O', 'P0TIAOVD7D', 'CPY0E0N72T', 'EKCBGJNR25', '06BF63C6C8')
      order by sale_id
  expected:
    connection: legacy_connection
    type: bigquery
    query: |
      select * 
      from fact_sales
      where sale_id in ('E9EYKDIW6C', 'R5QUFFYXF0', 'FF0YIVG63B', 'DZWG6FJQWO', '7Y8G0EQ3JD', '9PF09Z3A6O', 'P0TIAOVD7D', 'CPY0E0N72T', 'EKCBGJNR25', '06BF63C6C8')
      order by sale_id -- Always remember to add sorting for easier comparison.
```


**Batch Sampling:** This approach is also quick but requires a deeper functional understanding. It involves testing a broader set of data. Here’s an example:

``` yaml
Sales test:
  source:
    connection: target_connection
    type: mysql
    query: |
      select * 
      from fact_sales
      where country = 'france'
      order by sale_id
  expected:
    connection: legacy_connection
    type: bigquery
    query: |
      select * 
      from fact_sales
      where country = 'france'
      order by sale_id -- Always remember to add sorting for easier comparison.
```


##### Conclusion

Gone are the days of endless Excel comparisons! Using Ploosh allowed us to focus on sampling and migration while testing and retesting our feeds, saving us time and energy. I will be back soon with other concrete examples of use cases and different approaches for setting up tests with Ploosh. In the meantime, you can check out the source code and wiki on [GitHub](https://github.com/CSharplie/ploosh/).

* * *

See original post on LinkedIn (in french) : [Ploosh : comment faciliter ses tests de migration ?](https://www.linkedin.com/pulse/ploosh-un-framework-pour-automatiser-les-tests-en-data-collier-uiske/)

### ploosh: a framework to automatize tests in data project

Date: 2024-02-13 — Author: Charlie Collier

Source: https://ploosh.io/blog/ploosh-a-framework-to-automatize-tests-in-data-project

##### Introduction

In this article, I will present the issues related to testing in data projects and introduce one of my tools to address them. In a future article, I will return to the various approaches that can be applied as well as some use cases.

##### 1. The problem of automated testing in data

Testing tools for application development are not necessarily suited for data and BI projects. Indeed, data systems are often chains of complex workflows with multiple dependencies, making it difficult to test the entire process.

In the traditional development world, we are used to the following sequence: CI (build & tests) / deployment / execution. However, for the reasons mentioned earlier, this is not possible in a data project. Instead, we encounter the following sequence: CI (build only) / deployment / execution / tests.

![](/assets/img/blog/ploosh-a-framework-to-automatize-tests-in-data-project/schema-1.png)

##### 2. Why automate your tests?

*   **Reduce testing effort:** With the industrialization of tests, teams can focus on development or creating complex and high-value test cases.
*   **Reduce regression risks:** By continuously running tests, regressions can be detected quickly and fixed before they impact production.
*   **Increase test quality:** When a new bug is detected, new test cases can be added to the framework to control and prevent a recurrence of the issue, thereby enriching the test base.
*   **Improve project quality:** With fewer regression bugs and a more efficient team on development tasks, the product’s quality improves.

##### 3. Ploosh

###### What is Ploosh?

Ploosh is a framework that enables quick and simple implementation of tests within data projects. Based on YAML files, it is easily usable by technical teams, but also by business analysts or dedicated testing teams.

However, it’s important to remember that Ploosh is only a toolset, and the intelligence of the tests lies in the people who write them. The documentation is available on the Ploosh wiki.

###### How it works

Ploosh is based on a simple concept: a test consists of a value to be tested and a reference value. For a test to be valid, the value to be tested must match the reference value.

![](/assets/img/blog/ploosh-a-framework-to-automatize-tests-in-data-project/schema-2.png)

The framework offers three main features forming the test engine:

**Connectors:** They allow, based on the defined configuration, to query a data source and store the result in a homogeneous format (dataframe), which enables the compare engine to execute.

**The compare engine:** Its goal is to compare, for each test case, the data from the “source” connector configuration with the data from the “expected” connector configuration. It operates in three successive steps:

*   Row count comparison between the two datasets.
*   Structural equality check between the two datasets.
*   Row-by-row data comparison between the datasets to confirm if they match.

![](/assets/img/blog/ploosh-a-framework-to-automatize-tests-in-data-project/schema-3.png)

**Export:** This feature allows for leveraging the test results in reports (PowerBI, Tableau, etc.), available in different formats:

*   **JSON / CSV:** Results can be exported as CSV or JSON files, accompanied by an Excel file per test case showing any discrepancies.
*   **TRX:** Results can be exported in the T-Rex format (actually XML with subfolders) and can be used directly in the “Testing Plan” section of Azure DevOps.

![](/assets/img/blog/ploosh-a-framework-to-automatize-tests-in-data-project/schema-4.png)

##### 4. Exemple

Installing Ploosh is done via the “pip” command. You need to have Python 3.2 or higher installed on your machine or server.

Next, we create a `_**connections.yml**_` file to manage the various connections used during the tests. In this example, we create two connections:

*   **_`dwh`_** simulates a connection to a data warehouse.
*   **_`dmt`_** simulates a connection to a datamart. You’ll notice that for the passwords, we use variables. These will be passed via the command line that launches Ploosh.

``` yaml
dwh:
  type: mysql
  hostname: my_dwh.database.windows.net
  database: my_dwh
  username: my_user_name
  password: $var.my_dwh_password

dmt:
  type: mysql
  hostname: my_dmt.database.windows.net
  database: my_dmt
  username: my_user_name
  password: $var.my_dmt_password

```


After this, we create a folder containing our test cases. Inside, we create a file `test_cases/demo.yml` containing the test cases for this example.

In this case, we will create a test case, **“Test users data”**, where we verify that the data in the “users” table from the datamart is correctly aggregated by comparing it with the data from the data warehouse.

``` yaml
Test users data:
  options:
    sort:
      - gender
      - domain
  source:
    connection: dwh
    type: mysql
    query: | 
      select gender
          , right(email, length(email) - position("@" in email)) as domain
          , count(*) as count
        from users
        group by gender, domain
  expected:
    connection: dmt
    type: mysql
    query: | 
      select gender, domain, count
        from users

```


Finally, we execute Ploosh to run the tests:

```
ploosh --connections "connections.yml" --cases "test_cases" --export "JSON" --my_dwh_password "mypassword" --my_dmt_password "mypassword"

```


![](/assets/img/blog/ploosh-a-framework-to-automatize-tests-in-data-project/schema-5.png)

During execution, the status of the tests is displayed in real-time, and a summary will be shown once completed. At the end a file is generated with the details of the tests.

``` json
[
  {
    "name": "Test users data",
    "state": "passed",
    "source": {
      "start": "2024-02-11T15:40:55Z",
      "end": "2024-02-11T15:40:55Z",
      "duration": 0.003537366666666667
    },
    "expected": {
      "start": "2024-02-11T15:40:55Z",
      "end": "2024-02-11T15:40:56Z",
      "duration": 0.00014688333333333334
    },
    "compare": {
      "start": "2024-02-11T15:40:56Z",
      "end": "2024-02-11T15:40:56Z",
      "duration": 0.0010019833333333333
    }
  },
  {
    "name": "Test sales data",
    "state": "failed",
    "source": {
      "start": "2024-02-11T15:40:56Z",
      "end": "2024-02-11T15:40:56Z",
      "duration": 0.0018024166666666666
    },
    "expected": {
      "start": "2024-02-11T15:40:56Z",
      "end": "2024-02-11T15:40:56Z",
      "duration": 0.0000436666666666666
    },
    "compare": {
      "start": "2024-02-11T15:40:56Z",
      "end": "2024-02-11T15:40:56Z",
      "duration": 0.00035798333333333335
    },
    "error": {
      "type": "data",
      "message": "Some rows are not equals between source dataset and expected dataset"
    }
  }
]
```


##### 5. Conclusion

Ploosh est un framework flexible permettant de résoudre les problématiques de tests des projets de données. Il va être amené à être amélioré et à voir ses fonctionnalités augmentées.

Je reviendrai bientôt avec des exemples de cas d’usage, mais aussi avec différentes approches de la mise en place de tests à travers Ploosh.

Ploosh is a flexible framework that solves testing issues in data projects. It is expected to be improved and gain more features. I will come back soon with examples of use cases and different approaches for implementing tests through Ploosh. In the meantime, its source code and wiki are available on [GitHub](https://github.com/CSharplie/ploosh).

* * *

See original post on LinkedIn (in french) : [Ploosh : un framework pour automatiser les tests en data](https://www.linkedin.com/pulse/ploosh-un-framework-pour-automatiser-les-tests-en-data-collier-uiske/)

---

## Changelog

### Version 0.5.6 (2026-06-09)

Source: https://ploosh.io/changelog/0.5.6

### Version 0.5.6
**Release Date:** 2026-06-09

#### Features
##### Performance
* **Parallel Test Case Processing**: Test cases can now run concurrently through a new `--workers` command-line option, significantly reducing total execution time for large test suites. Log output has been reworked to remain readable under concurrent execution.

##### Connectors
* **Fabric KQL Connector — API Mode**: The Fabric KQL Spark connector now supports a `connection_mode` parameter (`native` or `api`), allowing KQL queries to be executed either through the native Spark path or via the API.

##### Experience
* **Richer Console Output**: Replaced `colorama` with `rich` to provide clearer, better-formatted, and more colorful console logging during test execution.

#### Bug Fixes
* **pyodbc Compatibility**: Updated `pyodbc` to 5.2.0 to fix installation and runtime issues on newer Python versions.
* **Dependency Constraints**: Added version constraints for `semantic-link-labs` and `sqlalchemy-bigquery` to prevent incompatible installs across supported Python versions.

#### Technical Enhancements
* **Python 3.11–3.13 Support**: The package and its dependencies are now validated against Python 3.11, 3.12, and 3.13.
* **Packaging Migration**: Consolidated `setup.py`, `setup-core.py`, and `setup-full.py` into a single modern `pyproject.toml`.
* **Code Quality**: Cleaned up imports across multiple connectors, removed obsolete Fabric sync scripts, and added Copilot contribution instructions.

### Version 0.4.11 (2026-04-16)

Source: https://ploosh.io/changelog/0.4.11

### Version 0.4.11
**Release Date:** 2026-04-16

#### Features
##### New Connectors
* **Spark Fabric Warehouse**: Added Spark support for reading Fabric Warehouse.
* **Spark Fabric Semantic Model**: Added Spark support for reading Fabric Semantic Models.

### Version 0.4.8 (2026-04-15)

Source: https://ploosh.io/changelog/0.4.8

### Version 0.4.8
**Release Date:** 2026-04-15

#### Features
##### New Connectors
* **Spark Dremio Connector**: Added Spark support for reading Dremio databases.

##### API Enhancements
* **Parameters**: Allow to pass parameters to the Python API to be processed into connection and case configuration files.

### Version 0.4.7 (2026-01-20)

Source: https://ploosh.io/changelog/0.4.7

### Version 0.4.7
**Release Date:** 2026-01-20

#### Features
##### New Connectors
* **Native Delta Connector**: Added native support for reading Delta Lake tables.
* **Microsoft Fabric KQL Connector**: Introduced a new Spark-based connector for executing KQL queries on Microsoft Fabric.

##### Connector Enhancements
* **CSV Schema Support**: Ability to define a strict schema for the CSV connector, supporting `int`, `float`, `string`, `bool`, and `datetime` types.

##### Compare Engine
* **Join Comparison Mode**: Added support for data comparison via joins using specific keys (`join_keys`) in Spark mode, providing an alternative to the `ORDER` mode.
* **Numeric Tolerance**: Improved handling of numeric comparisons in Spark by applying a tolerance threshold.

##### Exporters
* **Executed Action Tracking**: Exporters now include the specific action executed (e.g., file path or KQL query) for each test case.
* **Execution ID**: Added `execution_id` to the export process to improve the traceability of test runs.

#### Bug Fixes
* **Column Name Sanitization**: Column names are now automatically trimmed before comparison to prevent failures caused by leading or trailing whitespace.

#### Technical Enhancements
* **Dry Run Workflow**: Implemented a "dry run" step in GitHub Actions to validate the package and its execution before publishing.
* **Linting & Quality**: Enhanced the codebase maintainability by improving Pylint scores and updating configurations.
* **Unit Testing**: Expanded unit test coverage, particularly for new connectors and comparison engine logic.

### Version 0.3.8 (2025-01-15)

Source: https://ploosh.io/changelog/0.3.8

### Version 0.3.8
**Release Date:** 2025-01-15

#### Features 
##### New Connectors
* Spark parquet connector
* JSON spark connector
* JSON natif connector

##### Load Engine
* Re-design and reform of load engine to align the features between spark and native
* Add cast configuration to spark load engine

##### Compare Engine
* Re-design and reform of compare engine to align the features between spark and native
* Add case insensitive parameter to spark and native compare engine
* Add trim parameter to spark and native compare engine
* Add tolerance parameter to spark and native compare engine
* Align spark and native Excel outputs

### Version 0.2.32 (2024-12-22)

Source: https://ploosh.io/changelog/0.2.32

### Version 0.2.32
**Release Date:** 2024-12-22

#### Features
##### New Connectors
* Spark parquet connector
* JSON spark connector
* JSON natif connector

##### Load Engine
* Re-design and reform of load engine to align the features between spark and native
* Add cast configuration to spark load engine

##### Compare Engine
* Re-design and reform of compare engine to align the features between spark and native
* Add case insensitive parameter to spark and native compare engine
* Add trim parameter to spark and native compare engine
* Add tolerance parameter to spark and native compare engine
* Align spark and native Excel outputs

### Version 0.2.31 (2024-12-21)

Source: https://ploosh.io/changelog/0.2.31

### Version 0.2.31
**Release Date:** 2024-12-21

#### Features
* CSV spark connector : Add multiple parameters
* Parquet native connector
* Analysis native Services connector
* Semantic native Model XMLA connector

### Version 0.2.29 (2024-12-19)

Source: https://ploosh.io/changelog/0.2.29

### Version 0.2.29
**Release Date:** 2024-12-19

#### Features
* CSV native connector : Add multiple parameters

### Version 0.2.27 (2024-12-17)

Source: https://ploosh.io/changelog/0.2.27

### Version 0.2.27
**Release Date:** 2024-12-17

#### Bugfix
* Make error on connector loading not blocking (by display warning)

### Version 0.2.26 (2024-12-16)

Source: https://ploosh.io/changelog/0.2.26

### Version 0.2.26
**Release Date:** 2024-12-16

#### Features
* ODBC native connector : Add connection string option

### Version 0.2.25 (2024-12-15)

Source: https://ploosh.io/changelog/0.2.25

### Version 0.2.25
**Release Date:** 2024-12-15

#### Bugfix
* Fix default value for "trim" option

### Version 0.2.24 (2024-12-14)

Source: https://ploosh.io/changelog/0.2.24

### Version 0.2.24
**Release Date:** 2024-12-14

#### Features
* Add allow_no_rows option to configuration
* Add trim option to configuration
* Excel native connector

#### Bugfix
* Make headers compare case insensitive

### Version 0.2.20 (2024-12-10)

Source: https://ploosh.io/changelog/0.2.20

### Version 0.2.20
**Release Date:** 2024-12-10

#### Features
* Add pass rate option to configuration

#### Bugfix
* Allow cast with case insensitive columns matching

### Version 0.2.18 (2024-12-08)

Source: https://ploosh.io/changelog/0.2.18

### Version 0.2.18
**Release Date:** 2024-12-08

#### Features
* Add cast option to configuration (native)

### Version 0.2.17 (2024-12-07)

Source: https://ploosh.io/changelog/0.2.17

### Version 0.2.17
**Release Date:** 2024-12-07

#### Features
* Add count of expected and source rows in export results

#### Bugfix
* Remap data type for dataframe when type of column is object
* Remove timezone from datetime (native)

### Version 0.2.16 (2024-12-06)

Source: https://ploosh.io/changelog/0.2.16

### Version 0.2.16
**Release Date:** 2024-12-06

#### Bugfix
* Fix encoding on ODBC native connector
* Fix headers comparison

### Version 0.2.14 (2024-12-04)

Source: https://ploosh.io/changelog/0.2.14

### Version 0.2.14
**Release Date:** 2024-12-04

#### Features
* Persist logs into files

#### Bugfix
* Fix test duration in output file

### Version 0.2.12 (2024-12-02)

Source: https://ploosh.io/changelog/0.2.12

### Version 0.2.12
**Release Date:** 2024-12-02

#### Features
* Big query native connector : Allow to use the current user as authentication

### Version 0.2.11 (2024-12-01)

Source: https://ploosh.io/changelog/0.2.11

### Version 0.2.11
**Release Date:** 2024-12-01

#### Features
* ODBC native connector

#### Bugfix
* Allow command line execution in conda environment

### Version 0.2.9 (2024-11-30)

Source: https://ploosh.io/changelog/0.2.9

### Version 0.2.9
**Release Date:** 2024-11-30

#### Features
* Add spark mode
* New SQL spark connector
* New CSV spark connector
* New Delta spark connector
* New Empty spark connector