[pypy-svn] r17882 - in pypy/dist/pypy/translator/tool: . test

[pedronis at codespeak.net]

Author: pedronis
Date: Mon Sep 26 21:24:03 2005
New Revision: 17882

Added:
   pypy/dist/pypy/translator/tool/test/test_taskengine.py   (contents, props changed)
Modified:
   pypy/dist/pypy/translator/tool/taskengine.py
Log:
some more work on taskengine, proper test file



Modified: pypy/dist/pypy/translator/tool/taskengine.py
==============================================================================
--- pypy/dist/pypy/translator/tool/taskengine.py	(original)
+++ pypy/dist/pypy/translator/tool/taskengine.py	Mon Sep 26 21:24:03 2005
@@ -1,11 +1,9 @@
 
 
-
-
 class SimpleTaskEngine:
 
-
     def __init__(self):
+        self._plan_cache = {}
 
         self.tasks = tasks = {}
 
@@ -18,9 +16,14 @@
 
                 tasks[task_name] = task, task_deps
 
+    def _plan(self, goals, skip=[]):
+        skip = [toskip for toskip in skip if toskip not in goals]
 
-    def _plan(self, goal, skip=[]):
-
+        key = (tuple(goals), tuple(skip))
+        try:
+            return self._plan_cache[key]
+        except KeyError:
+            pass
         constraints = []
 
         def subgoals(task_name):
@@ -32,7 +35,6 @@
                         continue
                 yield dep
 
-
         seen = {}
                         
         def consider(subgoal):
@@ -46,7 +48,8 @@
                 constraints.append([subgoal, dep])
                 consider(dep)
 
-        consider(goal)
+        for goal in goals:
+            consider(goal)
 
         #sort
 
@@ -73,34 +76,33 @@
 
         plan.reverse()
 
-        return plan
-            
-            
-
-def test_simple():
-
-    class ABC(SimpleTaskEngine):
+        self._plan_cache[key] = plan
 
-        def task_A(self):
-            pass
-
-        task_A.task_deps = ['B', '?C']
-
-        def task_B(self):
-            pass
-
-        def task_C(self):
-            pass
+        return plan
 
-        task_C.task_deps = ['B']
+    def _execute(self, goals, *args, **kwds):
+        task_skip = kwds.get('task_skip', [])
+        for goal in self._plan(goals, skip=task_skip):
+            taskcallable, _ = self.tasks[goal]
+            self._event('pre', goal, taskcallable)
+            try:
+                self._do(goal, taskcallable, *args, **kwds)
+            except (SystemExit, KeyboardInterrupt):
+                raise
+            except:
+                self._error(goal)
+                raise
+            self._event('post', goal, taskcallable)
+        
+    def _do(self, goal, func, *args, **kwds):
+        func()
 
-    abc = ABC()
+    def _event(self, kind, goal, func):
+        pass
+    
+    def _error(self, goal):
+        pass
 
-    assert abc._plan('B') == ['B']
-    assert abc._plan('C') == ['B', 'C']
-    assert abc._plan('A') == ['B', 'C', 'A']
-    assert abc._plan('A', skip=['C']) == ['B', 'A']
-   
 
 """ sketch of tasks for translation:
 

Added: pypy/dist/pypy/translator/tool/test/test_taskengine.py
==============================================================================
--- (empty file)
+++ pypy/dist/pypy/translator/tool/test/test_taskengine.py	Mon Sep 26 21:24:03 2005
@@ -0,0 +1,73 @@
+from pypy.translator.tool.taskengine import SimpleTaskEngine
+
+def test_simple():
+
+    class ABC(SimpleTaskEngine):
+
+        def task_A(self):
+            pass
+
+        task_A.task_deps = ['B', '?C']
+
+        def task_B(self):
+            pass
+
+        def task_C(self):
+            pass
+
+        task_C.task_deps = ['B']
+
+    abc = ABC()
+
+    assert abc._plan('B') == ['B']
+    assert abc._plan('C') == ['B', 'C']
+    assert abc._plan('A') == ['B', 'C', 'A']
+    assert abc._plan('A', skip=['C']) == ['B', 'A']
+   
+def test_execute():
+
+    class ABC(SimpleTaskEngine):
+
+        def __init__(self):
+            SimpleTaskEngine.__init__(self)
+            self.done = []
+
+        def task_A(self):
+            self.done.append('A')
+
+        task_A.task_deps = ['B', '?C']
+
+        def task_B(self):
+            self.done.append('B')
+
+        def task_C(self):
+            self.done.append('C')
+
+        task_C.task_deps = ['B']
+
+        def _event(self, kind, goal, taskcallable):
+            self.done.append((kind, goal))
+
+    def test(goals, task_skip=[]):
+        if isinstance(goals, str):
+            gaols = [goals]
+        abc = ABC()
+        abc._execute(goals, task_skip=task_skip)
+        return abc.done
+
+    def trace(goals):
+        t = []
+        for goal in goals:
+            t.extend([('pre', goal), goal, ('post', goal)])
+        return t
+
+    assert test('B') == trace('B')
+    assert test('C') == trace(['B', 'C'])
+    assert test('A') == trace(['B', 'C', 'A'])
+    assert test('A', ['C']) == trace(['B', 'A'])
+    assert test(['B', 'C']) == trace(['B', 'C'])
+    assert test(['C', 'B']) == trace(['B', 'C'])
+    assert test(['B', 'A']) == trace(['B', 'C', 'A'])
+    assert test(['B', 'A'], ['C']) == trace(['B', 'A'])
+    assert test(['B', 'A', 'C']) == trace(['B', 'C', 'A'])
+    assert test(['B', 'A', 'C'], ['C']) == trace(['B', 'C', 'A'])