Tests

Author: yannbu

tests/fixtures/fake_builder.py renamed to tests/abstracts/__init__.py

@@ -0,0 +1,5 @@
+from zenaton.abstracts.job import Job
+
+
+def test_has_handle():
+    assert hasattr(Job(), 'handle')

tests/abstracts/test_job.py

@@ -0,0 +1,5 @@
+from zenaton.abstracts.task import Task
+
+
+def test_has_handle():
+    assert hasattr(Task(), 'handle')

tests/abstracts/test_task.py

@@ -0,0 +1,5 @@
+from zenaton.abstracts.workflow import Workflow
+
+
+def test_has_handle():
+    assert hasattr(Workflow(), 'handle')

tests/abstracts/test_workflow.py

@@ -1 +1,7 @@
-
+from .fixtures.fixtures import dummy_object
+from .fixtures.fixture_client import client
+from .fixtures.fixture_task import task0, task1, task2, task3
+from .fixtures.fixture_workflow import sequential_workflow, version_workflow
+from .fixtures.fixture_event import my_event
+from .fixtures.fixture_engine import engine
+from .fixtures.fixture_wait import wait, wait_event

tests/conftest.py

@@ -22,8 +22,5 @@
 
 
 @pytest.fixture
-def FakeClient():
+def client():
     return Client(app_id, api_token, app_env)
-
-
-client = Client(app_id, api_token, app_env)

tests/fixtures/fake_task.py

@@ -0,0 +1,10 @@
+import pytest
+
+from zenaton.engine import Engine
+from zenaton.client import Client
+
+
+@pytest.fixture
+def engine():
+    Client()
+    return Engine()

tests/fixtures/fake_workflow.py

@@ -0,0 +1,12 @@
+import pytest
+
+from zenaton.abstracts.event import Event
+
+
+class MyEvent(Event):
+    pass
+
+
+@pytest.fixture
+def my_event():
+    return MyEvent()

tests/fixtures/fixture_builder.py

@@ -0,0 +1,62 @@
+import time
+
+import pytest
+
+from zenaton.abstracts.task import Task
+from zenaton.traits.zenatonable import Zenatonable
+
+
+class Task0(Task, Zenatonable):
+
+    def handle(self):
+        print('Task 0 starts')
+        time.sleep(3)
+        print('Task 0 ends')
+        return 0
+
+
+@pytest.fixture
+def task0():
+    return Task0()
+
+
+class Task1(Task, Zenatonable):
+
+    def handle(self):
+        print('Task 1 starts')
+        time.sleep(5)
+        print('Task 1 ends')
+        return 1
+
+
+@pytest.fixture
+def task1():
+    return Task1()
+
+
+class Task2(Task, Zenatonable):
+
+    def handle(self):
+        print('Task 2 starts')
+        time.sleep(7)
+        print('Task 2 ends')
+        return 2
+
+
+@pytest.fixture
+def task2():
+    return Task2()
+
+
+class Task3(Task, Zenatonable):
+
+    def handle(self):
+        print('Task 3 starts')
+        time.sleep(9)
+        print('Task 3 ends')
+        return 3
+
+
+@pytest.fixture
+def task3():
+    return Task3()

tests/fixtures/fake_client.py renamed to tests/fixtures/fixture_client.py

@@ -0,0 +1,14 @@
+import pytest
+
+from zenaton.tasks.wait import Wait
+from .fixture_event import MyEvent
+
+
+@pytest.fixture
+def wait():
+    return Wait()
+
+
+@pytest.fixture
+def wait_event():
+    return Wait(MyEvent)

tests/fixtures/fixture_engine.py

@@ -0,0 +1,54 @@
+import pytest
+
+from .fixture_task import task0, task1, task2, task3
+from .fixture_event import MyEvent
+
+from zenaton.abstracts.workflow import Workflow
+from zenaton.traits.zenatonable import Zenatonable
+from zenaton.workflows.version import Version
+
+
+class SequentialWorkflow(Workflow, Zenatonable):
+
+    def __init__(self):
+        self.id_ = 'MyTestWorkflowId'
+
+    def handle(self):
+
+        a = task0.execute()
+
+        if a > 0:
+            task1.execute()
+        else:
+            task2.execute()
+
+        task3.execute()
+
+    def id(self):
+        return self.id_
+
+    def set_id(self, id_):
+        self.id_ = id_
+
+    def on_event(self, event):
+        if issubclass(MyEvent, type(event)):
+            return True
+        else:
+            return False
+
+
+
+@pytest.fixture
+def sequential_workflow():
+    return SequentialWorkflow()
+
+
+class VersionWorkflow(Version):
+
+    def versions(self):
+        return [SequentialWorkflow, ]
+
+
+@pytest.fixture
+def version_workflow():
+    return VersionWorkflow()

tests/fixtures/fixture_event.py

@@ -6,5 +6,5 @@ class DummyClass:
 
 
 @pytest.fixture
-def DummyObject():
+def dummy_object():
     return DummyClass()

tests/fixtures/fixture_task.py

@@ -1,8 +0,0 @@
-import pytest
-
-from zenaton.query.builder import Builder
-
-@pytest.mark.usefixtures("FakeSequentialWorkflow")
-def test_test(FakeSequentialWorkflow):
-    assert hasattr(FakeSequentialWorkflow, 'handle')
-

tests/fixtures/fixture_wait.py

@@ -0,0 +1,29 @@
+import pytest
+
+from zenaton.tasks.wait import Wait
+from zenaton.exceptions import ExternalError
+from ..fixtures.fixture_event import MyEvent
+
+
+def test_init():
+    with pytest.raises(ExternalError):
+        Wait(1)
+
+
+@pytest.mark.usefixtures("wait")
+def test_has_handle(wait):
+    assert hasattr(wait, 'handle')
+
+
+@pytest.mark.usefixtures("wait", "wait_event", "my_event")
+def test_valid_param(wait, my_event):
+    assert not wait.valid_param(1)
+    assert wait.valid_param(None)
+    assert wait.valid_param("foo")
+    assert wait.valid_param(MyEvent)
+
+
+@pytest.mark.usefixtures("wait", "my_event")
+def test_event_class(wait, my_event):
+    assert not wait.event_class("foo")
+    assert wait.event_class(MyEvent)