Fix filter server on dnszone DataStream.

core/datastream/base.py

@@ -399,6 +399,10 @@ class DataStream(object):
             if isinstance(meta[field], str):
                 # Old meta format
                 return False
+            if "$elemMatch" in field_value:
+                field_value = field_value["$elemMatch"]["$elemMatch"]["$in"]
+            if isinstance(meta[field][0], list) and isinstance(field_value, list):
+                return not set(field_value).intersection(set(meta[field][0]))
             if meta[field][0] != field_value:
                 return True
         return False

services/datastream/streams/dnszone.py

@@ -400,4 +400,4 @@ class DNSZoneDataStream(DataStream):
 
     @classmethod
     def filter_server(cls, name):
-        return {"%s.servers" % cls.F_META: name}
+        return {f"{cls.F_META}.servers": {"$elemMatch": {"$elemMatch": {"$in": [name]}}}}

services/datastream/streams/managedobject.py

@@ -475,11 +475,11 @@ class ManagedObjectDataStream(DataStream):
 
     @classmethod
     def filter_service_group(cls, name: str):
-        return {"%s.service_groups" % cls.F_META: name}
+        return {f"{cls.F_META}.service_groups": {"$elemMatch": {"$elemMatch": {"$in": [name]}}}}
 
     @classmethod
     def filter_client_group(cls, name: str):
-        return {"%s.client_groups" % cls.F_META: name}
+        return {f"{cls.F_META}.client_groups": {"$elemMatch": {"$elemMatch": {"$in": [name]}}}}
 
     @classmethod
     def get_msg_headers(cls, data: Dict[str, Any]) -> Optional[Dict[str, bytes]]:

tests/datastreams/test_clean_meta.py

@@ -74,5 +74,39 @@ from noc.core.datastream.base import DataStream
         ),
     ],
 )
-def test_parse_table(meta, current_meta, expected):
+def test_ds_clean_meta(meta, current_meta, expected):
     assert DataStream.clean_meta(meta, current_meta) == expected
+
+
+@pytest.mark.parametrize(
+    "meta,meta_filter,expected",
+    [
+        (
+            {"servers": [["ns1.example.com", "ns2.example.com"]]},
+            {"meta.servers": {"$elemMatch": {"$elemMatch": {"$in": ["ns1.example.com"]}}}},
+            False,
+        ),
+        (
+            {
+                "servers": [
+                    ["ns3.example.com", "ns2.example.com"],
+                    ["ns1.example.com", "ns2.example.com"],
+                ]
+            },
+            {"meta.servers": {"$elemMatch": {"$elemMatch": {"$in": ["ns1.example.com"]}}}},
+            True,
+        ),
+        (
+            {"pool": ["POOL1", "POOL2"]},
+            {"meta.pool": "POOL2"},
+            True,
+        ),
+        (
+            {"pool": ["POOL1", "POOL2"]},
+            {"meta.pool": "POOL1"},
+            False,
+        ),
+    ],
+)
+def test_ds_is_moved(meta, meta_filter, expected):
+    assert DataStream.is_moved(meta, meta_filter) == expected

tests/datastreams/test_dnszone.py

@@ -111,4 +111,4 @@ def test_filter_server():
     q = DNSZoneDataStream.filter_server("ns1.example.com")
     assert q
     assert "meta.servers" in q
-    assert q["meta.servers"] == "ns1.example.com"
+    assert q["meta.servers"] == {"$elemMatch": {"$elemMatch": {"$in": ["ns1.example.com"]}}}