splitio · mmelograno · Oct 20, 2020 · Oct 13, 2020 · Oct 13, 2020 · Oct 14, 2020
diff --git a/splitio/push/__init__.py b/splitio/push/__init__.py
diff --git a/splitio/push/parser.py b/splitio/push/parser.py
@@ -0,0 +1,73 @@
+import json
+
+from future.utils import raise_from
+
+MESSAGE = 'message'
+ERROR = 'error'
+OCCUPANCY = 'occupancy'
+UPDATE = 'update'
+TAG_OCCUPANCY = '[meta]occupancy'
+
+class EventParserException(Exception):
+    """Exception to be raised on parser errors."""
+
+    pass
+
+
+class AblyError(object):
+    def __init__(self, code, status_code, message, href):
+        self._code = code
+        self._status_code = status_code
+        self._message = message
+        self._event = ERROR
+        self._href = href
+
+class Occupancy(object):
+    def __init__(self, data, channel):
+        self._data = data
+        self._event = OCCUPANCY
+        self._channel = channel
+
+class Update(object):
+    def __init__(self, data, channel):
+        self._data = data
+        self._event = UPDATE
+        self._channel = channel
+
+
+def parseAblyError(parsed_data):
+    if 'statusCode' in parsed_data and 'code' in parsed_data and 'message' in parsed_data and 'href' in parsed_data:
+        return AblyError(parsed_data['code'], parsed_data['statusCode'], parsed_data['message'], parsed_data['href'])
+    return None
+
+def parseNotification(parsed_data):
+    if 'name' in parsed_data and parsed_data['name'] == TAG_OCCUPANCY and 'data' in parsed_data and 'channel' in parsed_data:
+        return Occupancy(parsed_data['data'], parsed_data['channel'])
+    elif 'data' in parsed_data and 'channel' in parsed_data:
+        return Update(parsed_data['data'], parsed_data['channel'])
+    return None
+
+_INCOMMING_EVENT_MAPPERS = {
+    ERROR: lambda d: parseAblyError(d),
+    MESSAGE: lambda d: parseNotification(d),
+}
+
+def parse_incoming_event(raw_event):
+    if raw_event is None or len(raw_event.strip()) == 0:
+        return None
+
+    try:
+        parsed_raw_event = json.loads(raw_event)
+        if parsed_raw_event is None:
+            return None
+
+        if not 'event' in parsed_raw_event or not 'data' in parsed_raw_event:
+            return None
+
+        parsed_data = json.loads(parsed_raw_event['data'])
+        mapper = _INCOMMING_EVENT_MAPPERS[parsed_raw_event['event']]
+        return mapper(parsed_data)
+    except KeyError as exc:
+        raise_from(EventParserException('No mapper registered for that event'), exc)
+    except ValueError as exc:
+        raise_from(EventParserException('Cannot parse json.'), exc)
diff --git a/tests/push/__init__.py b/tests/push/__init__.py
diff --git a/tests/push/test_parser.py b/tests/push/test_parser.py
@@ -0,0 +1,88 @@
+import json
+import pytest
+
+from splitio.push.parser import parse_incoming_event, Update, AblyError, Occupancy, EventParserException
+
+
+def wrap_json(channel, data):
+    return json.dumps({
+        'data': json.dumps({
+            'id':'ZlalwoKlXW:0:0',
+            'timestamp':1591996755043,
+            'encoding':'json',
+            'channel': channel,
+            'data': json.dumps(data)
+        }),
+        'event': 'message'
+    })
+
+class ParserTests(object):
+    """Parser tests."""
+
+    def test_exception(self):
+        """Test exceptions."""
+        assert parse_incoming_event(None) is None
+        assert parse_incoming_event('') is None
+        assert parse_incoming_event('  ') is None
+        assert parse_incoming_event(json.dumps({})) is None
+
+        with pytest.raises(EventParserException):
+            parse_incoming_event('asd')
+
+        with pytest.raises(EventParserException):
+            parse_incoming_event(json.dumps({
+            'data': json.dumps({'a':1}),
+            'event': 'some'
+        }))
+
+    def test_event_parsing(self):
+        """Test parse Update event."""
+        e0 = wrap_json(
+            'NDA5ODc2MTAyNg==_MzAyODY0NDkyOA==_splits',
+            {'type':'SPLIT_KILL','changeNumber':1591996754396,'defaultTreatment':'some','splitName':'test'},
+        )
+        assert isinstance(parse_incoming_event(e0), Update)
+
+        e1 = wrap_json(
+            'NDA5ODc2MTAyNg==_MzAyODY0NDkyOA==_splits',
+            {'type':'SPLIT_UPDATE','changeNumber':1591996685190},
+        )
+        assert isinstance(parse_incoming_event(e1), Update)
+
+        e2 = wrap_json(
+            'NDA5ODc2MTAyNg==_MzAyODY0NDkyOA==_segments',
+            {'type':'SEGMENT_UPDATE','changeNumber':1591988398533,'segmentName':'some'},
+        )
+        assert isinstance(parse_incoming_event(e2), Update)
+
+    def test_error_parsing(self):
+        """Test parse AblyError event."""
+        e0 = json.dumps({
+            'data': json.dumps({
+                'code': 40142,
+                'message': 'Token expired',
+                'statusCode': 401,
+                'href': 'https://help.io/error/40142',
+            }),
+            'event': 'error'
+        })
+        assert isinstance(parse_incoming_event(e0), AblyError)
+
+    def test_occupancy_parsing(self):
+        """Test parse Occupancy event."""
+        e0 = json.dumps({
+            'data': json.dumps({
+                'id':'ZlalwoKlXW:0:0',
+                'timestamp':1591996755043,
+                'encoding':'json',
+                'channel': '[?occupancy=metrics.publishers]control_sec',
+                'data': json.dumps({
+                    'metrics': json.dumps({
+                        'publishers': 1
+                    }),
+                }),
+                'name': '[meta]occupancy',
+            }),
+            'event': 'message'
+        })
+        assert isinstance(parse_incoming_event(e0), Occupancy)