Repository: thoughteer/edera

Issue:
Adapt routines to Python 3.7
Raising StopIteration manually is prohibited in Python 3.7: https://www.python.org/dev/peps/pep-0479/. Current routines rely on this feature, so we need a nice workaround.

Also, we need to try to clean up the stack trace of the routine (if possible). It's definitely possible in Python 2.7, because you can seamlessly change the traceback when raising an exception, but is not that easy in Python 3.x.

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index d1ed2a1..c9d5a8a 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,3 +1,4 @@
+dist: xenial
 sudo: false
 if: tag IS blank
 services:
@@ -6,8 +7,7 @@ language: python
 python:
   - "2.7"
   - "3.4"
-  - "3.5"
-  - "3.6"
+  - "3.7"
   - "pypy"
   - "pypy3.5"
 env:
@@ -28,7 +28,7 @@ jobs:
   include:
     - stage: release
       if: branch = master
-      python: 3.6
+      python: 3.7
       install: skip
       script: skip
       deploy:
diff --git a/edera/consumer.py b/edera/consumer.py
new file mode 100644
index 0000000..e37f77e
--- /dev/null
+++ b/edera/consumer.py
@@ -0,0 +1,25 @@
+import abc
+
+import six
+
+
+@six.add_metaclass(abc.ABCMeta)
+class Consumer(object):
+    """
+    A generic consumer protocol.
+
+    You can push elements to the consumer.
+    It's up to the consumer how to treat the elements.
+    """
+
+    @abc.abstractmethod
+    def push(self, element):
+        """
+        Push the element to the consumer.
+
+        Args:
+            element (Any)
+
+        Raises:
+            ConsumptionError if the consumer failed to accept the element
+        """
diff --git a/edera/consumers/__init__.py b/edera/consumers/__init__.py
new file mode 100644
index 0000000..d131c98
--- /dev/null
+++ b/edera/consumers/__init__.py
@@ -0,0 +1,1 @@
+from .interprocess import InterProcessConsumer
diff --git a/edera/consumers/interprocess.py b/edera/consumers/interprocess.py
new file mode 100644
index 0000000..658599a
--- /dev/null
+++ b/edera/consumers/interprocess.py
@@ -0,0 +1,67 @@
+import logging
+import multiprocessing
+
+import edera.helpers
+
+from edera.exceptions import ConsumptionError
+from edera.routine import deferrable
+from edera.routine import routine
+
+
+class InterProcessConsumer(object):
+    """
+    An inter-process consumer that buffers elements in a queue.
+
+    In order to start handling buffered elements you need to call $consume.
+
+    Attributes:
+        handler (Callable[[Any], Any]) - the handler called on every push
+        capacity (Integer) - the limit on the number of pending elements in the queue
+        backoff (TimeDelta) - the delay after each handling failure
+    """
+
+    def __init__(self, handler, capacity, backoff):
+        """
+        Args:
+            handler (Callable[[Any], Any]) - a handler to call on every push
+            capacity (Integer) - a limit on the number of pending elements in the queue
+            backoff (TimeDelta) - a delay after each handling failure
+        """
+        self.handler = handler
+        self.capacity = capacity
+        self.backoff = backoff
+        self.__fifo = multiprocessing.Queue(capacity)
+
+    @routine
+    def consume(self):
+        """
+        Run an infinite consumption loop.
+
+        Ignores all errors that occur in the handler, but logs them at the DEBUG level.
+        """
+        element = Void
+        while True:
+            if element is Void:
+                try:
+                    element = self.__fifo.get(False)
+                except Exception:
+                    yield edera.helpers.sleep.defer(self.backoff)
+                    continue
+            try:
+                yield deferrable(self.handler).defer(element)
+            except Exception as error:
+                logging.getLogger(__name__).debug("Failed to handle %r: %s", element, error)
+                yield edera.helpers.sleep.defer(self.backoff)
+            else:
+                element = Void
+                yield
+
+    def push(self, element):
+        try:
+            self.__fifo.put(element, False)
+        except Exception:
+            raise ConsumptionError("FIFO is full")
+
+
+class Void(object):
+    pass
diff --git a/edera/daemon/daemon.py b/edera/daemon/daemon.py
index cf81ae3..005c649 100644
--- a/edera/daemon/daemon.py
+++ b/edera/daemon/daemon.py
@@ -10,6 +10,7 @@ import six
 
 import edera.helpers
 
+from edera.consumers import InterProcessConsumer
 from edera.flags import InterProcessFlag
 from edera.flags import InterThreadFlag
 from edera.helpers import Sasha
@@ -72,6 +73,10 @@ class Daemon(object):
                 filtering by tag
         support (Optional[DaemonModule])
 
+    Constants:
+        CONSUMER_CAPACITY (Integer) - the capacity of the consumer that serves monitoring agents
+        CONSUMER_BACKOFF (TimeDelta) - the backoff of the consumer that serves monitoring agents
+
     See also:
         $DaemonAutoTester
         $DaemonModule
@@ -82,6 +87,15 @@ class Daemon(object):
         See $ProcessWorker's documentation for details.
     """
 
+    CONSUMER_CAPACITY = 1000
+    CONSUMER_BACKOFF = datetime.timedelta(seconds=1)
+
+    def __init__(self):
+        self.__consumer = InterProcessConsumer(
+            lambda kv: self.monitor.put(*kv),
+            self.CONSUMER_CAPACITY,
+            self.CONSUMER_BACKOFF)
+
     @property
     def autotester(self):
         pass
@@ -98,13 +112,12 @@ class Daemon(object):
     def executor(self):
         result = BasicWorkflowExecutor()
         if self.monitor is not None:
-            def agency():
-                host_name = socket.getfqdn()
-                process_name = multiprocessing.current_process().name
-                thread_name = threading.current_thread().name
-                agent_name = ":".join([host_name, process_name, thread_name])
-                return MonitoringAgent(agent_name, self.monitor)
-            result = MonitoringWorkflowExecutor(result, agency)
+            host_name = socket.getfqdn()
+            process_name = multiprocessing.current_process().name
+            thread_name = threading.current_thread().name
+            agent_name = ":".join([host_name, process_name, thread_name])
+            agent = MonitoringAgent(agent_name, self.monitor, self.__consumer)
+            result = MonitoringWorkflowExecutor(result, agent)
         result = ManagedWorkflowExecutor(result, self.manager)
         return result
 
@@ -205,6 +218,10 @@ class Daemon(object):
     @routine
     def __run(self):
 
+        @routine
+        def run_consumer():
+            yield self.__consumer.consume.defer()
+
         @routine
         def run_support():
             if self.support is not None:
@@ -219,6 +236,7 @@ class Daemon(object):
         timeout = 2 * self.interruption_timeout
         yield MultiProcessInvoker(
             {
+                "consumer": run_consumer,
                 "launcher#support": run_support,
                 "launcher#main": run_main,
             },
diff --git a/edera/demo/daemon/beans/__init__.py b/edera/demo/daemon/beans/__init__.py
index 51e9b18..cae65e3 100644
--- a/edera/demo/daemon/beans/__init__.py
+++ b/edera/demo/daemon/beans/__init__.py
@@ -11,7 +11,7 @@ def colorful(bean_class):
 
 
 def threadsafe(bean_class):
-    return split(os.getpid, threading.current_thread)(bean_class)
+    return split(os.getpid, lambda: threading.current_thread().ident)(bean_class)
 
 
 Beanbag(__name__)
diff --git a/edera/exceptions.py b/edera/exceptions.py
index 486f80a..c424278 100644
--- a/edera/exceptions.py
+++ b/edera/exceptions.py
@@ -244,3 +244,9 @@ class MonitorInconsistencyError(Error):
     That is a big no-no.
     Use a locker to prevent parallel execution.
     """
+
+
+class ConsumptionError(Error):
+    """
+    The consumer could not accept an element.
+    """
diff --git a/edera/helpers/beanbag.py b/edera/helpers/beanbag.py
index 045b41e..60f84a7 100644
--- a/edera/helpers/beanbag.py
+++ b/edera/helpers/beanbag.py
@@ -65,6 +65,9 @@ def split(*selectors):
     Args:
         selectors (Tuple[Callable[[], Any]...])
 
+    Returns:
+        Type
+
     See also:
         $MultiBox
         $Proxy
diff --git a/edera/helpers/proxy.py b/edera/helpers/proxy.py
index a2831ab..94b3895 100644
--- a/edera/helpers/proxy.py
+++ b/edera/helpers/proxy.py
@@ -46,10 +46,10 @@ class Proxy(object):
         [1, 2]
     """
 
-    def __new__(cls, instance_holder, subject_class, *args, **kwargs):
+    def __new__(cls, subject_holder, subject_class, *args, **kwargs):
         """
         Args:
-            instance_holder (Box) - a box to store the subject in
+            subject_holder (Box) - a box to store the subject in
             subject_class (Type) - a class to proxy
             *args, **kwargs - arguments to pass to the subject constructor
         """
@@ -60,20 +60,17 @@ class Proxy(object):
                 return subject_class(*args, **kwargs)
 
             def method(self, *args, **kwargs):
-                subject = instance_holder.get()
+                subject = subject_holder.get()
                 if subject is None:
                     subject = instantiate()
-                    instance_holder.put(subject)
+                    subject_holder.put(subject)
                 return object.__getattribute__(subject, method_name)(*args, **kwargs)
 
             return method
 
-        body = dict(object.__dict__)
-        special_subject_class_methods = {
-            name
-            for name, _ in inspect.getmembers(subject_class)
-            if name in SPECIAL_METHODS
+        body = {
+            method_name: proxy(method_name)
+            for method_name, _ in inspect.getmembers(subject_class)
+            if method_name in SPECIAL_METHODS
         }
-        for method_name in special_subject_class_methods:
-            body[method_name] = proxy(method_name)
-        return type.__new__(type, subject_class.__name__, subject_class.__bases__, body)()
+        return type.__new__(subject_class.__class__, subject_class.__name__, (), body)()
diff --git a/edera/monitoring/agent.py b/edera/monitoring/agent.py
index abb13e2..e0ed1bc 100644
--- a/edera/monitoring/agent.py
+++ b/edera/monitoring/agent.py
@@ -5,6 +5,7 @@ import traceback
 import edera.helpers
 
 from edera.condition import ConditionWrapper
+from edera.exceptions import ConsumptionError
 from edera.exceptions import ExcusableError
 from edera.exceptions import StorageOperationError
 from edera.helpers import CurrentException
@@ -26,25 +27,34 @@ class MonitoringAgent(Nameable):
     Agents are used to push updates into a storage during workflow execution.
     They are also used to retrieve updates from the storage.
 
+    Agents (unless they are read-only) use a consumer to push updates.
+    This allows features like buffering to be implemented naturally.
+
     Default agents:
       - track the topology of the task graph
       - publish task announcements
       - report statuses of the tasks
       - collect logs during task execution
 
+    Attributes:
+        readonly (Boolean) - True iff the agent does not have a consumer to push through
+
     See also:
         $MonitoringWorkflowExecutor
         $MonitorWatcher
     """
 
-    def __init__(self, name, monitor):
+    def __init__(self, name, monitor, consumer=None):
         """
         Args:
             name (String) - a unique name for the agent
             monitor (Storage) - a storage used to store all relevant information
+            consumer (Optional[Consumer]) - a consumer that accepts key-value pairs
+                We presume that this consumer puts records to the same $monitor.
         """
         self.__name = name
         self.__monitor = monitor
+        self.__consumer = consumer
         self.__key = "update/" + name
 
     @classmethod
@@ -52,6 +62,8 @@ class MonitoringAgent(Nameable):
         """
         Get the set of all agents registered in the monitor.
 
+        All discovered agents are read-only.
+
         Args:
             monitor (Storage) - a storage to search for agents
 
@@ -90,11 +102,7 @@ class MonitoringAgent(Nameable):
 
         Returns:
             Graph - an altered task graph
-
-        Raises:
-            StorageOperationError if something went wrong with the storage
         """
-        self.__monitor.put("agent", self.name)
         dependencies = {
             task.name: {parent.name for parent in workflow[task].parents}
             for task in workflow
@@ -104,7 +112,10 @@ class MonitoringAgent(Nameable):
             task.name: workflow[task].annotation.get("announcement")
             for task in workflow
         }
-        self.push(WorkflowUpdate(dependencies, phonies, announcements))
+        workflow_update = WorkflowUpdate(dependencies, phonies, announcements)
+        if not self.register() or not self.push(workflow_update):
+            logging.getLogger(__name__).warning("This time monitoring will be disabled")
+            return workflow
         result = workflow.clone()
         for task in result:
             if task.execute is Phony:
@@ -144,10 +155,38 @@ class MonitoringAgent(Nameable):
         Args:
             update (MonitoringSnapshotUpdate) - an update to push
 
+        Returns:
+            Boolean - whether the operation succeeded
+
         Raises:
-            StorageOperationError if something went wrong with the storage
+            AssertionError if the agent is read-only
+        """
+        return self.__push(self.__key, update.serialize())
+
+    @property
+    def readonly(self):
+        return self.__consumer is None
+
+    def register(self):
+        """
+        Register the agent in the monitor.
+
+        Returns:
+            Boolean - whether the operation succeeded
+
+        Raises:
+            AssertionError if the agent is read-only
         """
-        self.__monitor.put(self.__key, update.serialize())
+        return self.__push("agent", self.name)
+
+    def __push(self, key, value):
+        assert not self.readonly
+        try:
+            self.__consumer.push((key, value))
+        except ConsumptionError:
+            logging.getLogger(__name__).warning("Consumer rejected a monitoring record")
+            return False
+        return True
 
 
 class StatusReportingTaskWrapper(TaskWrapper):
@@ -202,17 +241,11 @@ class StatusReportingTaskWrapper(TaskWrapper):
             return StatusReportingConditionWrapper(condition, self, self.__agent)
 
     def __report_status(self, status):
-        try:
-            self.__agent.push(TaskStatusUpdate(self.name, status, edera.helpers.now()))
-        except StorageOperationError as error:
-            logging.getLogger(__name__).warning("Failed to report the status: %s", error)
+        self.__agent.push(TaskStatusUpdate(self.name, status, edera.helpers.now()))
 
     def __save_traceback(self):
-        try:
-            message = traceback.format_exc()
-            self.__agent.push(TaskLogUpdate(self.name, message, edera.helpers.now()))
-        except StorageOperationError as error:
-            logging.getLogger(__name__).warning("Failed to save the traceback: %s", error)
+        message = traceback.format_exc()
+        self.__agent.push(TaskLogUpdate(self.name, message, edera.helpers.now()))
 
 
 class StatusReportingConditionWrapper(ConditionWrapper):
@@ -250,17 +283,11 @@ class StatusReportingConditionWrapper(ConditionWrapper):
         yield result
 
     def __report_status(self, status):
-        try:
-            self.__agent.push(TaskStatusUpdate(self.__task.name, status, edera.helpers.now()))
-        except StorageOperationError as error:
-            logging.getLogger(__name__).warning("Failed to report the status: %s", error)
+        self.__agent.push(TaskStatusUpdate(self.__task.name, status, edera.helpers.now()))
 
     def __save_traceback(self):
-        try:
-            message = traceback.format_exc()
-            self.__agent.push(TaskLogUpdate(self.__task.name, message, edera.helpers.now()))
-        except StorageOperationError as error:
-            logging.getLogger(__name__).warning("Failed to save the traceback: %s", error)
+        message = traceback.format_exc()
+        self.__agent.push(TaskLogUpdate(self.__task.name, message, edera.helpers.now()))
 
 
 class LogCapturingTaskWrapper(TaskWrapper):
@@ -283,11 +310,8 @@ class LogCapturingTaskWrapper(TaskWrapper):
         def emit(self, record):
             if threading.current_thread() != self.__thread:
                 return
-            try:
-                message = self.format(record)
-                self.__agent.push(TaskLogUpdate(self.__task, message, edera.helpers.now()))
-            except StorageOperationError as error:
-                logging.getLogger(__name__).warning("Failed to save the log message: %s", error)
+            message = self.format(record)
+            self.__agent.push(TaskLogUpdate(self.__task, message, edera.helpers.now()))
 
     def __init__(self, base, agent):
         """
diff --git a/edera/routine.py b/edera/routine.py
index 1d9f2a7..17b2f93 100644
--- a/edera/routine.py
+++ b/edera/routine.py
@@ -86,7 +86,10 @@ class Routine(object):
             generator = self.__core(*args, **kwargs)
             seed, exception = None, None
             while True:
-                feed = generator.send(seed) if exception is None else generator.throw(*exception)
+                try:
+                    feed = generator.send(seed) if exception is None else generator.throw(*exception)
+                except StopIteration:
+                    return
                 if isinstance(feed, DeferredRoutineCall):
                     feed.instance = feed.instance[auditor]
                 try:
diff --git a/edera/workflow/executors/monitoring.py b/edera/workflow/executors/monitoring.py
index 967bad8..cb03407 100644
--- a/edera/workflow/executors/monitoring.py
+++ b/edera/workflow/executors/monitoring.py
@@ -11,15 +11,15 @@ class MonitoringWorkflowExecutor(WorkflowExecutor):
         $MonitoringAgent
     """
 
-    def __init__(self, base, agency):
+    def __init__(self, base, agent):
         """
         Args:
             base (WorkflowExecutor) - a base workflow executor
-            agency (Callable[[], MonitoringAgent]) - a monitoring agency (assigns agents on demand)
+            agent (MonitoringAgent) - a monitoring agent
         """
         self.__base = base
-        self.__agency = agency
+        self.__agent = agent
 
     @routine
     def execute(self, workflow):
-        yield deferrable(self.__base.execute).defer(self.__agency().embrace(workflow))
+        yield deferrable(self.__base.execute).defer(self.__agent.embrace(workflow))
diff --git a/edera/workflow/processors/target_cacher.py b/edera/workflow/processors/target_cacher.py
index 240eaac..a19cd98 100644
--- a/edera/workflow/processors/target_cacher.py
+++ b/edera/workflow/processors/target_cacher.py
@@ -76,14 +76,16 @@ class CachingConditionWrapper(ConditionWrapper):
                 yield True
                 return
             cached = False
-        except StorageOperationError:
-            pass  # not sure if really not cached
+        except StorageOperationError as error:
+            # not sure if really not cached
+            logging.getLogger(__name__).debug("Failed to read from cache: %s", error)
         result = yield deferrable(super(CachingConditionWrapper, self).check).defer()
         if result and not cached:
             try:
                 logging.getLogger(__name__).debug("Caching %r", self)
                 self.cache.put(edera.helpers.sha1(self.name), "!")
                 logging.getLogger(__name__).debug("Stored in cache")
-            except StorageOperationError:
-                pass  # whatever
+            except StorageOperationError as error:
+                # whatever
+                logging.getLogger(__name__).debug("Failed to write to cache: %s", error)
         yield result
diff --git a/edera/workflow/processors/task_segregator.py b/edera/workflow/processors/task_segregator.py
index 09a8cd0..54997f4 100644
--- a/edera/workflow/processors/task_segregator.py
+++ b/edera/workflow/processors/task_segregator.py
@@ -1,3 +1,5 @@
+import logging
+
 from edera.condition import ConditionWrapper
 from edera.helpers import Phony
 from edera.routine import deferrable
@@ -62,12 +64,14 @@ class SegregatingTaskWrapper(TaskWrapper):
 
     @routine
     def execute(self):
-        assert self.__box.get() is None
+        assert self.__box.get() is None, "color box already contains `%s`" % self.__box.get()
         self.__box.put(self.__color)
+        logging.getLogger(__name__).debug("Switched to color `%s`", self.__color)
         try:
             yield deferrable(super(SegregatingTaskWrapper, self).execute).defer()
         finally:
             self.__box.put(None)
+            logging.getLogger(__name__).debug("Emptied the color box")
 
     @property
     def target(self):
@@ -97,10 +101,12 @@ class SegregatingConditionWrapper(ConditionWrapper):
 
     @routine
     def check(self):
-        assert self.__box.get() is None
+        assert self.__box.get() is None, "color box already contains `%s`" % self.__box.get()
         self.__box.put(self.__color)
+        logging.getLogger(__name__).debug("Switched to color `%s`", self.__color)
         try:
             result = yield deferrable(super(SegregatingConditionWrapper, self).check).defer()
             yield result
         finally:
             self.__box.put(None)
+            logging.getLogger(__name__).debug("Emptied the color box")
diff --git a/edera/workflow/processors/workflow_trimmer.py b/edera/workflow/processors/workflow_trimmer.py
index db58aab..48938b1 100644
--- a/edera/workflow/processors/workflow_trimmer.py
+++ b/edera/workflow/processors/workflow_trimmer.py
@@ -92,8 +92,10 @@ class WorkflowTrimmer(WorkflowProcessor):
                 logging.getLogger(__name__).debug("Cutting at %r of volume %d", target, volume)
                 try:
                     completed = yield deferrable(target.check).defer()
-                except Exception as error:
-                    logging.getLogger(__name__).warning("Failed to check %r: %s", target, error)
+                except ExcusableError as error:
+                    logging.getLogger(__name__).info("Stopped checking %r: %s", target, error)
+                except Exception:
+                    logging.getLogger(__name__).exception("Failed to check %r:", target)
                 else:
                     if completed:
                         logging.getLogger(__name__).debug("Blacklisting ancestors")
diff --git a/setup.py b/setup.py
index 82f613f..3849c4e 100644
--- a/setup.py
+++ b/setup.py
@@ -17,6 +17,7 @@ setuptools.setup(
         "Programming Language :: Python :: 3.4",
         "Programming Language :: Python :: 3.5",
         "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
         "Topic :: System :: Distributed Computing",
     ],
     keywords="workflow task schedule distributed dependency management",
diff --git a/tox.ini b/tox.ini
index 5d86de5..33271dd 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
 [tox]
-envlist = py{27,34,35,36},pypy,pypy3
+envlist = py{27,34,37},pypy,pypy3
 skip_missing_interpreters = True
 
 [testenv]


Test patch:
diff --git a/tests/integration/consumers/__init__.py b/tests/integration/consumers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/integration/consumers/test_interprocess_consumer.py b/tests/integration/consumers/test_interprocess_consumer.py
new file mode 100644
index 0000000..670299c
--- /dev/null
+++ b/tests/integration/consumers/test_interprocess_consumer.py
@@ -0,0 +1,41 @@
+import datetime
+import multiprocessing
+
+import pytest
+
+from edera.consumers import InterProcessConsumer
+from edera.exceptions import ConsumptionError
+from edera.invokers import MultiProcessInvoker
+
+
+def test_consumer_limits_its_capacity():
+    consumer = InterProcessConsumer(lambda element: None, 3, datetime.timedelta(seconds=0))
+    consumer.push(1)
+    consumer.push(2)
+    consumer.push(3)
+    with pytest.raises(ConsumptionError):
+        consumer.push(0)
+
+
+def test_consumer_handles_elements_correctly():
+
+    def push():
+        for element in range(1, 4):
+            consumer.push(element)
+
+    def handle(element):
+        result.put(element)
+
+    def interrupt():
+        if datetime.datetime.utcnow() > start_time + datetime.timedelta(seconds=5):
+            raise SystemExit
+
+    consumer = InterProcessConsumer(handle, 3, datetime.timedelta(seconds=0.1))
+    result = multiprocessing.Queue()
+    invoker = MultiProcessInvoker({"p": push, "c": consumer.consume})
+    start_time = datetime.datetime.utcnow()
+    try:
+        invoker.invoke[interrupt]()
+    except SystemExit:
+        pass
+    assert [result.get(timeout=1.0) for _ in range(3)] == [1, 2, 3]
diff --git a/tests/integration/lockers/test_directory_locker.py b/tests/integration/lockers/test_directory_locker.py
index 739ad25..62b41d0 100644
--- a/tests/integration/lockers/test_directory_locker.py
+++ b/tests/integration/lockers/test_directory_locker.py
@@ -20,7 +20,7 @@ def test_sigkill_releases_locks(directory_locker):
     target_process = multiprocessing.Process(target=lock_and_wait, args=(ready_flag,))
     target_process.daemon = True
     target_process.start()
-    ready_flag.wait(3.0)
+    ready_flag.wait(10.0)
     assert ready_flag.is_set()
     with pytest.raises(LockAcquisitionError):
         with directory_locker.lock("key"):
diff --git a/tests/system/demo/test_demo_daemon.py b/tests/system/demo/test_demo_daemon.py
index dd7c5f6..42c98c2 100644
--- a/tests/system/demo/test_demo_daemon.py
+++ b/tests/system/demo/test_demo_daemon.py
@@ -46,7 +46,7 @@ def test_demo_daemon_works_fine_in_testing_mode(tmpdir):
     command = ["python", "-m", "edera.demo.daemon", "-d", "-t", str(tmpdir), "-s", "0"]
     process = subprocess.Popen(command, stderr=subprocess.PIPE)
     failure = start_analysis(process.stderr)
-    failure.wait(timeout=90.0)
+    failure.wait(timeout=180.0)
     try:
         assert not failure.is_set()
     finally:
@@ -62,7 +62,7 @@ def test_demo_daemon_works_fine_in_production_mode(debugger, tmpdir):
     command = ["python", "-m", "edera.demo.daemon", "-d", str(tmpdir), "-s", "0"]
     process = subprocess.Popen(command, stderr=subprocess.PIPE)
     failure = start_analysis(process.stderr)
-    failure.wait(timeout=90.0)
+    failure.wait(timeout=120.0)
     try:
         assert not failure.is_set()
     finally:
diff --git a/tests/unit/helpers/test_proxy.py b/tests/unit/helpers/test_proxy.py
index 878e2c7..53ba546 100644
--- a/tests/unit/helpers/test_proxy.py
+++ b/tests/unit/helpers/test_proxy.py
@@ -1,5 +1,6 @@
 import pytest
 
+from edera.helpers import MultiBox
 from edera.helpers import Proxy
 from edera.helpers import SimpleBox
 
@@ -41,3 +42,48 @@ def test_proxy_reinstantiates_subject_when_needed():
     assert proxy.counter == 1
     box.put(None)
     assert proxy.counter == 2
+
+
+def test_proxy_can_proxy_proxies_to_builtin_derivatives():
+
+    class T(tuple):
+        def __new__(cls):
+            return ()
+
+    class P(T):
+        def __new__(cls):
+            return Proxy(SimpleBox(), T)
+
+    assert len(Proxy(SimpleBox(), P)) == 0
+
+
+def test_proxy_uses_correct_new_method_to_create_itself():
+
+    class A(list):
+
+        def __new__(cls):
+            counter[0] += 1
+            return [1, 2, 3]
+
+    class B(A):
+
+        def __new__(cls):
+            return Proxy(SimpleBox(), A)
+
+    class C(B):
+
+        def __new__(cls):
+            return Proxy(MultiBox(lambda: dispatcher.get()), B)
+
+    counter = [0]
+    dispatcher = SimpleBox()
+    c = C()
+    dispatcher.put(1)
+    assert len(c) == 3
+    assert counter[0] == 1
+    dispatcher.put(2)
+    assert len(c) == 3
+    assert counter[0] == 2
+    dispatcher.put(1)
+    assert len(c) == 3
+    assert counter[0] == 2