Moar testing stuff…

DeltaMatcher.py

+
+
+# bool SubscriberMeta::matches(const std::string &other_topic, const std::vector<std::string> &other_topic_split) {
+#     // TODO: block subscriptions for invalid topics
+# 
+#     if(this->topic == "#")
+#       return true;
+# 
+#     // if(other_topic_split.size() == this->topic_split.size() - 1)
+#     //   if((other_topic + "/#") == this->topic)
+#     //     return true;
+# 
+#     for(u32 i = 0; i < this->topic_split.size(); i++) {
+#       auto &p = this->topic_split.at(i);
+# 
+#       if(p == "#")
+#         return true;
+# 
+#       if(p == "+")
+#         continue;
+# 
+#       if(p != other_topic_split.at(i))
+#         return false;
+#     }
+# 
+#     if(this->topic_split.size() == other_topic_split.size())
+#       return true;
+# 
+#     if(this->topic_split.at(this->topic_split.size() - 1) == "#")
+#       return true;
+# 
+#     return false;
+#   }
+
+
+def get(l: list, idx: int):
+  if idx >= 0 and idx < len(l):
+    return l[idx]
+  else:
+    return None
+
+
+def matches(topic, otherTopic) -> bool:
+
+  if topic == '#':
+    return True
+
+  topicSplit      = topic.split('/')
+  otherTopicSplit = otherTopic.split('/')
+
+  for i in range(len(topicSplit)):
+    p = get(topicSplit, i)
+
+    if p == '#':
+      return True
+
+    if p == '+':
+      continue
+
+    if p != get(otherTopicSplit, i):
+      return False
+
+  if len(topicSplit) == len(otherTopicSplit):
+    return True
+
+  # if get(topicSplit, len(topicSplit) - 2) == '#':
+  #   return True
+
+  return False
+

TopicMatcherTest.py

@@ -2,6 +2,7 @@
 #!nix-shell -i python -p "python3.withPackages(ps : with ps; [ pyyaml paho-mqtt ])"
 
 
+import DeltaMatcher
 from cProfile import run
 from SpiderMQTT import Topic
 
@@ -40,15 +41,25 @@ topicsNotMatching = [
 ]
 
 
-def runTest(t0: str, t1: str, expectation: bool):
+testFunc = Topic.compare
+# testFunc = DeltaMatcher.matches
 
-  result = Topic.compare(t0, t1) == expectation
 
-  expStr    = '=='     if expectation else '!='
-  resultSym = '✅'     if result      else '❌'
-  resultStr = 'PASSED' if result      else 'FAILED'
+def runTest(t0: str, t1: str, expectation: bool, swapInputs: bool = False):
 
-  print(f'[ {resultSym} ] Test: [ {t0} ]  {expStr}  [ {t1} ]   {resultStr}')
+  def _test(t0: str, t1: str):
+    result = testFunc(t0, t1) == expectation
+
+    expStr    = '=='     if expectation else '!='
+    resultSym = '✅'     if result      else '❌'
+    resultStr = 'PASSED' if result      else 'FAILED'
+
+    print(f'[ {resultSym} ] Test: [ {t0} ]  {expStr}  [ {t1} ]   {resultStr}')
+
+  _test(t0, t1)
+
+  if swapInputs:
+    _test(t1, t0)
 
 
 def runTestGroup(group: list, expectation: bool):