update ometa.tube.trampolinedparser

Author: introom

ometa/test/test_tube.py

@@ -0,0 +1,74 @@
+from __future__ import absolute_import
+
+from twisted.trial import unittest
+from twisted.python.compat import iterbytes
+
+
+from ometa.grammar import OMeta
+from ometa.tube import TrampolinedParser
+
+
+class TrampolinedReceiver():
+    """
+    Receive and store the passed in data.
+    """
+    def __init__(self):
+        self.received = []
+
+    def receive(self, data):
+        self.received.append(data)
+
+
+class TrampolinedParserTestCase(unittest.SynchronousTestCase):
+    """
+    Tests for L{ometa.tube.TrampolinedParser}
+    """
+
+    def _parseGrammar(self, grammar, name="Grammar"):
+        return OMeta(grammar).parseGrammar(name)
+
+    def setUp(self):
+        _grammar =  r"""
+            delimiter = '\r\n'
+            initial = <(~delimiter anything)*>:val delimiter -> receiver.receive(val)
+        """
+        self.grammar = self._parseGrammar(_grammar)
+
+    def test_dataNotFullyReceived(self):
+        """
+        Since the initial rule inside the grammar is not matched, the receiver
+        shouldn't receive any byte.
+        """
+        receiver = TrampolinedReceiver()
+        trampolinedParser = TrampolinedParser(self.grammar, receiver, {})
+        buf = b'foobarandnotreachdelimiter'
+        for c in iterbytes(buf):
+            trampolinedParser.receive(c)
+        self.assertEqual(receiver.received, [])
+
+
+    def test_dataFullyReceived(self):
+        """
+        The receiver should receive the data according to the grammar.
+        """
+        receiver = TrampolinedReceiver()
+        trampolinedParser = TrampolinedParser(self.grammar, receiver, {})
+        buf = b'\r\n'.join((b'foo', b'bar', b'foo', b'bar'))
+        for c in iterbytes(buf):
+            trampolinedParser.receive(c)
+        self.assertEqual(receiver.received, [b'foo', b'bar', b'foo'])
+        trampolinedParser.receive('\r\n')
+        self.assertEqual(receiver.received, [b'foo', b'bar', b'foo', b'bar'])
+
+
+    def test_bindings(self):
+        """
+        The passed-in bindings should be accessible inside the grammar.
+        """
+        receiver = TrampolinedReceiver()
+        grammar = r"""
+            initial = digit:d (-> int(d)+SMALL_INT):val -> receiver.receive(val)
+        """
+        bindings = {'SMALL_INT': 3}
+        TrampolinedParser(self._parseGrammar(grammar), receiver, bindings).receive('0')
+        self.assertEqual(receiver.received, [3])

ometa/tube.py

@@ -0,0 +1,50 @@
+from ometa.interp import TrampolinedGrammarInterpreter, _feed_me
+
+class TrampolinedParser:
+    """
+    A parser that incrementally parses incoming data.
+    """
+    def __init__(self, grammar, receiver, bindings):
+        """
+        Initializes the parser.
+
+        @param grammar: The grammar used to parse the incoming data.
+        @param receiver: Responsible for logic operation on the parsed data.
+            Typically, the logic operation will be invoked inside the grammar,
+            e.g., rule = expr1 expr2 (-> receiver.doSomeStuff())
+        @param bindings: The namespace that can be accessed inside the grammar.
+        """
+        self.grammar = grammar
+        self.bindings = dict(bindings)
+        self.bindings['receiver'] = self.receiver = receiver
+        self._setupInterp()
+
+
+    def _setupInterp(self):
+        """
+        Resets the parser. The parser will begin parsing with the rule named
+        'initial'.
+        """
+        self._interp = TrampolinedGrammarInterpreter(
+            grammar=self.grammar, ruleName='initial', callback=None,
+            globals=self.bindings)
+
+
+    def receive(self, data):
+        """
+        Receive the incoming data and begin parsing. The parser will parse the
+        data incrementally according to the 'initial' rule in the grammar.
+
+        @param data: The raw data received.
+        """
+        while data:
+            try:
+                status = self._interp.receive(data)
+            except Exception as e:
+                # maybe we should raise it?
+                raise e
+            else:
+                if status is _feed_me:
+                    return
+            data = ''.join(self._interp.input.data[self._interp.input.position:])
+            self._setupInterp()