Extended existing test framework with 'rust' command line switch.

src/sccd/cd/codegen/sccdlib.rs

@@ -92,38 +92,3 @@ Controller<EventType, OutputCallback, StatechartType> {
     }
   }
 }
-
-
-
-/// TEST CODE
-
-// #[derive(Copy, Clone)]
-// enum Event {
-//   A,
-//   B,
-// }
-
-// fn main() {
-//   let mut c: Controller<Event> = Default::default();
-//   c.add_input(Entry::<Event>{
-//     timestamp: 3,
-//     event: Event::A,
-//     target: Target::Broadcast,
-//   });
-//   c.add_input(Entry::<Event>{
-//     timestamp: 1,
-//     event: Event::A,
-//     target: Target::Broadcast,
-//   });
-//   c.add_input(Entry::<Event>{
-//     timestamp: 30,
-//     event: Event::A,
-//     target: Target::Broadcast,
-//   });
-//   c.add_input(Entry::<Event>{
-//     timestamp: 5,
-//     event: Event::A,
-//     target: Target::Broadcast,
-//   });
-//   c.run_until(Until::Timestamp(10));
-// }

src/sccd/statechart/cmd/gen_rust.py

@@ -20,8 +20,6 @@ if __name__ == "__main__":
 
     from sccd.statechart.parser.xml import *
     from sccd.test.parser.xml import *
-    from sccd.statechart.codegen.rust import compile_statechart
-    from sccd.test.codegen.rust import compile_test
     from sccd.util.indenting_writer import *
     from functools import partial
 
@@ -40,8 +38,12 @@ if __name__ == "__main__":
 
     if isinstance(statechart_or_test, Statechart):
         sys.stderr.write("Loaded statechart.\n")
+        
+        from sccd.statechart.codegen.rust import compile_statechart
         compile_statechart(statechart_or_test, globals, w)
 
     elif isinstance(statechart_or_test, list) and reduce(lambda x,y: x and y, (isinstance(test, TestVariant) for test in statechart_or_test)):
         sys.stderr.write("Loaded test.\n")
+
+        from sccd.test.codegen.rust import compile_test
         compile_test(statechart_or_test, w)

src/sccd/statechart/codegen/code_generation.txt

@@ -21,6 +21,7 @@ Roadmap
     - (DONE) goal: subset of SCCD tests passes:
         semantics/big_step_maximality/test_flag_takeone.xml
         semantics/big_step_maximality/test_ortho_takeone.xml
+        semantics/priority/test_source_parent.xml
 
     - no history
     - no action language

src/sccd/statechart/codegen/rust.py

@@ -243,15 +243,16 @@ def compile_statechart(sc: Statechart, globals: Globals, w: IndentingWriter):
         #
         # (1) The descendants of S2, S3, etc. if S1 is part of the "exit path":
         #
-        #      A        ---> And-state
+        #      A        ---> And-state on exit path
         #    / \   \
         #   S1  S2  S3 ...
-        # 
+        #
+        #    |
+        #    +--> S1 also on exit path
+        #
         # (2) The descendants of S, if S is the transition target
         def write_exit(exit_path: List[State]):
-            if len(exit_path) == 0:
-                w.writeln("%s.exit_current(output);" % ident_var(s))
-            else:
+            if len(exit_path) > 0:
                 s = exit_path[0]
                 if isinstance(s, HistoryState):
                     raise Exception("Can't deal with history yet!")
@@ -266,6 +267,9 @@ def compile_statechart(sc: Statechart, globals: Globals, w: IndentingWriter):
                         # Or-state
                         write_exit(exit_path[1:]) # continue recursively with the next child on the exit path
                 w.writeln("%s::exit_actions(output);" % ident_type(s))
+            else:
+                # w.writeln("%s.exit_current(output);" % ident_var(s))
+                pass
 
         def write_new_configuration(enter_path: List[State]):
             if len(enter_path) > 0:

src/sccd/test/cmd/run.py

@@ -1,28 +1,19 @@
 import argparse
 import unittest
-import threading
-import queue
 import functools
 
-import sys
-if sys.version_info.minor >= 7:
-  QueueImplementation = queue.SimpleQueue
-else:
-  QueueImplementation = queue.Queue
-
 from sccd.util.os_tools import *
 from sccd.util.debug import *
-from sccd.cd.static.cd import *
-from sccd.controller.controller import *
 from sccd.statechart.parser.xml import *
 from sccd.test.parser.xml import *
 from sccd.util import timer
 
 # A TestCase loading and executing a statechart test file.
 class Test(unittest.TestCase):
-  def __init__(self, src: str):
+  def __init__(self, src: str, enable_rust: bool):
     super().__init__()
     self.src = src
+    self.enable_rust = enable_rust
 
   def __str__(self):
     return self.src
@@ -37,72 +28,14 @@ class Test(unittest.TestCase):
       with timer.Context("parse test"):
         test_variants = parse_f(self.src, {"test" :test_rules})
 
-      for test in test_variants:
-        print_debug('\n'+test.name)
-        pipe = QueueImplementation()
-
-        current_big_step = []
-        def on_output(event: OutputEvent):
-          nonlocal current_big_step
-          if event.port == "trace":
-            if event.name == "big_step_completed":
-              if len(current_big_step) > 0:
-                pipe.put(current_big_step)
-              current_big_step = []
-          else:
-            current_big_step.append(event)
-
-
-        controller = Controller(test.cd, on_output)
-
-        for bag in test.input:
-          controller.schedule(
-            bag.timestamp.eval(None),
-            bag.events,
-            controller.all_instances()) # broadcast input events to all instances
-
-        def controller_thread():
-          try:
-            # Run as-fast-as-possible, always advancing time to the next item in event queue, no sleeping.
-            # The call returns when the event queue is empty and therefore the simulation is finished.
-            controller.run_until(None)
-          except Exception as e:
-            print_debug(e)
-            pipe.put(e, block=True, timeout=None)
-            return
-          # Signal end of output
-          pipe.put(None, block=True, timeout=None)
-
-        # start the controller
-        thread = threading.Thread(target=controller_thread)
-        thread.daemon = True # make controller thread exit when main thread exits
-        thread.start()
-
-        # check output
-        expected = test.output
-        actual = []
-
-        def fail(msg):
-          thread.join()
-          def pretty(output):
-            return '\n'.join("%d: %s" % (i, str(big_step)) for i, big_step in enumerate(output))
-          self.fail('\n'+test.name + '\n'+msg + "\n\nActual:\n" + pretty(actual) + "\n\nExpected:\n" + pretty(expected))
-
-        while True:
-          data = pipe.get(block=True, timeout=None)
-
-          if isinstance(data, Exception):
-            raise data # Exception was caught in Controller thread, throw it here instead.
-
-          elif data is None:
-            # End of output
-            break
-
-          else:
-            actual.append(data)
-
-        if actual != expected:
-          fail("Output differs from expected.")
+
+      if self.enable_rust:
+        from sccd.test.dynamic.test_rust import run_variants
+        run_variants(test_variants, self)
+      else:
+        from sccd.test.dynamic.test_interpreter import run_variant
+        for v in test_variants:
+          run_variant(v, self)
 
     except Exception as e:
       print_debug(e)
@@ -120,6 +53,7 @@ if __name__ == '__main__':
         epilog="Set environment variable SCCDDEBUG=1 to display debug information about the inner workings of the runtime.")
     argparser.add_argument('path', metavar='PATH', type=str, nargs='*', help="Tests to run. Can be a XML file or a directory. If a directory, it will be recursively scanned for XML files.")
     argparser.add_argument('--build-dir', metavar='BUILD_DIR', type=str, default='build', help="Directory for built tests. Defaults to 'build'")
+    argparser.add_argument('--rust', action='store_true', help="Instead of testing the interpreter, generate Rust code from test and run it.")
     args = argparser.parse_args()
 
     src_files = get_files(args.path,
@@ -137,8 +71,8 @@ if __name__ == '__main__':
         should_fail = os.path.basename(src_file).startswith("fail_")
 
         if should_fail:
-            suite.addTest(FailingTest(src_file))
+            suite.addTest(FailingTest(src_file, args.rust))
         else:
-            suite.addTest(Test(src_file))
+            suite.addTest(Test(src_file, args.rust))
 
     unittest.TextTestRunner(verbosity=2).run(suite)

src/sccd/test/dynamic/test_interpreter.py

@@ -0,0 +1,78 @@
+import queue
+import threading
+from sccd.test.static.syntax import TestVariant
+from sccd.controller.controller import *
+from sccd.util.debug import *
+
+import sys
+if sys.version_info.minor >= 7:
+  QueueImplementation = queue.SimpleQueue
+else:
+  QueueImplementation = queue.Queue
+
+def run_variant(test: TestVariant, unittest):
+  print_debug('\n'+test.name)
+  pipe = QueueImplementation()
+
+  current_big_step = []
+  def on_output(event: OutputEvent):
+    nonlocal current_big_step
+    if event.port == "trace":
+      if event.name == "big_step_completed":
+        if len(current_big_step) > 0:
+          pipe.put(current_big_step)
+        current_big_step = []
+    else:
+      current_big_step.append(event)
+
+
+  controller = Controller(test.cd, on_output)
+
+  for bag in test.input:
+    controller.schedule(
+      bag.timestamp.eval(None),
+      bag.events,
+      controller.all_instances()) # broadcast input events to all instances
+
+  def controller_thread():
+    try:
+      # Run as-fast-as-possible, always advancing time to the next item in event queue, no sleeping.
+      # The call returns when the event queue is empty and therefore the simulation is finished.
+      controller.run_until(None)
+    except Exception as e:
+      print_debug(e)
+      pipe.put(e, block=True, timeout=None)
+      return
+    # Signal end of output
+    pipe.put(None, block=True, timeout=None)
+
+  # start the controller
+  thread = threading.Thread(target=controller_thread)
+  thread.daemon = True # make controller thread exit when main thread exits
+  thread.start()
+
+  # check output
+  expected = test.output
+  actual = []
+
+  def fail(msg):
+    thread.join()
+    def pretty(output):
+      return '\n'.join("%d: %s" % (i, str(big_step)) for i, big_step in enumerate(output))
+    unittest.fail('\n'+test.name + '\n'+msg + "\n\nActual:\n" + pretty(actual) + "\n\nExpected:\n" + pretty(expected))
+
+  while True:
+    data = pipe.get(block=True, timeout=None)
+
+    if isinstance(data, Exception):
+      raise data # Exception was caught in Controller thread, throw it here instead.
+
+    elif data is None:
+      # End of output
+      break
+
+    else:
+      actual.append(data)
+
+  if actual != expected:
+    fail("Output differs from expected.")

src/sccd/test/dynamic/test_rust.py

@@ -0,0 +1,57 @@
+import threading
+import subprocess
+import tempfile
+import os
+from typing import *
+from sccd.test.static.syntax import TestVariant
+from sccd.test.codegen.rust import compile_test
+from sccd.util.indenting_writer import IndentingWriter
+from sccd.util.debug import *
+
+# Generate Rust code from the test case. This Rust code is piped to a Rust compiler (rustc) process, which reads from stdin. The Rust compiler outputs a binary in a temp dir. We then run the created binary as a subprocess.
+# If the result code of either the Rust compiler or the created binary is not 0 ("success"), the 'unittest' fails.
+def run_variants(variants: List[TestVariant], unittest):
+    if DEBUG:
+        stdout = None
+        # stderr = subprocess.STDOUT
+        stderr = None
+    else:
+        stdout = subprocess.DEVNULL
+        stderr = subprocess.STDOUT
+
+    output_file = os.path.join(tempfile.gettempdir(), "sccd_rust_out")
+    print_debug("Writing binary to " + output_file)
+
+    with subprocess.Popen(["rustc", "-o", output_file, "-"],
+        stdin=subprocess.PIPE,
+        stdout=stdout,
+        stderr=stderr) as pipe:
+
+        class PipeWriter:
+            def __init__(self, pipe):
+                self.pipe = pipe
+            def write(self, s):
+                self.pipe.stdin.write(s.encode(encoding='UTF-8'))
+
+        w = IndentingWriter(out=PipeWriter(pipe))
+
+        compile_test(variants, w)
+
+        pipe.stdin.close()
+
+        print_debug("Done generating Rust code")
+
+        pipe.wait()
+
+    print_debug("Done generating binary")
+
+    with subprocess.Popen([output_file],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE) as binary:
+
+        status = binary.wait()
+
+        if status != 0:
+            unittest.fail("Status code %d:\n%s%s" % (status, binary.stdout.read().decode('UTF-8'), binary.stderr.read().decode('UTF-8')))
+
+    print_debug("Done running binary")