Added CNML support

netdiff/__init__.py

@@ -3,4 +3,5 @@
 from .parsers.olsr import OlsrParser  # noqa
 from .parsers.batman import BatmanParser  # noqa
 from .parsers.netjson import NetJsonParser  # noqa
+from .parsers.cnml import CnmlParser  # noqa
 from .utils import diff  # noqa

netdiff/parsers/cnml.py

@@ -0,0 +1,51 @@
+import networkx
+import libcnml
+import os
+import six
+
+try:
+    import urlparse
+except ImportError:
+    import urllib.parse as urlparse
+
+from .base import BaseParser
+from ..exceptions import NetParserException
+
+
+class CnmlParser(BaseParser):
+    """ CNML 0.1 parser """
+    protocol = 'static'
+    version = '0.1'
+    metric = None
+
+    def _to_python(self, data):
+        if isinstance(data, six.string_types):
+            up = urlparse.urlparse(data)
+            # if it looks like a file path
+            if os.path.isfile(data) or up.scheme in ['http', 'https']:
+                return libcnml.CNMLParser(data)
+            else:
+                raise NetParserException('Could not decode CNML data')
+        elif isinstance(data, libcnml.CNMLParser):
+            return data
+        else:
+            raise NetParserException('Could not find valid data to parse')
+
+    def parse(self, data):
+        """
+        Extract information from a CNML file to generate a NetworkX Graph object.
+        """
+        graph = networkx.Graph()
+
+        # loop over links and create networkx graph
+        # Add only working nodes with working links
+        for link in data.get_inner_links():
+            if link.status != libcnml.libcnml.Status.WORKING:
+                continue
+            interface_a, interface_b = link.getLinkedInterfaces()
+            source = interface_a.ipv4
+            dest = interface_b.ipv4
+            # add link to Graph
+            graph.add_edge(source, dest, weight=1)
+
+        self.graph = graph

requirements.txt

@@ -1,3 +1,4 @@
 networkx
 requests
 six
+libcnml

tests/cnml/__init__.py

@@ -0,0 +1 @@
+from .tests import TestCnmlParser  # noqa

tests/cnml/tests.py

@@ -0,0 +1,119 @@
+import os
+import six
+import networkx
+
+from netdiff import CnmlParser
+from netdiff import diff
+from netdiff.exceptions import NetParserException
+from netdiff.tests import TestCase
+
+
+__all__ = ['TestCnmlParser']
+
+
+CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
+cnml1 = '{0}/../static/26494_detail_1.cnml'.format(CURRENT_DIR)
+cnml2 = '{0}/../static/26494_detail_2.cnml'.format(CURRENT_DIR)
+cnml3 = '{0}/../static/26494_detail_3.cnml'.format(CURRENT_DIR)
+
+
+class TestCnmlParser(TestCase):
+
+    def test_parse(self):
+        p = CnmlParser(cnml1)
+        self.assertIsInstance(p.graph, networkx.Graph)
+
+    def test_parse_exception(self):
+        with self.assertRaises(NetParserException):
+            CnmlParser('{ "test": "test" }')
+
+    def test_parse_exception2(self):
+        with self.assertRaises(NetParserException):
+            CnmlParser('telnet://127.0.0.1:9090')
+
+    def test_json_dict(self):
+        p = CnmlParser(cnml1)
+        data = p.json(dict=True)
+        self.assertIsInstance(data, dict)
+        self.assertEqual(data['type'], 'NetworkGraph')
+        self.assertEqual(data['protocol'], 'static')
+        self.assertEqual(data['version'], '0.1')
+        self.assertEqual(data['revision'], None)
+        self.assertEqual(data['metric'], None)
+        self.assertIsInstance(data['nodes'], list)
+        self.assertIsInstance(data['links'], list)
+        self.assertEqual(len(data['nodes']), 5)
+        self.assertEqual(len(data['links']), 3)
+
+    def test_json_string(self):
+        p = CnmlParser(cnml1)
+        data = p.json()
+        self.assertIsInstance(data, six.string_types)
+        self.assertIn('NetworkGraph', data)
+        self.assertIn('protocol', data)
+        self.assertIn('version', data)
+        self.assertIn('revision', data)
+        self.assertIn('metric', data)
+        self.assertIn('0.1', data)
+        self.assertIn('links', data)
+        self.assertIn('nodes', data)
+
+    def test_no_changes(self):
+        old = CnmlParser(cnml1)
+        new = CnmlParser(cnml1)
+        result = diff(old, new)
+        self.assertTrue(type(result) is dict)
+        self.assertTrue(type(result['added']) is list)
+        self.assertTrue(type(result['removed']) is list)
+        # ensure there are no differences
+        self.assertEqual(len(result['added']), 0)
+        self.assertEqual(len(result['removed']), 0)
+
+    def test_added_1_link(self):
+        old = CnmlParser(cnml1)
+        new = CnmlParser(cnml2)
+        result = diff(old, new)
+        # ensure there are differences
+        self.assertEqual(len(result['added']), 1)
+        self.assertEqual(len(result['removed']), 0)
+        # ensure 1 link added
+        self.assertIn('10.228.172.97', result['added'][0])
+        self.assertIn('10.228.172.101', result['added'][0])
+
+    def test_removed_1_link(self):
+        old = CnmlParser(cnml2)
+        new = CnmlParser(cnml1)
+        result = diff(old, new)
+        self.assertTrue(type(result) is dict)
+        self.assertTrue(type(result['added']) is list)
+        self.assertTrue(type(result['removed']) is list)
+        # ensure there are differences
+        self.assertEqual(len(result['added']), 0)
+        self.assertEqual(len(result['removed']), 1)
+        # ensure 1 link removed
+        self.assertIn('10.228.172.97', result['removed'][0])
+        self.assertIn('10.228.172.101', result['removed'][0])
+
+    def test_simple_diff(self):
+        old = CnmlParser(cnml1)
+        new = CnmlParser(cnml3)
+        result = diff(old, new)
+        # ensure there are differences
+        self.assertEqual(len(result['added']), 2)
+        self.assertEqual(len(result['removed']), 2)
+        # ensure 2 links added
+        self._test_expected_links(
+            links=result['added'],
+            expected_links=[
+                ('10.228.172.97', '10.228.172.101'),
+                ('10.228.172.194', '10.228.172.193'),
+            ]
+        )
+        # ensure 2 links removed
+        self._test_expected_links(
+            links=result['removed'],
+            expected_links=[
+                ('10.228.172.33', '10.228.172.34'),
+                ('10.228.172.33', '10.228.172.36'),
+            ]
+        )