Repository: getindata/dbt-airflow-factory (language: python)

Issue:
Release 0.2.0
Bump version and CHANGELOG for next release.


Discussion / hints:

Fix patch:
diff --git a/.codeclimate.yml b/.codeclimate.yml
new file mode 100644
index 0000000..d84f5fc
--- /dev/null
+++ b/.codeclimate.yml
@@ -0,0 +1,3 @@
+plugins:
+  pylint:
+    enabled: true
\ No newline at end of file
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
new file mode 100644
index 0000000..c44dd34
--- /dev/null
+++ b/.github/CODEOWNERS
@@ -0,0 +1,2 @@
+# Define global code owners
+* @p-pekala
\ No newline at end of file
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
new file mode 100644
index 0000000..4ab52dd
--- /dev/null
+++ b/.github/pull_request_template.md
@@ -0,0 +1,8 @@
+`<description>`
+
+Resolves `<issue nr here>`
+
+---
+Keep in mind:
+- [ ] Documentation updates
+- [ ] [Changelog](CHANGELOG.md) updates 
\ No newline at end of file
diff --git a/.github/workflows/prepare-release.yml b/.github/workflows/prepare-release.yml
new file mode 100644
index 0000000..3062e9b
--- /dev/null
+++ b/.github/workflows/prepare-release.yml
@@ -0,0 +1,58 @@
+name: Prepare release
+
+on:
+  workflow_dispatch:
+    inputs:
+      version_part:
+        description: The part of the version to update (patch, minor or major)
+        required: true
+        default: 'minor'
+
+jobs:
+  prepare-release:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.9]
+    env:
+      PYTHON_PACKAGE: dbt_airflow_manifest_parser
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v1
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Validate inputs
+        run: |
+          echo "INPUT_VERSION_PART: ${{ github.event.inputs.version_part }}"
+          python -c "if '${{ github.event.inputs.version_part }}' not in ['patch', 'minor', 'major']:   raise ValueError(\"'${{ github.event.inputs.version_part }}' must be one of ['patch', 'minor', 'major'])\")"
+      - name: Bump the version number  # bump2version is a maintained fork of original bumpversion
+        id: bump_version
+        run: |
+          pip install bump2version
+          bump2version ${{ github.event.inputs.version_part }}
+          echo "::set-output name=package_version::$(cat $PYTHON_PACKAGE/__init__.py | grep -Po  '\d+\.\d+\.\d+')"
+      - name: Update the CHANGELOG according to 'Keep a Changelog' guidelines
+        uses: thomaseizinger/keep-a-changelog-new-release@v1
+        with:
+          version: ${{ steps.bump_version.outputs.package_version }}
+      - name: Create a new release branch
+        run: |
+          git config user.name github-actions
+          git config user.email github-actions@github.com
+          git checkout -b release-${{ steps.bump_version.outputs.package_version }}
+          git push -u origin release-${{ steps.bump_version.outputs.package_version }}
+      - name: Open a PR to merge the release to master
+        id: open_pr
+        uses: vsoch/pull-request-action@1.0.12
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          PULL_REQUEST_BRANCH: main
+          PULL_REQUEST_FROM_BRANCH: release-${{ steps.bump_version.outputs.package_version }}
+          PULL_REQUEST_TITLE: "Release ${{ steps.bump_version.outputs.package_version }}"
+          PULL_REQUEST_BODY: "Bump version and CHANGELOG for next release."
+          PULL_REQUEST_ASSIGNEES: " p-pekala"
+      - name: Commit the changes
+        run: |
+          git commit -am "FIX #${{ steps.open_pr.outputs.pull_request_number }}  - Bump version and CHANGELOG for release ${{ steps.bump_version.outputs.package_version }}"
+          git push
\ No newline at end of file
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
new file mode 100644
index 0000000..7375da3
--- /dev/null
+++ b/.github/workflows/publish.yml
@@ -0,0 +1,67 @@
+name: Publish
+
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.7, 3.9]
+    env:
+      PYTHON_PACKAGE: dbt_airflow_manifest_parser
+    steps:
+      - name: Checkout the repo
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 0  # necessary to enable merging, all the history is needed
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v1
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Build package dist from source # A better way will be : https://packaging.python.org/guides/publishing-package-distribution-releases-using-github-actions-ci-cd-workflows/ but pep 517 is still marked as experimental
+        run: |
+          python setup.py sdist
+      - name:  Merge back to develop  # we have to set the config first on a fresh machine
+        run: |
+          git config user.name github-actions
+          git config user.email github-actions@github.com
+          git checkout -b develop --track origin/develop
+          git merge master
+          git push
+      - name: Set dynamically package version as output variable # see https://github.com/actions/create-release/issues/39
+        # see https://docs.github.com/en/free-pro-team@latest/actions/reference/workflow-commands-for-github-actions#setting-an-environment-variable
+        id: set_package_version
+        run: |
+          echo "::set-output name=PACKAGE_VERSION::$(cat $PYTHON_PACKAGE/__init__.py | grep -Po  '\d+\.\d+\.\d+')"
+      - name: Create temporary file with the body content for the release
+        run: |
+          grep -Poz "## \[${{steps.set_package_version.outputs.PACKAGE_VERSION}}] - \d{4}-\d{2}-\d{2}[\S\s]+?(?=## \[\d+\.\d+\.\d+\]|\[.+\]:)" CHANGELOG.md > release_body.md
+      - name: Create Release # https://github.com/actions/create-release
+        id: create_release
+        uses: actions/create-release@v1.1.4
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} # This token is provided by Actions, you do not need to create your own token
+        with:
+          tag_name: ${{ steps.set_package_version.outputs.PACKAGE_VERSION }}
+          release_name: Release ${{ steps.set_package_version.outputs.PACKAGE_VERSION }}
+          body_path: ./release_body.md
+          draft: false
+          prerelease: false
+      - name: Rollback Release in case of run failure
+        if: failure() && steps.create_release.outputs.id != ''
+        uses: author/action-rollback@stable
+        with:
+          # Using a known release ID
+          release_id: ${{ steps.create_release.outputs.id }}
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      - name: Publish distribution to PyPI  # official action from python maintainers
+        uses: pypa/gh-action-pypi-publish@master
+        with:
+          user: __token__
+          password: ${{ secrets.PYPI_PASSWORD }}
+          verbose: true  # trace if the upload fails
\ No newline at end of file
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
new file mode 100644
index 0000000..e6b8f31
--- /dev/null
+++ b/.github/workflows/python-package.yml
@@ -0,0 +1,37 @@
+name: Tests
+
+on:
+  push:
+    branches:
+      - master
+      - develop
+  pull_request:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Setup python
+        uses: actions/setup-python@v2.2.1
+        with:
+          python-version: 3.7
+
+      - name: Setup virtualenv
+        run: |
+          python -V
+          python -m pip install virtualenv pipdeptree
+          virtualenv venv
+          source venv/bin/activate
+          pip install --upgrade pip
+
+      - name: Check pre-commit status
+        run: |
+          pip install .[tests]
+          pip freeze
+          pipdeptree
+          pre-commit run --all-files
+      - name: Test with tox
+        run: |
+          tox -e py37
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..915978d
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,15 @@
+# Default ignored files
+target/
+build/
+dist/
+dbt_modules/
+logs/
+dbt_airflow_manifest_parser.egg-info
+.idea
+
+.user.yml
+
+.coverage
+coverage.xml
+.tox
+
diff --git a/.isort.cfg b/.isort.cfg
new file mode 100644
index 0000000..7fcc210
--- /dev/null
+++ b/.isort.cfg
@@ -0,0 +1,9 @@
+[tool:isort]
+include_trailing_comma = True
+multi_line_output = 3
+line_length = 120
+known_first_party = dbt_airflow_manifest_parser
+default_section = THIRDPARTY
+
+[settings]
+known_third_party = airflow,setuptools
\ No newline at end of file
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
new file mode 100644
index 0000000..29163bb
--- /dev/null
+++ b/.pre-commit-config.yaml
@@ -0,0 +1,18 @@
+repos:
+  - repo: https://github.com/asottile/seed-isort-config
+    rev: v2.2.0
+    hooks:
+      - id: seed-isort-config
+  - repo: https://github.com/pycqa/isort
+    rev: 5.5.4
+    hooks:
+      - id: isort
+        args: ["--profile", "black"]
+  - repo: https://github.com/psf/black
+    rev: 21.9b0
+    hooks:
+      - id: black
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v2.3.0
+    hooks:
+      - id: flake8
\ No newline at end of file
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..fbbbe33
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,9 @@
+## [Unreleased]
+
+## [0.2.0] - 2021-10-25
+
+-   Initial implementation of `dbt_airflow_manifest_parser` library. 
+
+[Unreleased]: https://github.com/getindata/dbt-airflow-manifest-parser/compare/0.2.0...HEAD
+
+[0.2.0]: https://github.com/getindata/dbt-airflow-manifest-parser/compare/6395f7ea175caa3bd1aca361e9d2f7fb7f7a7820...0.2.0
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..14ba945
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,24 @@
+## PR Guidelines
+1. Fork branch from `develop`.
+1. Ensure to provide unit tests for new functionality.
+1. Install dev requirements: `pip install -r requirements-dev.txt` and setup a hook: `pre-commit install`
+1. Update documentation accordingly.
+1. Update [changelog](CHANGELOG.md) according to ["Keep a changelog"](https://keepachangelog.com/en/1.0.0/) guidelines.
+1. Squash changes with a single commit as much as possible and ensure verbose PR name.
+1. Open a PR against `develop`
+
+*We reserve the right to take over and modify or abandon PRs that do not match the workflow or are abandoned.*
+
+## Release workflow
+
+1. Create the release candidate:
+    - Go to the [Prepare release](https://github.com/getindata/kedro-airflow-k8s/actions?query=workflow%3A%22Prepare+release%22) action.
+    - Click "Run workflow"
+    - Enter the part of the version to bump (one of `<major>.<minor>.<patch>`). Minor (x.**x**.x) is a default.
+2. If the workflow has run sucessfully:
+    - Go to the newly openened PR named `Release candidate <version>`
+    - Check that changelog and version have been properly updated. If not pull the branch and apply manual changes if necessary.
+    - Merge the PR to master
+3. Checkout the [Publish](https://github.com/getindata/kedro-airflow-k8s/actions?query=workflow%3APublish) workflow to see if:
+    - The package has been uploaded on PyPI successfully
+    - The changes have been merged back to develop
\ No newline at end of file
diff --git a/README.md b/README.md
index a76b961..54a247a 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,5 @@
-# dbt-airflow-manifest-parser
+# DBT Airflow Manifest Parser
+
+## About
+
 Library to convert DBT manifest metadata to Airflow tasks
diff --git a/dbt_airflow_manifest_parser/__init__.py b/dbt_airflow_manifest_parser/__init__.py
new file mode 100644
index 0000000..181e9f0
--- /dev/null
+++ b/dbt_airflow_manifest_parser/__init__.py
@@ -0,0 +1,1 @@
+version = "0.2.0"
diff --git a/dbt_airflow_manifest_parser/builder.py b/dbt_airflow_manifest_parser/builder.py
new file mode 100644
index 0000000..0a3d50a
--- /dev/null
+++ b/dbt_airflow_manifest_parser/builder.py
@@ -0,0 +1,66 @@
+import json
+import logging
+
+from dbt_airflow_manifest_parser.tasks import ModelExecutionTask, ModelExecutionTasks
+
+
+class DbtAirflowTasksBuilder:
+    def __init__(self, operator_builder):
+        self.operator_builder = operator_builder
+
+    def _load_dbt_manifest(self, manifest_path):
+        with open(manifest_path, "r") as f:
+            manifest_content = json.load(f)
+            logging.debug("Manifest content: " + str(manifest_content))
+            return manifest_content
+
+    def _make_dbt_test_task(self, model_name):
+        return self.operator_builder.create(model_name + "_test", model_name, "test")
+
+    def _make_dbt_run_task(self, model_name):
+        return self.operator_builder.create(model_name + "_run", model_name, "run")
+
+    @staticmethod
+    def _is_model_run_task(node_name):
+        return node_name.split(".")[0] == "model"
+
+    def _create_tasks_for_each_model(self, manifest):
+        tasks = {}
+        for node_name in manifest["nodes"].keys():
+            if self._is_model_run_task(node_name):
+                logging.info("Creating tasks for: " + node_name)
+                model_name = node_name.split(".")[-1]
+                run_task = self._make_dbt_run_task(model_name)
+                test_task = self._make_dbt_test_task(model_name)
+                run_task >> test_task
+                tasks[node_name] = ModelExecutionTask(run_task, test_task)
+        return tasks
+
+    def _create_tasks_dependencies(self, manifest, tasks):
+        starting_tasks = list(tasks.keys())
+        ending_tasks = list(tasks.keys())
+        for node_name in tasks.keys():
+            for upstream_node in manifest["nodes"][node_name]["depends_on"]["nodes"]:
+                if self._is_model_run_task(upstream_node):
+                    (
+                        tasks[upstream_node].test_airflow_task
+                        >> tasks[node_name].run_airflow_task
+                    )
+                    if node_name in starting_tasks:
+                        starting_tasks.remove(node_name)
+                    if upstream_node in ending_tasks:
+                        ending_tasks.remove(upstream_node)
+        return ModelExecutionTasks(tasks, starting_tasks, ending_tasks)
+
+    def _make_dbt_tasks(self, manifest_path):
+        manifest = self._load_dbt_manifest(manifest_path)
+        tasks = self._create_tasks_for_each_model(manifest)
+        tasks_with_context = self._create_tasks_dependencies(manifest, tasks)
+        logging.info(f"Created {str(tasks_with_context.length())} tasks groups")
+        return tasks_with_context
+
+    def parse_manifest_into_tasks(self, manifest_path):
+        return self._make_dbt_tasks(manifest_path)
+
+    def create_seed_task(self):
+        return self.operator_builder.create("dbt_seed", None, "seed")
diff --git a/dbt_airflow_manifest_parser/operator.py b/dbt_airflow_manifest_parser/operator.py
new file mode 100644
index 0000000..3894e07
--- /dev/null
+++ b/dbt_airflow_manifest_parser/operator.py
@@ -0,0 +1,39 @@
+import abc
+
+from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
+
+
+class DbtRunOperatorBuilder(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def create(self, name, model, command):
+        raise NotImplementedError
+
+
+class KubernetesPodOperatorBuilder(DbtRunOperatorBuilder):
+    def __init__(self, dbt_execution_env_parameters, kubernetes_execution_parameters):
+        self.dbt_execution_env_parameters = dbt_execution_env_parameters
+        self.kubernetes_execution_parameters = kubernetes_execution_parameters
+
+    def create(self, name, model, command):
+        return KubernetesPodOperator(
+            namespace=self.kubernetes_execution_parameters.namespace,
+            image=self.kubernetes_execution_parameters.image,
+            cmds=["bash", "-c"],
+            node_selectors=self.kubernetes_execution_parameters.node_selectors,
+            tolerations=self.kubernetes_execution_parameters.tolerations,
+            arguments=[
+                f"set -e; "
+                f"dbt --no-write-json {command} "
+                f"--target {self.dbt_execution_env_parameters.target} "
+                f"--models {model} "
+                f"--project-dir {self.dbt_execution_env_parameters.project_dir_path} "
+                f"--profiles-dir {self.dbt_execution_env_parameters.profile_dir_path}"
+            ],
+            labels=self.kubernetes_execution_parameters.labels,
+            name=name,
+            task_id=name,
+            resources=self.kubernetes_execution_parameters.resources,
+            secrets=self.kubernetes_execution_parameters.secrets,
+            is_delete_operator_pod=self.kubernetes_execution_parameters.is_delete_operator_pod,
+            hostnetwork=False,
+        )
diff --git a/dbt_airflow_manifest_parser/parameters.py b/dbt_airflow_manifest_parser/parameters.py
new file mode 100644
index 0000000..5040a78
--- /dev/null
+++ b/dbt_airflow_manifest_parser/parameters.py
@@ -0,0 +1,27 @@
+class DbtExecutionEnvironmentParameters:
+    def __init__(self, target, project_dir_path, profile_dir_path):
+        self.target = target
+        self.project_dir_path = project_dir_path
+        self.profile_dir_path = profile_dir_path
+
+
+class KubernetesExecutionParameters:
+    def __init__(
+        self,
+        namespace,
+        image,
+        node_selectors=None,
+        tolerations=None,
+        labels=None,
+        resources=None,
+        secrets=None,
+        is_delete_operator_pod=True,
+    ):
+        self.namespace = namespace
+        self.image = image
+        self.node_selectors = node_selectors
+        self.tolerations = tolerations
+        self.labels = labels
+        self.resources = resources
+        self.secrets = secrets
+        self.is_delete_operator_pod = is_delete_operator_pod
diff --git a/dbt_airflow_manifest_parser/tasks.py b/dbt_airflow_manifest_parser/tasks.py
new file mode 100644
index 0000000..adb691a
--- /dev/null
+++ b/dbt_airflow_manifest_parser/tasks.py
@@ -0,0 +1,33 @@
+class ModelExecutionTasks:
+    def __init__(self, tasks, starting_task_names, ending_task_names):
+        self._tasks = tasks
+        self._starting_task_names = starting_task_names
+        self._ending_task_names = ending_task_names
+
+    def get_task(self, node_name):
+        return self._tasks[node_name]
+
+    def length(self):
+        return len(self._tasks)
+
+    def get_starting_tasks(self):
+        return self._extract_by_keys(self._starting_task_names)
+
+    def get_ending_tasks(self):
+        return self._extract_by_keys(self._ending_task_names)
+
+    def _extract_by_keys(self, keys):
+        tasks = []
+        for key in keys:
+            tasks.append(self._tasks[key])
+        return tasks
+
+
+class ModelExecutionTask:
+    def __init__(self, run_airflow_task, test_airflow_task):
+        self.run_airflow_task = run_airflow_task
+        self.test_airflow_task = test_airflow_task
+
+    # todo only for airflow 2.x
+    def to_task_group(self):
+        return None
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..e69de29
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..71c17a9
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,6 @@
+[bumpversion]
+current_version = 0.2.0
+
+[bumpversion:file:setup.py]
+
+[bumpversion:file:dbt_airflow_manifest_parser/__init__.py]
diff --git a/setup.py b/setup.py
new file mode 100644
index 0000000..fa1d3a6
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,50 @@
+"""kedro_airflow_k8s module."""
+
+from setuptools import find_packages, setup
+
+with open("README.md") as f:
+    README = f.read()
+
+# Runtime Requirements.
+INSTALL_REQUIRES = []
+
+# Dev Requirements
+EXTRA_REQUIRE = {
+    "tests": [
+        "pytest>=6.2.2, <7.0.0",
+        "pytest-cov>=2.8.0, <3.0.0",
+        "tox==3.21.1",
+        "pre-commit==2.9.3",
+        "pandas==1.2.5",
+        "apache-airflow[kubernetes]==1.10.14",
+    ],
+    "docs": [
+        "sphinx==3.4.2",
+        "recommonmark==0.7.1",
+        "sphinx_rtd_theme==0.5.2",
+    ],
+}
+
+setup(
+    name="dbt-airflow-manifest-parser",
+    version="0.2.0",
+    description="Library to convert DBT manifest metadata to Airflow tasks",
+    long_description=README,
+    long_description_content_type="text/markdown",
+    license="Apache Software License (Apache 2.0)",
+    python_requires=">=3",
+    classifiers=[
+        "Development Status :: 3 - Alpha",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+    ],
+    keywords="dbt airflow manifest parser python",
+    author=u"Piotr Pekala",
+    author_email="piotr.pekala@getindata.com",
+    url="https://github.com/getindata/dbt-airflow-manifest-parser/",
+    packages=find_packages(exclude=["ez_setup", "examples", "tests", "docs"]),
+    include_package_data=True,
+    zip_safe=False,
+    install_requires=INSTALL_REQUIRES,
+    extras_require=EXTRA_REQUIRE,
+)
diff --git a/tox.ini b/tox.ini
new file mode 100644
index 0000000..bdfa482
--- /dev/null
+++ b/tox.ini
@@ -0,0 +1,49 @@
+[tox]
+envlist = py37,py39
+
+[testenv]
+extras =
+    tests
+commands=
+    python -m pytest --cov dbt_airflow_manifest_parser --cov-report xml --cov-report term-missing --ignore=venv
+
+# Lint
+[flake8]
+exclude = .git,__pycache__,docs/source/conf.py,old,build,dist
+max-line-length = 120
+
+[mypy]
+no_strict_optional = True
+ignore_missing_imports = True
+
+# Autoformatter
+[testenv:black]
+basepython = python3
+skip_install = true
+deps =
+    black
+commands =
+    black
+
+# Release tooling
+[testenv:build]
+basepython = python3
+skip_install = true
+deps =
+    wheel
+    setuptools
+commands =
+    python setup.py sdist
+
+[testenv:release]
+basepython = python3
+skip_install = true
+setenv =
+    TWINE_USERNAME = {env:TWINE_USERNAME}
+    TWINE_PASSWORD = {env:TWINE_PASSWORD}
+deps =
+    {[testenv:build]deps}
+    twine >= 1.5.0
+commands =
+    {[testenv:build]commands}
+    twine upload --skip-existing dist/*
\ No newline at end of file


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_config_propagation.py b/tests/test_config_propagation.py
new file mode 100644
index 0000000..acb7e70
--- /dev/null
+++ b/tests/test_config_propagation.py
@@ -0,0 +1,38 @@
+from airflow.contrib.kubernetes.secret import Secret
+
+from .utils import TEST_DAG, manifest_file_with_models, task_builder
+
+
+def test_configuration():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models({"model.dbt_test.dim_users": []})
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    run_task = tasks.get_task("model.dbt_test.dim_users").run_airflow_task
+    assert run_task.namespace == "apache-airflow"
+    assert run_task.image == "dbt-platform-poc:123"
+    assert run_task.node_selectors == {"group": "data-processing"}
+    assert run_task.tolerations == [
+        {"key": "group", "value": "data-processing", "effect": "NoExecute"},
+        {
+            "key": "group",
+            "operator": "Equal",
+            "value": "data-processing",
+            "effect": "NoSchedule",
+        },
+    ]
+
+    assert run_task.labels == {"runner": "airflow"}
+    assert run_task.secrets == [
+        Secret("env", None, "snowflake-access-user-key", None),
+        Secret("volume", "/var", "snowflake-access-user-key", None),
+    ]
+    assert run_task.is_delete_operator_pod
+    assert "--project-dir /dbt" in run_task.arguments[0]
+    assert "--profiles-dir /root/.dbt" in run_task.arguments[0]
+    assert "--target dev" in run_task.arguments[0]
diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py
new file mode 100644
index 0000000..64af9c4
--- /dev/null
+++ b/tests/test_dependencies.py
@@ -0,0 +1,95 @@
+from .utils import TEST_DAG, manifest_file_with_models, task_builder
+
+
+def test_run_test_dependency():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models({"model.dbt_test.model1": []})
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    assert (
+        "model1_test"
+        in tasks.get_task("model.dbt_test.model1").run_airflow_task.downstream_task_ids
+    )
+    assert (
+        "model1_run"
+        in tasks.get_task("model.dbt_test.model1").test_airflow_task.upstream_task_ids
+    )
+
+
+def test_dependency():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models(
+        {
+            "model.dbt_test.model1": [],
+            "model.dbt_test.model2": ["model.dbt_test.model1"],
+        }
+    )
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    assert tasks.length() == 2
+    assert (
+        "model1_test"
+        in tasks.get_task("model.dbt_test.model2").run_airflow_task.upstream_task_ids
+    )
+    assert (
+        "model2_run"
+        in tasks.get_task("model.dbt_test.model1").test_airflow_task.downstream_task_ids
+    )
+
+
+def test_more_complex_dependencies():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models(
+        {
+            "model.dbt_test.model1": [],
+            "model.dbt_test.model2": ["model.dbt_test.model1"],
+            "model.dbt_test.model3": ["model.dbt_test.model1", "model.dbt_test.model2"],
+            "model.dbt_test.model4": ["model.dbt_test.model3"],
+        }
+    )
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    assert tasks.length() == 4
+    assert (
+        "model1_test"
+        in tasks.get_task("model.dbt_test.model2").run_airflow_task.upstream_task_ids
+    )
+    assert (
+        "model1_test"
+        in tasks.get_task("model.dbt_test.model3").run_airflow_task.upstream_task_ids
+    )
+    assert (
+        "model2_run"
+        in tasks.get_task("model.dbt_test.model1").test_airflow_task.downstream_task_ids
+    )
+    assert (
+        "model3_run"
+        in tasks.get_task("model.dbt_test.model1").test_airflow_task.downstream_task_ids
+    )
+    assert (
+        "model1_test"
+        in tasks.get_task("model.dbt_test.model3").run_airflow_task.upstream_task_ids
+    )
+    assert (
+        "model2_test"
+        in tasks.get_task("model.dbt_test.model3").run_airflow_task.upstream_task_ids
+    )
+    assert (
+        "model4_run"
+        in tasks.get_task("model.dbt_test.model3").test_airflow_task.downstream_task_ids
+    )
diff --git a/tests/test_eadges.py b/tests/test_eadges.py
new file mode 100644
index 0000000..9803230
--- /dev/null
+++ b/tests/test_eadges.py
@@ -0,0 +1,52 @@
+from .utils import TEST_DAG, manifest_file_with_models, task_builder
+
+
+def test_starting_tasks():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models(
+        {
+            "model.dbt_test.model1": [],
+            "model.dbt_test.model2": [],
+            "model.dbt_test.model3": ["model.dbt_test.model1", "model.dbt_test.model2"],
+            "model.dbt_test.model4": ["model.dbt_test.model3"],
+            "model.dbt_test.model5": [],
+        }
+    )
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    starting_tasks_names = [
+        task.run_airflow_task.task_id for task in tasks.get_starting_tasks()
+    ]
+    assert "model1_run" in starting_tasks_names
+    assert "model2_run" in starting_tasks_names
+    assert "model5_run" in starting_tasks_names
+
+
+def test_ending_tasks():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models(
+        {
+            "model.dbt_test.model1": [],
+            "model.dbt_test.model2": [],
+            "model.dbt_test.model3": ["model.dbt_test.model1", "model.dbt_test.model2"],
+            "model.dbt_test.model4": ["model.dbt_test.model3"],
+            "model.dbt_test.model5": [],
+        }
+    )
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    ending_tasks_names = [
+        task.test_airflow_task.task_id for task in tasks.get_ending_tasks()
+    ]
+    assert "model4_test" in ending_tasks_names
+    assert "model5_test" in ending_tasks_names
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
new file mode 100644
index 0000000..5ffc84d
--- /dev/null
+++ b/tests/test_tasks.py
@@ -0,0 +1,53 @@
+from .utils import TEST_DAG, manifest_file_with_models, task_builder
+
+
+def test_get_dag():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models({"model.dbt_test.dim_users": []})
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    assert tasks.length() == 1
+    assert tasks.get_task("model.dbt_test.dim_users") is not None
+    assert tasks.get_task("model.dbt_test.dim_users").run_airflow_task is not None
+    assert tasks.get_task("model.dbt_test.dim_users").test_airflow_task is not None
+
+
+def test_run_task():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models({"model.dbt_test.dim_users": []})
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    run_task = tasks.get_task("model.dbt_test.dim_users").run_airflow_task
+    assert run_task.cmds == ["bash", "-c"]
+    assert "set -e; dbt --no-write-json run " in run_task.arguments[0]
+    assert "--models dim_users" in run_task.arguments[0]
+    assert run_task.name == "dim-users-run"
+    assert run_task.task_id == "dim_users_run"
+
+
+def test_test_task():
+    # given
+    builder = task_builder()
+    manifest_path = manifest_file_with_models({"model.dbt_test.dim_users": []})
+
+    # when
+    with TEST_DAG:
+        tasks = builder.parse_manifest_into_tasks(manifest_path)
+
+    # then
+    test_task = tasks.get_task("model.dbt_test.dim_users").test_airflow_task
+    assert test_task.cmds == ["bash", "-c"]
+    assert "set -e; dbt --no-write-json test " in test_task.arguments[0]
+    assert "--models dim_users" in test_task.arguments[0]
+    assert test_task.name == "dim-users-test"
+    assert test_task.task_id == "dim_users_test"
diff --git a/tests/utils.py b/tests/utils.py
new file mode 100644
index 0000000..06d8c69
--- /dev/null
+++ b/tests/utils.py
@@ -0,0 +1,66 @@
+import json
+import tempfile
+from datetime import datetime
+
+from airflow import DAG
+from airflow.contrib.kubernetes.secret import Secret
+
+from dbt_airflow_manifest_parser.builder import DbtAirflowTasksBuilder
+from dbt_airflow_manifest_parser.operator import KubernetesPodOperatorBuilder
+from dbt_airflow_manifest_parser.parameters import (
+    DbtExecutionEnvironmentParameters,
+    KubernetesExecutionParameters,
+)
+
+
+def manifest_file_with_models(nodes_with_dependencies):
+    content_nodes = {}
+    for node_name in nodes_with_dependencies.keys():
+        content_nodes[node_name] = {
+            "depends_on": {"nodes": nodes_with_dependencies[node_name]}
+        }
+    content = {"nodes": content_nodes}
+    with tempfile.NamedTemporaryFile(delete=False) as tmp:
+        tmp.write(str.encode(json.dumps(content)))
+        return tmp.name
+
+
+def execution_environment_parameters():
+    return DbtExecutionEnvironmentParameters(
+        target="dev", project_dir_path="/dbt", profile_dir_path="/root/.dbt"
+    )
+
+
+def kubernetes_parameters():
+    return KubernetesExecutionParameters(
+        namespace="apache-airflow",
+        image="dbt-platform-poc:123",
+        node_selectors={"group": "data-processing"},
+        tolerations=[
+            {"key": "group", "value": "data-processing", "effect": "NoExecute"},
+            {
+                "key": "group",
+                "operator": "Equal",
+                "value": "data-processing",
+                "effect": "NoSchedule",
+            },
+        ],
+        labels={"runner": "airflow"},
+        resources={"limit_memory": "1024M", "limit_cpu": "1"},
+        secrets=[
+            Secret("env", None, "snowflake-access-user-key", None),
+            Secret("volume", "/var", "snowflake-access-user-key", None),
+        ],
+        is_delete_operator_pod=True,
+    )
+
+
+def task_builder():
+    return DbtAirflowTasksBuilder(
+        KubernetesPodOperatorBuilder(
+            execution_environment_parameters(), kubernetes_parameters()
+        )
+    )
+
+
+TEST_DAG = DAG("test", default_args={"start_date": datetime(2021, 10, 13)})