Merge pull request #489 from chisholm/fix_xform_followedby_order

Fix observation expression DNF transformer to preserve FOLLOWEDBY order.

stix2/equivalence/pattern/transform/observation.py

@@ -282,6 +282,7 @@ class AbsorptionTransformer(
 
         A or (A and B) = A
         A or (A followedby B) = A
+        A or (B followedby A) = A
 
     Other variants do not hold for observation expressions.
     """
@@ -435,28 +436,35 @@ class DNFTransformer(ObservationExpressionTransformer):
 
         A and (B or C) => (A and B) or (A and C)
         A followedby (B or C) => (A followedby B) or (A followedby C)
+        (A or B) followedby C => (A followedby C) or (B followedby C)
     """
 
     def __transform(self, ast):
 
-        root_type = type(ast)  # will be AST class for AND or FOLLOWEDBY
-        changed = False
-        or_children = []
-        other_children = []
-        for child in ast.operands:
-            if isinstance(child, OrObservationExpression):
-                or_children.append(child.operands)
-            else:
-                other_children.append(child)
+        # If no OR children, nothing to do
+        if any(
+            isinstance(child, OrObservationExpression)
+            for child in ast.operands
+        ):
+            # When we distribute FOLLOWEDBY over OR, it is important to
+            # preserve the original FOLLOWEDBY order!  We don't need to do that
+            # for AND, but we do it anyway because it doesn't hurt, and we can
+            # use the same code for both.
+            iterables = []
+            for child in ast.operands:
+                if isinstance(child, OrObservationExpression):
+                    iterables.append(child.operands)
+                else:
+                    iterables.append((child,))
 
-        if or_children:
+            root_type = type(ast)  # will be AST class for AND or FOLLOWEDBY
             distributed_children = [
                 root_type([
                     _dupe_ast(sub_ast) for sub_ast in itertools.chain(
-                        other_children, prod_seq,
+                        prod_seq,
                     )
                 ])
-                for prod_seq in itertools.product(*or_children)
+                for prod_seq in itertools.product(*iterables)
             ]
 
             # Need to recursively continue to distribute AND/FOLLOWEDBY over OR
@@ -470,6 +478,7 @@ class DNFTransformer(ObservationExpressionTransformer):
 
         else:
             result = ast
+            changed = False
 
         return result, changed
 

stix2/test/test_pattern_equivalence.py

@@ -223,6 +223,10 @@ def test_obs_absorb_not_equivalent(patt1, patt2):
             "([a:b=1] OR [a:b=2]) FOLLOWEDBY ([a:b=3] OR [a:b=4])",
             "([a:b=1] FOLLOWEDBY [a:b=3]) OR ([a:b=1] FOLLOWEDBY [a:b=4]) OR ([a:b=2] FOLLOWEDBY [a:b=3]) OR ([a:b=2] FOLLOWEDBY [a:b=4])",
         ),
+        (
+            "([a:b=1] OR [a:b=2]) FOLLOWEDBY ([a:b=5] AND [a:b=6])",
+            "([a:b=1] FOLLOWEDBY ([a:b=5] AND [a:b=6])) OR ([a:b=2] FOLLOWEDBY ([a:b=5] AND [a:b=6]))",
+        ),
     ],
 )
 def test_obs_dnf_equivalent(patt1, patt2):
@@ -243,6 +247,10 @@ def test_obs_dnf_equivalent(patt1, patt2):
             "[a:b=1] WITHIN 2 SECONDS",
             "[a:b=1] REPEATS 2 TIMES",
         ),
+        (
+            "[a:b=1] FOLLOWEDBY ([a:b=2] OR [a:b=3])",
+            "([a:b=2] FOLLOWEDBY [a:b=1]) OR ([a:b=1] FOLLOWEDBY [a:b=3])",
+        ),
     ],
 )
 def test_obs_not_equivalent(patt1, patt2):