make interface classification more testable

inventory_provider/tasks/worker.py

@@ -698,30 +698,31 @@ class PollerServiceCategory(str, enum.Enum):
     BACKBONE = 'backbone'
 
 
+def _classify_interface(ifc):
+    if ifc['description'].startswith('SRV_MDVPN CUSTOMER'):
+        yield PollerServiceCategory.MDVPN
+    if 'LHCONE' in ifc['description'] \
+            and 'SRV_L3VPN CUSTOMER' in ifc['description']:
+        yield PollerServiceCategory.LHCONE_CUST
+    if 'LHCONE' in ifc['description'] \
+            and 'SRV_L3VPN RE' in ifc['description']:
+        yield PollerServiceCategory.LHCONE_PEER
+    if 'SRV_L2CIRCUIT' in ifc['description'] \
+            and 'SRV_L3VPN' in ifc['description']:
+        yield PollerServiceCategory.L2_Circuits
+    if 'PHY CUSTOMER' in ifc['description'] \
+            and 'LAG CUSTOMER' in ifc['description'] \
+            and 'SRV_GLOBAL CUSTOMER' in ifc['description']:
+        yield PollerServiceCategory.RE_CUST
+    if 'SRV_GLOBAL RE_INTERCONNECT' in ifc['description']:
+        yield PollerServiceCategory.RE_PEER
+    if 'SRV_IAS CUSTOMER' in ifc['description']:
+        yield PollerServiceCategory.IAS
+
+
 @log_task_entry_and_exit
 def _build_service_category_interface_list(update_callback=lambda s: None):
 
-    def _classify(ifc):
-        if ifc['description'].startswith('SRV_MDVPN CUSTOMER'):
-            yield PollerServiceCategory.MDVPN
-        if 'LHCONE' in ifc['description'] \
-                and 'SRV_L3VPN CUSTOMER' in ifc['description']:
-            yield PollerServiceCategory.LHCONE_CUST
-        if 'LHCONE' in ifc['description'] \
-                and 'SRV_L3VPN RE' in ifc['description']:
-            yield PollerServiceCategory.LHCONE_PEER
-        if 'SRV_L2CIRCUIT' in ifc['description'] \
-                and 'SRV_L3VPN' in ifc['description']:
-            yield PollerServiceCategory.L2_Circuits
-        if 'PHY CUSTOMER' in ifc['description'] \
-                and 'LAG CUSTOMER' in ifc['description'] \
-                and 'SRV_GLOBAL CUSTOMER' in ifc['description']:
-            yield PollerServiceCategory.RE_CUST
-        if 'SRV_GLOBAL RE_INTERCONNECT' in ifc['description']:
-            yield PollerServiceCategory.RE_PEER
-        if 'SRV_IAS CUSTOMER' in ifc['description']:
-            yield PollerServiceCategory.IAS
-
     update_callback('loading all known interfaces')
     interfaces = data.build_service_interface_user_list(InventoryTask.config)
     interfaces = list(interfaces)
@@ -732,7 +733,7 @@ def _build_service_category_interface_list(update_callback=lambda s: None):
     rp = r.pipeline()
 
     for ifc in interfaces:
-        for service_category in _classify(ifc):
+        for service_category in _classify_interface(ifc):
             rp.set(
                 f'interface-services:{service_category.value}'
                 f':{ifc["router"]}:{ifc["interface"]}',

test/test_worker_utils.py

@@ -5,6 +5,7 @@ import json
 import re
 
 import jsonschema
+import pytest
 
 from inventory_provider.tasks import worker
 from inventory_provider.tasks import common
@@ -118,3 +119,42 @@ def test_build_subnet_db(mocked_worker_module):
         assert value['name'] == address
 
     assert found_record
+
+
+# TODO: more tests
+# TODO: then share with neteng to make sure the test data is sensible
+# this data is just enough to be an example
+#   ... seems the method is not yet ready for testing
+@pytest.mark.parametrize('description,expected_categories', [
+    [
+        'SRV_L3VPN CUSTOMER ESNET LHCONE SRF9928635 | ASN293 | GEN-EEX-ESNET-LHCONE',  # noqa
+        {
+            worker.PollerServiceCategory.LHCONE_CUST
+        }
+    ],
+    [
+        'SRV_MDVPN CUSTOMER SANET SRF9939441 | VPN-Proxy to NREN CE',
+        {
+            worker.PollerServiceCategory.MDVPN
+        }
+    ]
+])
+def test_interface_classification(description, expected_categories):
+    categories = worker._classify_interface({'description': description})
+    assert set(list(categories)) == expected_categories
+
+
+# def _all_interfaces():
+#     import redis
+#     r = redis.StrictRedis(host='test-dashboard-storage01.geant.org')
+#     for k in r.scan_iter('netconf-interfaces:*'):
+#         k = k.decode('utf-8')
+#         ifc = json.loads(r.get(k).decode('utf-8'))
+#         fields = k.split(':')
+#         ifc['router'] = fields[1]
+#         yield ifc
+#
+#
+# def test_123():
+#     with open('interfaces.json', 'w') as f:
+#         f.write(json.dumps(list(_all_interfaces())))