Metadata-Version: 2.1
Name: pytest-dbt-postgres
Version: 0.2.0
Summary: Pytest tooling to unittest DBT & Postgres models
Author: albert.franzi
Requires-Python: >=3.10,<4.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: absl-py (>=2.1.0,<3.0.0)
Requires-Dist: dbt-core (>=1.8.4,<2.0.0)
Requires-Dist: dbt-postgres (>=1.8.2,<2.0.0)
Requires-Dist: numpy (>=2.1.0,<3.0.0)
Requires-Dist: pandas (>=2.2.2,<3.0.0)
Requires-Dist: psycopg2-binary (>=2.9.9,<3.0.0)
Requires-Dist: pydantic (>=2.7.3,<3.0.0)
Requires-Dist: pytest (>=8.3.2,<9.0.0)
Requires-Dist: pytest-env (>=1.1.3,<2.0.0)
Requires-Dist: pytest-postgresql (>=6.0.1,<7.0.0)
Description-Content-Type: text/markdown

# PyTest DBT Postgres

This project aims to provide an easy way to validate a DBT project by enabling the
unit-testing capability.

## Dependencies

This project is based on the [pytest-postgresql](https://pypi.org/project/pytest-postgresql/) library, so we can
validate our DBT postgres project without relying on an existing database.

## Usage

The initial idea resides on pytest-postgres providing us an empty and isolated postgres where we can
validate different use-cases with our DBT code.

> If your project uses external sources, you will need to make sure they are properly defined in the yml files, since
> the project generates the DDLs from there.

```python
import unittest

from pytest_dbt_postgres import auto_inject_fixtures
from pytest_dbt_postgres.dbt_executor import DbtExecutor
from pytest_dbt_postgres.dbt_validator import DbtValidator
from pytest_dbt_postgres.postgres.connector import PostgresConnector
from pytest_dbt_postgres.postgres.sql_methods import export_credentials_to_env, get_credentials

dbt_project_dir = '...'
resources_folder = '...'


@auto_inject_fixtures("postgresql")
class TestDbtValidator(unittest.TestCase):
    def setUp(self) -> None:
        credentials = get_credentials(self.postgresql)
        self.connector = PostgresConnector(credentials)

        # SETUP
        export_credentials_to_env(credentials)
        executor = DbtExecutor(dbt_project_dir=dbt_project_dir, profiles_dir=resources_folder)
        self.validator = DbtValidator(connector=self.connector, executor=executor, resources_folder=resources_folder)

    def test_orders(self) -> None:
        # GIVEN
        sources_to_load = [
            ("external.customers", "csv/input/customers.csv"),
            ("external.orders", "csv/input/orders.csv"),
        ]
        selector = "orders"
        outputs_to_validate = [
            ("dbt_unittest.orders", "csv/output/orders.csv"),
        ]
        # THEN
        self.validator.validate(sources_to_load, selector, outputs_to_validate)
```

