Add unit test

octopoes/nibbles/definitions.py

@@ -52,11 +52,11 @@ def __init__(
         self.default_enabled = default_enabled
         self.config_ooi_relation_path = config_ooi_relation_path
 
-    def __call__(self, *_):
+    def __call__(self, args):
         if self.payload is None:
             raise NotImplementedError
         else:
-            return self.payload(*_)
+            return self.payload(*args)
 
 
 def get_nibble_definitions() -> list[NibbleDefinition]:

octopoes/nibbles/runner.py

@@ -52,10 +52,7 @@ def _run(self, ooi: OOI, valid_time: datetime) -> dict[str, set[OOI]]:
             # TODO: filter OOI not abiding the parameters from radix
             radix = [self.objects_by_type_cache[sgn.ooi_type] for sgn in nibble.signature]
             results = set(
-                filter(
-                    lambda ooi: ooi is not None,
-                    chain(map(lambda x: nibble(*x), filter(lambda x: ooi in x, product(*radix)))),
-                )
+                filter(lambda ooi: ooi is not None, chain(map(nibble, filter(lambda x: ooi in x, product(*radix)))))
             )
             if results:
                 retval |= {nibble.id: results}

octopoes/tests/integration/test_nibbles.py

@@ -0,0 +1,45 @@
+import os
+import sys
+from datetime import datetime
+from unittest.mock import Mock
+
+import pytest
+from nibbles.definitions import NibbleDefinition, NibbleParameterDefinition
+
+from octopoes.core.service import OctopoesService
+from octopoes.models import OOI, ScanLevel
+from octopoes.models.ooi.network import Network
+
+if os.environ.get("CI") != "1":
+    pytest.skip("Needs XTDB multinode container.", allow_module_level=True)
+
+dummy_nibble = NibbleDefinition(name="dummy", signature=[NibbleParameterDefinition(ooi_type=Network)])
+
+
+def dummy(network: Network):
+    if len(network.name) < 27:
+        new_name = network.name + "I"
+        return Network(name=new_name)
+
+
+dummy_nibble.payload = getattr(sys.modules[__name__], "dummy")
+
+
+def test_unicode_network(xtdb_octopoes_service: OctopoesService, event_manager: Mock, valid_time: datetime):
+    xtdb_octopoes_service.nibbles.nibbles = [dummy_nibble]
+    network = Network(name="internet")
+    xtdb_octopoes_service.ooi_repository.save(network, valid_time)
+    event_manager.complete_process_events(xtdb_octopoes_service)
+
+    assert xtdb_octopoes_service.ooi_repository.list_oois({Network}, valid_time).count == 1
+    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 3
+
+    sp = xtdb_octopoes_service.scan_profile_repository.get(network.reference, valid_time)
+    new_sp = sp.model_copy()
+    new_sp.level = ScanLevel.L2
+    xtdb_octopoes_service.scan_profile_repository.save(sp, new_sp, valid_time)
+    event_manager.complete_process_events(xtdb_octopoes_service)
+
+    ctx = 28 - len(network.name)
+    assert xtdb_octopoes_service.ooi_repository.list_oois({Network}, valid_time).count == ctx
+    assert xtdb_octopoes_service.ooi_repository.list_oois({OOI}, valid_time).count == 3 * ctx