testing bags

tests/core/test_inputs.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2012-2014 Romain Dorgueil
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from queue import Empty
+
+import pytest
+
+from bonobo.core.errors import InactiveWritableError, InactiveReadableError
+from bonobo.core.inputs import Input
+from bonobo.util.tokens import Begin, End
+
+
+def test_input_runlevels():
+    q = Input()
+
+    # Before BEGIN, noone should be able to write in an Input queue.
+    assert not q.alive
+    with pytest.raises(InactiveWritableError):
+        q.put('hello, unborn queue.')
+
+    # Begin
+    q.put(Begin)
+    assert q.alive and q._runlevel == 1
+    q.put('foo')
+
+    # Second Begin
+    q.put(Begin)
+    assert q.alive and q._runlevel == 2
+    q.put('bar')
+    q.put(End)
+
+    # FIFO
+    assert q.get() == 'foo'
+    assert q.get() == 'bar'
+
+    # self.assertEqual(q.alive, False) XXX queue don't know it's dead yet, but it is ...
+    # Async get raises Empty (End is not returned)
+    with pytest.raises(Empty):
+        q.get(block=False)
+    assert q.alive
+
+    # Before killing, let's slide some data in.
+    q.put('baz')
+
+    # Now kill the queue...
+    q.put(End)
+    with pytest.raises(InactiveWritableError):
+        q.put('foo')
+
+    # Still can get remaining data
+    assert q.get() == 'baz'
+    with pytest.raises(InactiveReadableError):
+        q.get()