#!/usr/bin/env python # # Automatically generated Python sequence chart (MSC) implementation import os import sys import time import signal import Queue # The following two lines are not needed as the tool is now installed with pip #taste_inst = os.popen('taste-config --prefix').readlines()[0].strip() #sys.path.append(taste_inst+'/share/asn1-editor') from asn1_value_editor.Scenario import Scenario, PollerThread from PySide.QtCore import QCoreApplication, Qt from asn1_value_editor.udpcontroller import tasteUDP # Generated due to "ground_trace_201810202250.msc" # From the section: MSCDOCUMENT automade # You may edit the scenario below or create new ones (use the @Scenario decorator) # When you add new scenarios, they will all run in parallel. # # You can use these three API functions to communicate with the main binary: # (1) queue.sendMsg('Name of your Provided Interface', 'Parameter value in ASN.1 format') # The parameters are expressed textually in ASN.1 Value Notation # (also called GSER). For example a record's syntax is: # { fieldName1 value1, fieldName2 value2 } # (2) queue.expectMsg ('Name of RI', # 'Parameter value in Extended ASN.1 format', # lineNo=optional line reference, # ignoreOther=True/False) # Extended ASN.1 format lets you replace a field value with a star (*) # meaning that you do not want the tool to check it against any specific value # ignoreOther: set to True if you want the tool to ignore other messages # and want to trigger an error only when you get this message with the wrong parameters # (3) (msgId, val) = queue.getNextMsg(timeout=10) # if msgId == 'Name of an interface': # print 'The value is', val.fieldName.Get() @Scenario def Exercise_ground(queue): # queue is actually an instance of the Scenario class '''ground processing''' random_message = '{a FALSE, b FALSE, c 0, d 1, e {3, 4}, f {foo TRUE, bar FALSE}, g {huhu FALSE, b FALSE}, h {}, ii x: FALSE, jj uiop: TRUE, k hop: TRUE}' queue.sendMsg('go', random_message, lineNo=11) try: queue.expectMsg('gone', random_message, lineNo=12, ignoreOther=False) except TypeError as err: raise return 0 def runScenario(pipe_in=None, pipe_out=None, udpController=None): # Queue for getting scenario status log = Queue.Queue() if udpController: ground = Exercise_ground(log, name='Exercise_ground') udpController.slots.append(ground.msq_q) ground.wait() udpController.slots.remove(ground.msg_q) return 0 # ground.status else: # Use old-style message queue poller = PollerThread() ground = Exercise_ground(log, name='Exercise_ground') poller.slots.append(ground.msg_q) poller.start() ground.start() # Wait and log messages from both scenarii while True: time.sleep(0.001) try: scenario, severity, msg = log.get(block=False) except Queue.Empty: pass else: log.task_done() try: # If called from the GUI, send log through pipe pipe_out.send((scenario, severity, msg)) except AttributeError: print('[{level}] {name} - {msg}'.format (level=severity, name=scenario, msg=msg)) if severity == 'ERROR' or msg == 'END': # Stop execution on first error or completed scenario try: pipe_out.send(('All', 'COMMAND', 'END')) except AttributeError: ground.stop() poller.stop() return try: if pipe_out.poll(): cmd = pipe_out.recv() if cmd == 'STOP': ground.stop() poller.stop() return except AttributeError: pass if __name__ == "__main__": signal.signal(signal.SIGINT, signal.SIG_DFL) udpController = None if '--udp' in sys.argv: # Create UDP Controller with default IP/Port values (127.0.0.1:7755:7756) udpController = tasteUDP() QCoreApplication(sys.argv) sys.exit(runScenario(udpController))