From d7a73b26a1d8b1fa97a23f1c1b57005e5c087583 Mon Sep 17 00:00:00 2001
From: jrybicki-jsc <j.rybicki@fz-juelich.de>
Date: Wed, 1 Sep 2021 11:35:41 +0200
Subject: [PATCH] adding test stub
---
.gitlab-ci.yml | 2 ++
tests/test_b2shareoperator.py | 22 ++++++++++++++++++++++
tests/test_dag.py | 13 +++++++++++++
3 files changed, 37 insertions(+)
create mode 100644 tests/test_b2shareoperator.py
create mode 100644 tests/test_dag.py
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 37127cb..862ece7 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -15,6 +15,7 @@ test:
before_script:
- airflow db init
- pip install -r requirements.txt
+ - pip install nose==1.3.7
- airflow connections add --conn-uri https://b2share-testing.fz-juelich.de/ default_b2share
script:
- ls
@@ -23,4 +24,5 @@ test:
- airflow dags list
- airflow connections list
- airflow dags test firsto 2021-08-18
+ - nosetests
diff --git a/tests/test_b2shareoperator.py b/tests/test_b2shareoperator.py
new file mode 100644
index 0000000..6d9adf2
--- /dev/null
+++ b/tests/test_b2shareoperator.py
@@ -0,0 +1,22 @@
+import unittest
+from airflow.utils.state import State
+from dags.b2shareoperator import B2ShareOperator
+from airflow import DAG
+from airflow.models.taskinstance import TaskInstance
+
+DEFAULT_DATE = '2019-10-03'
+TEST_DAG_ID = 'test_my_custom_operator'
+
+class B2ShareOperatorTest(unittest.TestCase):
+ def setUp(self):
+ self.dag = DAG(TEST_DAG_ID, schedule_interval='@daily', default_args={'start_date' : DEFAULT_DATE})
+ self.op = B2ShareOperator(
+ dag=self.dag,
+ task_id='test',
+ )
+ self.ti = TaskInstance(task=self.op, execution_date=DEFAULT_DATE)
+
+ def test_execute_no_trigger(self):
+ self.ti.run(ignore_ti_state=True)
+ assert self.ti.state == State.SUCCESS
+ # Assert something related to tasks results
\ No newline at end of file
diff --git a/tests/test_dag.py b/tests/test_dag.py
new file mode 100644
index 0000000..cfc812e
--- /dev/null
+++ b/tests/test_dag.py
@@ -0,0 +1,13 @@
+from airflow.models import DagBag
+import unittest
+
+class TestADag(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.dagbag = DagBag()
+
+ def test_dag_loaded(self):
+ dag = self.dagbag.get_dag(dag_id='firsto')
+ assert self.dagbag.import_errors == {}
+ assert dag is not None
+ assert len(dag.tasks) == 1
\ No newline at end of file
--
GitLab