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: Unit tests
+ run: poetry run pytest tests/unit_tests --disable-warnings --full-trace
+ shell: bash
+ - name: Integration 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,