Add ci that runs tests on PR

.github/workflows/on_pr.yaml

@@ -0,0 +1,15 @@
+name: CI for verifying PR
+
+on:
+  pull_request:
+    branches:
+      - dev
+      - main
+    types: [ opened, synchronize, reopened ]
+    paths-ignore:
+      - README.md
+
+jobs:
+  run-tests:
+    uses: ./.github/workflows/run_tests.yaml
+    secrets: inherit

.github/workflows/pypi-publish.yaml

@@ -5,6 +5,9 @@ on:
 
 
 jobs:
+  run-tests:
+    uses: ./.github/workflows/run_tests.yaml
+    secrets: inherit
   pypi-publish:
     name: Upload release to PyPI
     runs-on: ubuntu-latest

.github/workflows/run_tests.yaml

@@ -0,0 +1,34 @@
+name: CI for unit testing
+
+on:
+  workflow_call:
+
+jobs:
+  execute-tests:
+    name: Run tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Install Python
+        id: setup-python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.9
+      - name: Install and configure Poetry
+        uses: snok/install-poetry@v1
+        with:
+          virtualenvs-create: true
+          virtualenvs-in-project: true
+      - name: Install dependencies
+        working-directory: .
+        run: poetry install --no-interaction --no-ansi -vvv
+        shell: bash
+      - name: Run tests
+        run: poetry run pytest tests/unit_tests --disable-warnings --full-trace
+        shell: bash
+      - name: Run tests
+        run: poetry run pytest tests/integration_tests --disable-warnings --full-trace
+        shell: bash
+        env:
+          PROD_API_TOKEN: ${{ secrets.PROD_API_TOKEN }}

tests/conftest.py

@@ -5,9 +5,12 @@
 import os
 
 import pytest
+from airflow import DAG
 from airflow.models import Connection
 from pytest_mock import MockerFixture
 
+from tests.unit_tests.operator.test_run import TEST_DAG_ID, DEFAULT_START
+
 
 @pytest.fixture(scope="function", autouse=True)
 def test_default_conn(mocker: MockerFixture):
@@ -23,3 +26,13 @@ def test_default_conn(mocker: MockerFixture):
 @pytest.fixture(scope="function")
 def clean_airflow_db():
     os.system("airflow db reset --yes")
+
+
+@pytest.fixture()
+def dag():
+    with DAG(
+        dag_id=TEST_DAG_ID,
+        schedule="@daily",
+        start_date=DEFAULT_START,
+    ) as dag:
+        yield dag

tests/integration_tests/conftest.py

@@ -10,18 +10,18 @@
 
 
 @pytest.fixture(scope="function")
-def staging_conn(mocker: MockerFixture):
-    staging_host = "api.staging.wherobots.com"
-    staging_api_token = os.getenv("STAGING_API_TOKEN")
-    if not staging_api_token:
-        raise ValueError("STAGING_API_TOKEN is not set")
-    staging_conn = Connection(
-        conn_id="wherobots_staging_conn",
+def prod_conn(mocker: MockerFixture):
+    prod_host = "api.cloud.wherobots.com"
+    prod_api_token = os.getenv("PROD_API_TOKEN")
+    if not prod_api_token:
+        raise ValueError("PROD_API_TOKEN is not set")
+    prod_conn = Connection(
+        conn_id="wherobots_prod_conn",
         conn_type="http",
-        host=staging_host,
-        password=staging_api_token,
+        host=prod_host,
+        password=prod_api_token,
     )
     mocker.patch.dict(
-        "os.environ", AIRFLOW_CONN_WHEROBOTS_STAGING_CONN=staging_conn.get_uri()
+        "os.environ", AIRFLOW_CONN_WHEROBOTS_PROD_CONN=prod_conn.get_uri()
     )
-    return staging_conn
+    return prod_conn

tests/integration_tests/operator/test_run.py

@@ -8,9 +8,10 @@
 import pytest
 from airflow import DAG
 from airflow.models import Connection
+from airflow.utils.state import TaskInstanceState
 
 from airflow_providers_wherobots.operators.run import WherobotsRunOperator
-from tests.unit_tests.operator.test_run import execute_dag
+from tests.unit_tests.operator.test_run import build_ti
 
 DEFAULT_START = pendulum.datetime(2021, 9, 13, tz="UTC")
 DEFAULT_END = DEFAULT_START + datetime.timedelta(days=1)
@@ -20,14 +21,16 @@
 
 
 @pytest.mark.usefixtures("clean_airflow_db")
-def test_staging_run_success(staging_conn: Connection, dag: DAG) -> None:
+def test_prod_run_success(prod_conn: Connection, dag: DAG) -> None:
     operator = WherobotsRunOperator(
-        wherobots_conn_id=staging_conn.conn_id,
+        wherobots_conn_id=prod_conn.conn_id,
         task_id="test_run_smoke",
         name="airflow_operator_test_run_{{ ts_nodash }}",
         run_python={
-            "uri": "s3://wbts-wbc-rcv7vl73oy/hao9o6y8ci/data/customer-z4asgjn7clrcbz/very_simple_job.py"
+            "uri": "s3://wbts-wbc-m97rcg45xi/42ly7mi0p1/data/shared/very_simple_job.py"
         },
         dag=dag,
     )
-    execute_dag(dag, task_id=operator.task_id)
+    ti = build_ti(dag, task_id=operator.task_id)
+    ti.run(ignore_ti_state=True)
+    assert ti.state == TaskInstanceState.SUCCESS

tests/unit_tests/operator/test_run.py

@@ -33,16 +33,6 @@
 TEST_TASK_ID = "run_operator"
 
 
-@pytest.fixture()
-def dag():
-    with DAG(
-        dag_id=TEST_DAG_ID,
-        schedule="@daily",
-        start_date=DEFAULT_START,
-    ) as dag:
-        yield dag
-
-
 def build_ti(dag: DAG, task_id: str, start=DEFAULT_START, end=DEFAULT_END):
     dag_run: DagRun = dag.create_dagrun(
         state=DagRunState.RUNNING,