test with airflow

Author: weaverba137

.github/workflows/CI.yml

@@ -16,9 +16,8 @@ jobs:
     #         fail-fast: true
     #         matrix:
     #             os: [ubuntu-latest]
-    #             # numpy-version: ['<1.19', '<1.20', '>=1.20']
     #             python-version: ['3.9', '3.10', '3.11', '3.12']
-    #             astropy-version: ['<6.0', '<6.1', '<7.0']
+    #             airflow-version: ['<2.10', '<2.11']
 
     #     steps:
     #         - name: Checkout code
@@ -32,8 +31,8 @@ jobs:
     #         - name: Install Python dependencies
     #           run: |
     #             python -m pip install --upgrade pip setuptools wheel
-    #             python -m pip install pytest pyyaml jinja2 numpy\<2.0
-    #             python -m pip install "astropy${{ matrix.astropy-version }}"
+    #             python -m pip install pytest
+    #             python -m pip install "apache-airflow${{ matrix.airflow-version }}"
     #         - name: Run the test
     #           run: pytest
 
@@ -45,6 +44,7 @@ jobs:
             matrix:
                 os: [ubuntu-latest]
                 python-version: ['3.9']
+                airflow-version: ['<2.10']
 
         steps:
             - name: Checkout code
@@ -59,6 +59,7 @@ jobs:
               run: |
                 python -m pip install --upgrade pip setuptools wheel
                 python -m pip install pytest pytest-cov coveralls
+                python -m pip install "apache-airflow${{ matrix.airflow-version }}"
             - name: Run the test with coverage
               run: pytest --cov
     #         - name: Coveralls

dlairflow/test/test_postgresql.py

@@ -0,0 +1,32 @@
+# Licensed under a BSD-style 3-clause license - see LICENSE.md.
+# -*- coding: utf-8 -*-
+"""Test dlairflow.postgresql.
+"""
+import os
+from ..postgresql import pg_dump_schema, pg_restore_schema
+from airflow.hooks.base import BaseHook
+
+
+class MockConnection(object):
+
+    def __init__(self, connection):
+        foo = connection.split(',')
+        self.login = foo[0]
+        self.password = foo[1]
+        self.host = foo[2]
+        self.schema = foo[3]
+        return
+
+
+def test_pg_dump_schema(monkeypatch):
+    """Test pg_dump task.
+    """
+    def mock_connection(connection):
+        conn = MockConnection(connection)
+        return conn
+
+    monkeypatch.setattr(BaseHook, "get_connection", mock_connection)
+
+    test_operator = pg_dump_schema("login,password,host,schema")
+
+    print(dir(test_operator))