Add yarn main program

1 files changed, 182 insertions, 0 deletions

diff --git a/yarn b/yarn
new file mode 100755
index 0000000..94d2aae
--- /dev/null
+++ b/yarn
@@ -0,0 +1,182 @@
+#!/usr/bin/python
+# Copyright 2013  Lars Wirzenius
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# =*= License: GPL-3+ =*=
+
+
+import cliapp
+import logging
+import markdown
+import os
+import re
+import shutil
+import StringIO
+import sys
+import tempfile
+import time
+import ttystatus
+from markdown.treeprocessors import Treeprocessor
+
+import yarnlib
+
+
+class StoryTestRunner(cliapp.Application):
+
+    def setup(self):
+        self.ts = ttystatus.TerminalStatus(period=0.001)
+        self.ts.format(
+            '%ElapsedTime() %Index(story,stories): %String(story_name): '
+            'step %Index(step,steps): %String(step_name)')
+
+    def process_args(self, args):
+        blocks = []
+        for filename in args:
+            blocks.extend(self.parse_story_test_code_blocks(filename))
+
+        stories, implementations = self.create_objects_from_code_blocks(blocks)
+        self.connect_implementations(stories, implementations)
+
+        self.ts['stories'] = stories
+        self.ts['num_stories'] = len(stories)
+        logging.info('Found %d stories' % len(stories))
+
+        start_time = time.time()
+        failed_stories = []
+        for story in stories:
+            if not self.run_story(story):
+                failed_stories.append(story)
+
+        self.ts.clear()
+        self.ts.finish()
+
+        if failed_stories:
+            raise cliapp.AppException(
+                'Test suite FAILED in %s stories' % len(failed_stories))
+
+        duration = time.time() - start_time
+        print ('Story test suite PASS, with %d stories, in %.1f seconds' %
+            (len(stories), duration))
+
+    def parse_story_test_code_blocks(self, filename):
+        logging.info('Parsing story test file %s' % filename)
+        ext = ParseStoryTestBlocks()
+        f = StringIO.StringIO()
+        markdown.markdownFromFile(filename, output=f, extensions=[ext])
+        return ext.blocks
+
+    def create_objects_from_code_blocks(self, blocks):
+        cp = CodeBlockParser()
+        cp.parse(blocks)
+        return cp.stories, cp.implementations
+
+    def connect_implementations(self, stories, implementations):
+        for story in stories:
+            for step in story.steps:
+                self.connect_implementation(story, step, implementations)
+
+    def connect_implementation(self, story, step, implementations):
+        matching = [i for i in implementations
+                    if step.what == i.what and
+                       re.match('(%s)$' % i.regexp, step.text, re.I)]
+
+        if len(matching) == 0:
+            raise StoryTestSyntaxError(
+                'Error: Story %s, step "%s %s" has no matching '
+                'implementation' %
+                (story.name, step.what, step.text))
+        if len(matching) > 1:
+            s = '\n'.join(
+                'IMPLEMENTS %s %s' % (i.what, i.regexp)
+                for i in matching)
+            raise StoryTestSyntaxError(
+                'Story "%s", step "%s %s" has more than one '
+                'matching implementations:\n%s' %
+                (story.name, step.what, step.text, s))
+
+        assert step.implementation is None
+        step.implementation = matching[0]
+
+    def run_story(self, story):
+        logging.info('Running story %s' % story.name)
+        self.ts['story'] = story
+        self.ts['story_name'] = story.name
+        self.ts['steps'] = story.steps
+
+        datadir = tempfile.mkdtemp()
+
+        cleanup = [s for s in story.steps if s.what == 'FINALLY']
+        normal = [s for s in story.steps if s not in cleanup]
+
+        ok = True
+
+        for step in normal:
+            exit = self.run_step(datadir, story, step)
+            if exit != 0:
+                ok = False
+                break
+
+        for step in cleanup:
+            exit = self.run_step(datadir, story, step)
+            if exit != 0:
+                ok = False
+                break
+
+        shutil.rmtree(datadir)
+
+        return ok
+
+    def run_step(self, datadir, story, step):
+        logging.info('Running step "%s %s"' % (step.what, step.text))
+        logging.info('DATADIR is %s' % datadir)
+        self.ts['step'] = step
+        self.ts['step_name'] = '%s %s' % (step.what, step.text)
+
+        m = re.match(step.implementation.regexp, step.text)
+        assert m is not None
+        env = os.environ.copy()
+        env['DATADIR'] = datadir
+        for i, match in enumerate(m.groups('')):
+            env['MATCH_%d' % (i+1)] = match
+
+        exit, stdout, stderr = cliapp.runcmd_unchecked(
+            ['sh', '-euc', step.implementation.shell], env=env)
+
+        logging.debug('Exit code: %d' % exit)
+        if stdout:
+            logging.debug('Standard output:\n%s' % self.indent(stdout))
+        else:
+            logging.debug('Standard output: empty')
+        if stderr:
+            logging.debug('Standard error:\n%s' % self.indent(stderr))
+        else:
+            logging.debug('Standard error: empty')
+
+        if exit != 0:
+            self.ts.error(
+                'ERROR: In story "%s"\nstep "%s %s" failed,\n'
+                'with exit code %d:\n'
+                'Standard output from shell command:\n%s'
+                'Standard error from shell command:\n%s' %
+                (story.name, step.what, step.text, exit, self.indent(stdout),
+                 self.indent(stderr)))
+
+        return exit
+
+    def indent(self, s):
+        return ''.join('    %s\n' % line for line in s.splitlines())
+
+
+StoryTestRunner().run()