Repository: incubator-airflow Updated Branches: refs/heads/master 24d41b890 -> a59ba049c
[AIRFLOW-269] Add some unit tests for PostgreSQL Closes #1616 from sekikn/AIRFLOW-269 Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/a59ba049 Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/a59ba049 Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/a59ba049 Branch: refs/heads/master Commit: a59ba049c6f25b80492dab366b8b6f0757fe5f1a Parents: 24d41b8 Author: Kengo Seki <[email protected]> Authored: Thu Jun 30 13:19:30 2016 -0700 Committer: Chris Riccomini <[email protected]> Committed: Thu Jun 30 13:19:30 2016 -0700 ---------------------------------------------------------------------- .travis.yml | 1 + tests/operators/operators.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/a59ba049/.travis.yml ---------------------------------------------------------------------- diff --git a/.travis.yml b/.travis.yml index 87c0183..0b52183 100644 --- a/.travis.yml +++ b/.travis.yml @@ -23,6 +23,7 @@ addons: - slapd - ldap-utils - openssh-server + postgresql: "9.2" python: - "2.7" - "3.4" http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/a59ba049/tests/operators/operators.py ---------------------------------------------------------------------- diff --git a/tests/operators/operators.py b/tests/operators/operators.py index d9bc0c2..2365ba0 100644 --- a/tests/operators/operators.py +++ b/tests/operators/operators.py @@ -125,6 +125,40 @@ class PostgresTest(unittest.TestCase): end_date=DEFAULT_DATE, force=True) + def postgres_operator_test_multi(self): + sql = [ + "TRUNCATE TABLE test_airflow", + "INSERT INTO test_airflow VALUES ('X')", + ] + import airflow.operators.postgres_operator + t = operators.postgres_operator.PostgresOperator( + task_id='postgres_operator_test_multi', sql=sql, dag=self.dag) + t.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, force=True) + + def test_postgres_to_postgres(self): + sql = "SELECT * FROM INFORMATION_SCHEMA.TABLES LIMIT 100;" + import airflow.operators.generic_transfer + t = operators.generic_transfer.GenericTransfer( + task_id='test_p2p', + preoperator=[ + "DROP TABLE IF EXISTS test_postgres_to_postgres", + "CREATE TABLE IF NOT EXISTS " + "test_postgres_to_postgres (LIKE INFORMATION_SCHEMA.TABLES)" + ], + source_conn_id='postgres_default', + destination_conn_id='postgres_default', + destination_table="test_postgres_to_postgres", + sql=sql, + dag=self.dag) + t.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, force=True) + + def test_sql_sensor(self): + t = operators.sensors.SqlSensor( + task_id='sql_sensor_check', + conn_id='postgres_default', + sql="SELECT count(1) FROM INFORMATION_SCHEMA.TABLES", + dag=self.dag) + t.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, force=True) @skipUnlessImported('airflow.operators.hive_operator', 'HiveOperator') @skipUnlessImported('airflow.operators.postgres_operator', 'PostgresOperator')
