Fix #5: Move json parsing and error handling into the event factory

Pass the raw string into the event factory and parse it to a json
object there.  If the string is not valid json, generate an error
event.

Add a unit test for handling of invalid json on the event stream.

Change-Id: I209a89fd28c3a594b71443fc106e25d58c5cc1ea

6 files changed, 59 insertions, 36 deletions

diff --git a/example.py b/example.py
index d919857..4a836d1 100755
--- a/example.py
+++ b/example.py
@@ -33,7 +33,7 @@ import time
 
 from pygerrit.client import GerritClient
 from pygerrit.error import GerritError
-from pygerrit.stream import GerritStreamErrorEvent
+from pygerrit.events import ErrorEvent
 
 
 def _main():
@@ -77,7 +77,7 @@ def _main():
                                      timeout=options.timeout)
             if event:
                 logging.info("Event: %s", event)
-                if isinstance(event, GerritStreamErrorEvent):
+                if isinstance(event, ErrorEvent):
                     logging.error(event.error)
                     errors.set()
                     break
diff --git a/pygerrit/client.py b/pygerrit/client.py
index b6502e1..3c9de68 100644
--- a/pygerrit/client.py
+++ b/pygerrit/client.py
@@ -112,15 +112,15 @@ class GerritClient(object):
         except Empty:
             return None
 
-    def put_event(self, json_data):
-        """ Create event from `json_data` and add it to the queue.
+    def put_event(self, data):
+        """ Create event from `data` and add it to the queue.
 
         Raise GerritError if the queue is full, or the factory could not
         create the event.
 
         """
         try:
-            event = self._factory.create(json_data)
+            event = self._factory.create(data)
             self._events.put(event)
         except Full:
             raise GerritError("Unable to add event: queue is full")
diff --git a/pygerrit/events.py b/pygerrit/events.py
index 8cf44cd..e91fbf4 100644
--- a/pygerrit/events.py
+++ b/pygerrit/events.py
@@ -23,6 +23,9 @@
 
 """ Gerrit event classes. """
 
+import json
+import logging
+
 from .error import GerritError
 from .models import Account, Approval, Change, Patchset, RefUpdate
 
@@ -53,13 +56,22 @@ class GerritEventFactory(object):
         return decorate
 
     @classmethod
-    def create(cls, json_data):
+    def create(cls, data):
         """ Create a new event instance.
 
-        Return an instance of the `GerritEvent` subclass from `json_data`
-        Raise GerritError if `json_data` does not contain a `type` key.
+        Return an instance of the `GerritEvent` subclass after converting
+        `data` to json.
+
+        Raise GerritError if json parsed from `data` does not contain a `type`
+        key.
 
         """
+        try:
+            json_data = json.loads(data)
+        except ValueError as err:
+            logging.debug("Failed to load json data: %s: [%s]", str(err), data)
+            json_data = ErrorEvent.error_json(str(err))
+
         if not "type" in json_data:
             raise GerritError("`type` not in json_data")
         name = json_data["type"]
@@ -90,6 +102,22 @@ class UnhandledEvent(GerritEvent):
         super(UnhandledEvent, self).__init__(json_data)
 
 
+@GerritEventFactory.register("error-event")
+class ErrorEvent(GerritEvent):
+
+    """ Error occurred when processing json data from Gerrit's event stream. """
+
+    def __init__(self, json_data):
+        super(ErrorEvent, self).__init__(json_data)
+
+    @classmethod
+    def error_json(cls, error):
+        """ Return a json string for the `error`. """
+        data = '{"type":"error-event",' \
+               '"error":"%s"}' % str(error)
+        return json.loads(data)
+
+
 @GerritEventFactory.register("patchset-created")
 class PatchsetCreatedEvent(GerritEvent):
 
diff --git a/pygerrit/stream.py b/pygerrit/stream.py
index a9f94ae..dcf57ba 100644
--- a/pygerrit/stream.py
+++ b/pygerrit/stream.py
@@ -26,23 +26,12 @@ Class to listen to the Gerrit event stream and dispatch events.
 
 """
 
-import json
 import logging
 from select import select
 from threading import Thread, Event
 
 from .error import GerritError
-from .events import GerritEvent, GerritEventFactory
-
-
-@GerritEventFactory.register("gerrit-stream-error")
-class GerritStreamErrorEvent(GerritEvent):
-
-    """ Represents an error when handling the gerrit event stream. """
-
-    def __init__(self, json_data):
-        super(GerritStreamErrorEvent, self).__init__()
-        self.error = json_data["error"]
+from .events import ErrorEvent
 
 
 class GerritStream(Thread):
@@ -62,16 +51,14 @@ class GerritStream(Thread):
 
     def _error_event(self, error):
         """ Dispatch `error` to the Gerrit client. """
-        json_data = json.loads('{"type":"gerrit-stream-error",'
-                               '"error":"%s"}' % str(error))
-        self._gerrit.put_event(json_data)
+        self._gerrit.put_event(ErrorEvent.error_json(str(error)))
 
     def run(self):
         """ Listen to the stream and send events to the client. """
         try:
             result = self._ssh_client.run_gerrit_command("stream-events")
-        except GerritError as e:
-            self._error_event(e)
+        except GerritError as err:
+            self._error_event(err)
         else:
             stdout = result.stdout
             inputready, _outputready, _exceptready = \
@@ -79,10 +66,8 @@ class GerritStream(Thread):
             while not self._stop.is_set():
                 for _event in inputready:
                     try:
-                        line = stdout.readline()
-                        json_data = json.loads(line)
-                        self._gerrit.put_event(json_data)
-                    except (ValueError, IOError) as err:
+                        self._gerrit.put_event(stdout.readline())
+                    except IOError as err:
                         self._error_event(err)
                     except GerritError as err:
                         logging.error("Failed to put event: %s", err)
diff --git a/testdata/invalid-json.txt b/testdata/invalid-json.txt
new file mode 100644
index 0000000..f7a60bb
--- /dev/null
+++ b/testdata/invalid-json.txt
@@ -0,0 +1,4 @@
+)]}'
+{"type":"user-defined-event",
+ "title":"Event title",
+ "description":"Event description"}
diff --git a/unittests.py b/unittests.py
index 358d7f4..1cf617b 100755
--- a/unittests.py
+++ b/unittests.py
@@ -32,7 +32,8 @@ import unittest
 from pygerrit.events import PatchsetCreatedEvent, \
     RefUpdatedEvent, ChangeMergedEvent, CommentAddedEvent, \
     ChangeAbandonedEvent, ChangeRestoredEvent, \
-    DraftPublishedEvent, GerritEventFactory, GerritEvent, UnhandledEvent
+    DraftPublishedEvent, GerritEventFactory, GerritEvent, UnhandledEvent, \
+    ErrorEvent
 from pygerrit.client import GerritClient
 from setup import REQUIRES as setup_requires
 
@@ -55,10 +56,10 @@ def _create_event(name, gerrit):
     data, then add as an event in the `gerrit` client.
 
     """
-    data = open(os.path.join("testdata", name + ".txt"))
-    json_data = json.loads(data.read().replace("\n", ""))
-    gerrit.put_event(json_data)
-    return json_data
+    testfile = open(os.path.join("testdata", name + ".txt"))
+    data = testfile.read().replace("\n", "")
+    gerrit.put_event(data)
+    return data
 
 
 class TestConsistentDependencies(unittest.TestCase):
@@ -249,10 +250,15 @@ class TestGerritEvents(unittest.TestCase):
         self.assertEquals(event.description, "Event description")
 
     def test_unhandled_event(self):
-        json_data = _create_event("unhandled-event", self.gerrit)
+        data = _create_event("unhandled-event", self.gerrit)
         event = self.gerrit.get_event(False)
         self.assertTrue(isinstance(event, UnhandledEvent))
-        self.assertEquals(event.json, json_data)
+        self.assertEquals(event.json, json.loads(data))
+
+    def test_invalid_json(self):
+        _create_event("invalid-json", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, ErrorEvent))
 
     def test_add_duplicate_event(self):
         try: