Merge pull request #78 from steve-anunknown/random-wp-method

add implementation for random wp method

aalpy/oracles/WpMethodEqOracle.py

@@ -3,6 +3,8 @@
 from aalpy.base.Oracle import Oracle
 from aalpy.base.SUL import SUL
 
+import random
+
 
 def state_characterization_set(hypothesis, alphabet, state):
     """
@@ -109,3 +111,58 @@ def find_cex(self, hypothesis):
                 self.cache.add(seq)
 
         return None
+
+
+class RandomWpMethodEqOracle(Oracle):
+    """
+    Implements the Random Wp-Method as described in "Complementing Model
+    Learning with Mutation-Based Fuzzing" by Rick Smetsers, Joshua Moerman,
+    Mark Janssen, Sicco Verwer.
+        1) sample uniformly from the states for a prefix
+        2) sample geometrically a random word
+        3) sample a word from the set of suffixes / state identifiers
+    """
+
+    def __init__(
+        self, alphabet: list, sul: SUL, expected_length=10, min_length=1, bound=1000
+    ):
+        super().__init__(alphabet, sul)
+        self.expected_length = expected_length
+        self.min_length = min_length
+        self.bound = bound
+
+    def find_cex(self, hypothesis):
+        if not hypothesis.characterization_set:
+            hypothesis.characterization_set = hypothesis.compute_characterization_set()
+
+        state_mapping = {s : state_characterization_set(hypothesis, self.alphabet, s) for s in hypothesis.states}
+
+        for _ in range(self.bound):
+            state = random.choice(hypothesis.states)
+            input = state.prefix
+            limit = self.min_length
+            while limit > 0 or random.random() > 1 / (self.expected_length + 1):
+                letter = random.choice(self.alphabet)
+                input += (letter,)
+                limit -= 1
+            if random.random() > 0.5:
+                # global suffix with characterization_set
+                input += random.choice(hypothesis.characterization_set)
+            else:
+                # local suffix
+                _ = hypothesis.execute_sequence(hypothesis.initial_state, input)
+                if state_mapping[hypothesis.current_state]:
+                    input += random.choice(state_mapping[hypothesis.current_state])
+                else:
+                    continue
+
+            self.reset_hyp_and_sul(hypothesis)
+            for ind, letter in enumerate(input):
+                out_hyp = hypothesis.step(letter)
+                out_sul = self.sul.step(letter)
+                self.num_steps += 1
+
+                if out_hyp != out_sul:
+                    self.sul.post()
+                    return input[: ind + 1]
+        return None

aalpy/oracles/__init__.py

@@ -8,7 +8,7 @@
 from .TransitionFocusOracle import TransitionFocusOracle
 from .UserInputEqOracle import UserInputEqOracle
 from .WMethodEqOracle import RandomWMethodEqOracle, WMethodEqOracle
-from .WpMethodEqOracle import WpMethodEqOracle
+from .WpMethodEqOracle import RandomWpMethodEqOracle, WpMethodEqOracle
 from .PacOracle import PacOracle
 from .ProvidedSequencesOracleWrapper import ProvidedSequencesOracleWrapper
 from .PerfectKnowledgeEqOracle import PerfectKnowledgeEqOracle

tests/test_deterministic.py

@@ -5,7 +5,7 @@
 from aalpy.learning_algs import run_Lstar
 from aalpy.oracles import WMethodEqOracle, WpMethodEqOracle, RandomWalkEqOracle, StatePrefixEqOracle, TransitionFocusOracle, \
     RandomWMethodEqOracle, BreadthFirstExplorationEqOracle, RandomWordEqOracle, CacheBasedEqOracle, \
-    KWayStateCoverageEqOracle
+    KWayStateCoverageEqOracle, RandomWpMethodEqOracle
 from aalpy.utils import get_Angluin_dfa, load_automaton_from_file
 from aalpy.utils.ModelChecking import bisimilar
 
@@ -113,6 +113,7 @@ def test_eq_oracles(self):
                                                        same_state_prob=0.3)
             w_method_eq_oracle = WMethodEqOracle(alphabet, sul, max_number_of_states=len(angluin_example.states) + 1)
             wp_method_eq_oracle = WpMethodEqOracle(alphabet, sul, max_number_of_states=len(angluin_example.states) + 1)
+            rwp_method_eq_oracle = RandomWpMethodEqOracle(alphabet, sul)
             random_W_method_eq_oracle = RandomWMethodEqOracle(alphabet, sul, walks_per_state=10, walk_len=50)
             bf_exploration_eq_oracle = BreadthFirstExplorationEqOracle(alphabet, sul, 4)
             random_word_eq_oracle = RandomWordEqOracle(alphabet, sul)
@@ -121,9 +122,10 @@ def test_eq_oracles(self):
 
             oracles = [random_walk_eq_oracle, random_word_eq_oracle,
                        random_W_method_eq_oracle, w_method_eq_oracle,
-                       wp_method_eq_oracle, kWayStateCoverageEqOracle,
-                       cache_based_eq_oracle, bf_exploration_eq_oracle,
-                       tran_cov_eq_oracle, state_origin_eq_oracle]
+                       wp_method_eq_oracle, rwp_method_eq_oracle,
+                       kWayStateCoverageEqOracle, cache_based_eq_oracle,
+                       bf_exploration_eq_oracle, tran_cov_eq_oracle,
+                       state_origin_eq_oracle]
 
             for oracle in oracles:
                 sul = AutomatonSUL(angluin_example)
@@ -166,15 +168,17 @@ def test_all_configuration_combinations(self):
                                                                  max_number_of_states=len(angluin_example.states))
                             wp_method_eq_oracle = WpMethodEqOracle(alphabet, sul,
                                                                  max_number_of_states=len(angluin_example.states))
+                            rwp_method_eq_oracle = RandomWpMethodEqOracle(alphabet, sul)
                             random_W_method_eq_oracle = RandomWMethodEqOracle(alphabet, sul,
                                                                               walks_per_state=10, walk_len=50)
                             bf_exploration_eq_oracle = BreadthFirstExplorationEqOracle(alphabet, sul, 4)
                             random_word_eq_oracle = RandomWordEqOracle(alphabet, sul)
                             cache_based_eq_oracle = CacheBasedEqOracle(alphabet, sul)
 
                             oracles = [random_walk_eq_oracle, random_word_eq_oracle, random_W_method_eq_oracle,
-                                       cache_based_eq_oracle, bf_exploration_eq_oracle, wp_method_eq_oracle,
-                                       tran_cov_eq_oracle, w_method_eq_oracle, state_origin_eq_oracle]
+                                       rwp_method_eq_oracle, cache_based_eq_oracle, bf_exploration_eq_oracle, 
+                                       wp_method_eq_oracle, tran_cov_eq_oracle, w_method_eq_oracle,
+                                       state_origin_eq_oracle]
 
                             if not cache:
                                 oracles.remove(cache_based_eq_oracle)

tests/test_rwpmethod_oracle.py

@@ -0,0 +1,113 @@
+import unittest
+
+try:
+    from aalpy.automata import MooreMachine, MooreState
+    from aalpy.learning_algs import run_Lstar
+    from aalpy.oracles.WpMethodEqOracle import RandomWpMethodEqOracle
+    from aalpy.SULs import AutomatonSUL
+    from aalpy.utils import visualize_automaton
+except ImportError:
+    import sys
+    from pathlib import Path
+
+    # if you want to run the test directly from CLI
+    # either from root or from tests folder
+    p = Path(__file__).parent.resolve()
+    sys.path.append(str(p))
+    sys.path.append(str(p.parent))
+    from aalpy.automata import MooreMachine, MooreState
+    from aalpy.learning_algs import run_Lstar
+    from aalpy.oracles.WpMethodEqOracle import RandomWpMethodEqOracle
+    from aalpy.SULs import AutomatonSUL
+    from aalpy.utils import visualize_automaton
+
+
+class TestRandomWpMethodOracle(unittest.TestCase):
+    @staticmethod
+    def gen_moore_from_state_setup(state_setup) -> MooreMachine:
+        # state_setup shoud map from state_id to tuple(output and transitions_dict)
+
+        # build states with state_id and output
+        states = {key: MooreState(key, val[0]) for key, val in state_setup.items()}
+
+        # add transitions to states
+        for state_id, state in states.items():
+            for _input, target_state_id in state_setup[state_id][1].items():
+                state.transitions[_input] = states[target_state_id]
+
+        # states to list
+        states = [state for state in states.values()]
+
+        # build moore machine with first state as starting state
+        mm = MooreMachine(states[0], states)
+
+        for state in states:
+            state.prefix = mm.get_shortest_path(mm.initial_state, state)
+
+        return mm
+
+    def generate_real_automata(self) -> MooreMachine:
+        state_setup = {
+            "a": ("a", {"x": "b1", "y": "a"}),
+            "b1": ("b", {"x": "b2", "y": "a"}),
+            "b2": ("b", {"x": "b3", "y": "a"}),
+            "b3": ("b", {"x": "b4", "y": "a"}),
+            "b4": ("b", {"x": "c", "y": "a"}),
+            "c": ("c", {"x": "a", "y": "a"}),
+        }
+
+        mm = self.gen_moore_from_state_setup(state_setup)
+        mm.characterization_set = mm.compute_characterization_set() + [tuple()]
+        return mm
+
+    def generate_hypothesis(self) -> MooreMachine:
+        state_setup = {
+            "a": ("a", {"x": "b", "y": "a"}),
+            "b": ("b", {"x": "b", "y": "a"}),
+        }
+
+        mm = self.gen_moore_from_state_setup(state_setup)
+        # ! computer_characterization_set does not work for Moore machines in general!
+        # mm.characterization_set = mm.compute_characterization_set() + [tuple()]
+        mm.characterization_set = [tuple(), ("x",), ("y",)]
+        return mm
+
+    def test_rwpmethod_oracle(self):
+        real = self.generate_real_automata()
+        hyp = self.generate_hypothesis()
+        # visualize_automaton(real)
+        # visualize_automaton(hyp)
+        assert set(real.get_input_alphabet()) == {"x", "y"}
+        assert set(hyp.get_input_alphabet()) == {"x", "y"}
+        assert len(real.states) == 6
+        assert len(hyp.states) == 2
+        alphabet = real.get_input_alphabet()
+        oracle = RandomWpMethodEqOracle(
+            alphabet, AutomatonSUL(real)
+        )
+        cex = oracle.find_cex(hyp)
+        assert cex is not None, "Expected a counterexample, but got None"
+
+    def test_rwpmethod_oracle_with_lstar(self):
+        real = self.generate_real_automata()
+        hyp = self.generate_hypothesis()
+        # visualize_automaton(real)
+        # visualize_automaton(hyp)
+        assert real.get_input_alphabet() == ["x", "y"]
+        assert hyp.get_input_alphabet() == ["x", "y"]
+        assert len(real.states) == 6
+        assert len(hyp.states) == 2
+        alphabet = real.get_input_alphabet()
+        oracle = RandomWpMethodEqOracle(
+            alphabet, AutomatonSUL(real)
+        )
+        lstar_hyp = run_Lstar(alphabet, AutomatonSUL(real), oracle, "moore")
+        # print(lstar_hyp)
+        # visualize_automaton(lstar_hyp)
+        assert (
+            len(lstar_hyp.states) == 6
+        ), f"Expected {6} states got {len(lstar_hyp.states)} in lstar hypothesis"
+
+
+if __name__ == "__main__":
+    unittest.main()