Explore-then-search trace generation

atest/robotMBT tests/07__processor_options/random_seeds/02__reusing_seed_reproduces_trace.robot

@@ -5,7 +5,7 @@ Documentation     This suite uses the `seed` argument to reproduce a previously
 ...               indicated by the number string.
 Suite Setup       Run keywords    Set suite variable    ${trace}    ${empty}
 ...                        AND    Treat this test suite Model-based    seed=aqmou-eelcuu-sniu-ugsyek-jyhoor
-Suite Teardown    Should be equal    ${trace}    6930142758
+Suite Teardown    Should be equal    ${trace}    6879523014
 Library           robotmbt
 
 *** Test Cases ***

atest/robotMBT tests/07__processor_options/random_seeds/03__retrace_with_refinement.robot

@@ -9,7 +9,7 @@ Documentation     This suite uses the `seed` argument to reproduce a previously
 ...               traces possible of varying length.
 Suite Setup       Run keywords    Set suite variable    ${trace}    ${empty}
 ...                        AND    Treat this test suite Model-based    seed=xou-pumj-ihj-oibiyc-surer
-Suite Teardown    Should be equal    ${trace}    B1A5BY3B4BX6B2
+Suite Teardown    Should be equal    ${trace}    B3AY2A5A6B4BX1
 Library           robotmbt
 
 *** Test Cases ***

atest/robotMBT tests/07__processor_options/random_seeds/04__retrace_with_step_modifiers.robot

@@ -6,7 +6,7 @@ Documentation     This suite uses the `seed` argument to reproduce a previously
 ...               included, the modifers will number the scenarios at random. The reproduced trace
 ...               must match for both the scenario order and the data order.
 Suite Setup       Treat this test suite Model-based    seed=iulr-vih-esycu-eyl-yfa
-Suite Teardown    Should be equal    ${trace}    H6G3E5I4D9F8J2B1A7C0
+Suite Teardown    Should be equal    ${trace}    H2G3C1E7I5B9F0A4D6J8
 Library           robotmbt
 
 *** Test Cases ***

atest/robotMBT tests/07__processor_options/random_seeds/05__retrace_combined.robot

@@ -15,8 +15,8 @@ Documentation     This suite uses the `seed` argument to reproduce a previously
 ...               a data choice by using step modifiers. The lower level includes a path choice.
 ...               Both low-level scenarios are equally valid, the only difference is that the data
 ...               choice is included either once or twice.
-Suite Setup       Treat this test suite Model-based    seed=gujuqt-iakm-oexo-xnu-huba
-Suite Teardown    Should be equal    ${trace}    ATQQPRSPRPXY
+Suite Setup       Treat this test suite Model-based    seed=kece-zwu-eihho-yli-rbixx
+Suite Teardown    Should be equal    ${trace}    APPSRTTPPXY
 Library           robotmbt
 
 *** Test Cases ***

robotmbt/suiteprocessors.py

@@ -87,26 +87,103 @@ def process_test_suite(self, in_suite, *, seed='new'):
                      "\n\t".join([f"{s.src_id}: {s.name}" for s in self.scenarios]))
 
         self._init_randomiser(seed)
-        self.shuffled = [s.src_id for s in self.scenarios]
-        random.shuffle(self.shuffled)  # Keep a single shuffle for all TraceStates (non-essential)
+        self._prio_order = self._pre_explore_paths()[0]
 
         # a short trace without the need for repeating scenarios is preferred
-        tracestate = self._try_to_reach_full_coverage(allow_duplicate_scenarios=False)
+        tracestate = self._try_to_reach_full_coverage(allow_duplicate_scenarios=False, randomise=False)
 
         if not tracestate.coverage_reached():
             logger.debug("Direct trace not available. Allowing repetition of scenarios")
-            tracestate = self._try_to_reach_full_coverage(allow_duplicate_scenarios=True)
+            tracestate = self._try_to_reach_full_coverage(allow_duplicate_scenarios=True, randomise=True)
             if not tracestate.coverage_reached():
                 raise Exception("Unable to compose a consistent suite")
 
         self.out_suite.scenarios = tracestate.get_trace()
         self._report_tracestate_wrapup(tracestate)
         return self.out_suite
 
-    def _try_to_reach_full_coverage(self, allow_duplicate_scenarios):
-        tracestate = TraceState(self.shuffled)
+    def _pre_explore_paths(self):
+        id_list = [s.src_id for s in self.scenarios]
+        random.shuffle(id_list)  # pre-shuffle to prevent scenario 1 from always getting first prio
+        tracestates = []
+        for prio_id in id_list:
+            # For each available id, give it pole position in the prio order and randomise the rest.
+            # The randomised bit will give some good variation and the prio_id will always be tried
+            # first giving more (not all) insight in its dependencies.
+            scenarios = id_list[:]
+            scenarios.remove(prio_id)
+            random.shuffle(scenarios)
+            scenarios.insert(0, prio_id)
+            logger.debug(f"Exploring with prio order {scenarios}")
+            tracestate = TraceState(scenarios)
+            count = 0
+            candidate_id = tracestate.next_candidate(retry=False, randomise=False)
+            while candidate_id is not None:
+                count += 1
+                candidate = self._select_scenario_variant(candidate_id, tracestate)
+                if candidate:  # No valid variant available in the current state
+                    modeller.try_to_fit_in_scenario(candidate, tracestate)
+                else:
+                    tracestate.reject_scenario(candidate_id)
+                candidate_id = tracestate.next_candidate(retry=False)
+            tracestates.append(tracestate)
+            logger.debug(f"Result trace: {tracestate.id_trace} "
+                         f"of length {len(tracestate)} after exploring {count} options.")
+
+        # first suggested prio-order:
+        #    - all ids from src_id list of longest trace
+        #    - but all unreached scenarios (over all traces) are moved to the front
+        #    - and all unreached in this trace move into second place
+        unreached = self._unreached_scenarios(tracestates)
+        longest = self._longest_trace(tracestates)
+        first_suggestion = [id for id in longest.c_pool if id in unreached]
+        first_suggestion += [id for id in longest.c_pool if id not in first_suggestion and longest.count(id) == 0]
+        first_suggestion += [id for id in longest.c_pool if longest.count(id) > 0]
+        logger.debug(f"Prio order suggestion: {first_suggestion}")
+
+        # second suggested prio-order:
+        #     - Take the last trace that saw a scenario inserted that had not been reached in any of the prior
+        #       traces. There may be a unique sequence in here to reach that scenario.
+        #     - reordering to insert unreached sceanrio is the same as with the first suggested prio-order
+        latest_new = self._last_new_coverage(tracestates)
+        second_suggestion = [id for id in latest_new.c_pool if id in unreached]
+        second_suggestion += [id for id in latest_new.c_pool
+                              if id not in second_suggestion and latest_new.count(id) == 0]
+        second_suggestion += [id for id in latest_new.c_pool if latest_new.count(id) > 0]
+
+        suggestions = [first_suggestion]
+        if first_suggestion != second_suggestion:
+            logger.debug(f"2nd suggestion: {second_suggestion}")
+            suggestions.append(second_suggestion)
+        return suggestions
+
+    @staticmethod
+    def _longest_trace(tracestate_list):
+        lengths = [len(ts) for ts in tracestate_list]
+        return tracestate_list[lengths.index(max(lengths))]
+
+    @staticmethod
+    def _last_new_coverage(tracestate_list):
+        ids = []
+        last_trace = tracestate_list[0]
+        for tracestate in tracestate_list:
+            for id in [int(float(long_id)) for long_id in tracestate.id_trace]:
+                if id not in ids:
+                    ids.append(id)
+                    last_trace = tracestate
+        return last_trace
+
+    @staticmethod
+    def _unreached_scenarios(tracestate_list):
+        total_coverage = dict().fromkeys(tracestate_list[0].c_pool, 0)
+        for ts in tracestate_list:
+            total_coverage = {k: total_coverage[k]+v for k, v in ts.c_pool.items()}
+        return [id for id in total_coverage if total_coverage[id] == 0]
+
+    def _try_to_reach_full_coverage(self, allow_duplicate_scenarios, randomise=False):
+        tracestate = TraceState(self._prio_order)
         while not tracestate.coverage_reached():
-            candidate_id = tracestate.next_candidate(retry=allow_duplicate_scenarios)
+            candidate_id = tracestate.next_candidate(retry=allow_duplicate_scenarios, randomise=randomise)
             if candidate_id is None:  # No more candidates remaining for this level
                 if not tracestate.can_rewind():
                     break

robotmbt/tracestate.py

@@ -30,6 +30,8 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+import random
+
 class TraceState:
     def __init__(self, scenario_indexes: list[int]):
         self.c_pool = {index: 0 for index in scenario_indexes}
@@ -39,6 +41,14 @@ def __init__(self, scenario_indexes: list[int]):
         self._snapshots = []  # Keeps details for elements in trace
         self._open_refinements = []
 
+    def copy(self):
+        cp = TraceState(self.c_pool.keys())
+        cp.c_pool.update(self.c_pool)
+        cp._tried = [triedlist[:] for triedlist in self._tried]
+        cp._snapshots = self._snapshots[:]
+        cp._open_refinements = self._open_refinements[:]
+        return cp
+
     @property
     def model(self):
         """returns the model as it is at the end of the current trace"""
@@ -68,16 +78,17 @@ def coverage_reached(self):
     def get_trace(self):
         return [snap.scenario for snap in self._snapshots]
 
-    def next_candidate(self, retry=False):
-        for i in self.c_pool:
-            if i not in self._tried[-1] and not self.is_refinement_active(i) and self.count(i) == 0:
-                return i
-        if not retry:
+    def next_candidate(self, retry: bool=False, randomise=False):
+        untried_candidates = [i for i in self.c_pool if i not in self._tried[-1]
+                              and not self.is_refinement_active(i)]
+        uncovered_candidates = [i for i in untried_candidates if self.count(i) == 0]
+
+        if uncovered_candidates:
+            return random.choice(uncovered_candidates) if randomise else uncovered_candidates[0]
+        elif not retry or not untried_candidates:
             return None
-        for i in self.c_pool:
-            if i not in self._tried[-1] and not self.is_refinement_active(i):
-                return i
-        return None
+        else:
+            return random.choice(untried_candidates) if randomise else untried_candidates[0]
 
     def count(self, index):
         """

utest/test_tracestate.py

@@ -512,6 +512,30 @@ def test_only_completed_scenarios_affect_drought(self):
         ts.confirm_full_scenario(2, 'two remainder', {})
         self.assertEqual(ts.coverage_drought, 0)
 
+    def test_tracestates_can_be_copied(self):
+        ts = TraceState([1, 2, 3])
+        ts.confirm_full_scenario(1, 'one', {})
+        ts.confirm_full_scenario(2, 'two', {})
+        cp = ts.copy()
+        self.assertEqual(ts.c_pool, cp.c_pool)
+        self.assertEqual(ts.tried, cp.tried)
+        self.assertEqual(ts.active_refinements, cp.active_refinements)
+        self.assertEqual(ts[-1], cp[-1])
+
+    def test_tracestate_copies_are_independent(self):
+        ts = TraceState([1, 2, 3])
+        ts.confirm_full_scenario(1, 'one', {})
+        ts.confirm_full_scenario(2, 'two', {})
+        cp = ts.copy()
+        cp.push_partial_scenario(3, 'three', {})
+        self.assertNotEqual(ts.active_refinements, cp.active_refinements)
+        cp.confirm_full_scenario(3, 'two', {})
+        self.assertEqual(len(cp), len(ts)+2)
+        self.assertNotEqual(ts.c_pool, cp.c_pool)
+        cp.rewind()
+        self.assertIn(3, cp.tried)
+        self.assertNotIn(3, ts.tried)
+
 
 if __name__ == '__main__':
     unittest.main()