Merge pull request #6 from airflow-laminar/tkp/start

Initialize common utilities

airflow_common_operators/__init__.py

@@ -1 +1,3 @@
+from .common import *
+
 __version__ = "0.1.0"

airflow_common_operators/common.py

@@ -0,0 +1,35 @@
+from subprocess import call
+from typing import Callable
+
+from airflow.exceptions import AirflowFailException, AirflowSkipException
+
+__all__ = (
+    "skip",
+    "fail",
+    "pass_",
+    "ping",
+)
+
+
+def skip():
+    raise AirflowSkipException
+
+
+def fail():
+    raise AirflowFailException
+
+
+def pass_():
+    pass
+
+
+def ping(host, *, local=True) -> Callable:
+    if host != "localhost" and host.count(".") == 0 and local:
+        host = f"{host}.local"
+
+    def _ping(hostname=host):
+        if call(["ping", "-c", "1", f"{hostname}"]) == 0:
+            return True
+        raise AirflowSkipException
+
+    return _ping

airflow_common_operators/tests/test_common.py

@@ -0,0 +1,30 @@
+from unittest.mock import patch
+
+import pytest
+from airflow.exceptions import AirflowFailException, AirflowSkipException
+
+from airflow_common_operators import fail, pass_, ping, skip
+
+
+class TestCommon:
+    def test_pass(self):
+        pass_()
+
+    def test_skip(self):
+        with pytest.raises(AirflowSkipException):
+            skip()
+
+    def test_fail(self):
+        with pytest.raises(AirflowFailException):
+            fail()
+
+    def test_ping(self):
+        assert ping("localhost")()
+        with pytest.raises(AirflowSkipException):
+            ping("nonexistent")()
+
+    def test_ping_localappend(self):
+        with patch("airflow_common_operators.common.call") as call:
+            call.return_value = 0
+            ping("blerg")()
+            assert call.call_args[0][0] == ["ping", "-c", "1", "blerg.local"]