Writing clean scripts 

Contents

Writing clean scripts

A recurring issue when developing scripts using ANDi is how to ensure that after the scripts are executed or stopped, all their references have to be managed and cleaned out correctly.

Unmanaged references can cause incremental executions of scripts to yield wrong or duplicate results (example: event function callback executing more than once and with old implementation).

Note

Unmanaged references in a script can refer to:

Events callback functions not removed after script execution (-= operator not used).
Unhandled exception in script ending script prematurely and preventing script clean up code block to be executed.
Inaccessible objects not cleaned in scripts.

ANDi will attempt to cleanup all object referenced by variables (at the end of the script execution, ANDi loops all global variables).
If for some reason the object is not accessible (wrapped inside a python class, not referenced by python code or no longer referenced…) then ANDi won’t be able to clean it.
So it is important to write clean scripts to avoid these issues.

Dereferencing event callback functions 

Events are a powerful tool to execute a specific code when a certain event occurs, this is done through adding an event callback function to the event (using += operator to the event).
However, when an event function callback is not dereferenced when no longer needed, it will always keep executing with the event (because the event itself keeps referencing this function).
Therefor a script must always dereference the callback functions that it adds to events to ensure that scripts execute normally (using -= operator to the event).

How to use

# adding callback function to event
obj.event_name += callback_function

# other code to execute

# Dereferencing callback function from event
obj.event_name -= callback_function

Important

There is a possibility that code execution will stop before dereferencing a callback function either because the code executed before it raises an exception or because script was stopped.

So dereferencing callback functions needs to be done in a code block that will always execute (using try/except/finally or atexit library).

Using try/except/finally blocks 

An exception is an error that happens during execution of a program.
When that error occurs, an exception will be generated that can be handled.
This is done using try/except/finally python blocks.

How to use

try:
    # code to execute
except:
    # code to execute in case an exception is thrown
finally:
    # code that will always execute after the try/except block
    # this is where cleaning code happens

Note

It is possible to use except with specific exception type (for example except ValueError).
More about try/except/finally can be found in python exceptions doc.

Sample script

from time import sleep

def on_elapsed(source, event_args):
    # execute timer related code here
    print(event_args.time)

l_timer = andi.create_timer()
try: 
    # do some code
    l_timer.on_time_elapsed += on_elapsed
    l_timer.start()
    sleep(5)
    print("this will print if code executes normally")
except: 
    # in case of exception do this
    print("this will print if an exception is caught")
    # rethrow exception to be caught and displayed/handled by ANDi
    raise 
finally: 
    # this will always run after the try catch function
    # even when a code is stopped by the user
	# so cleaning up the code should be done here
    print("this will always run after try/except block even if script is stopped")
    l_timer.stop()
    l_timer.on_time_elapsed -= on_elapsed

Using atexit library 

atexit module defines functions to register and unregister cleanup functions, these functions are automatically executed upon normal script termination.

atexit will run these functions in the reverse ordre in which they were registered.

How to use

import atexit
def function_to_register(first, second):
    # clean up code to execute

atexit.register(function_to_register, 'first_value', 'second_value')
# or:
atexit.register(function_to_register, first='first_value', second='second_value')

Note

It is possible to use @atexit.register decorator for functions that have no arguments.
More about atexit library can be found in python atexit doc.

Sample script using decorator

from time import sleep
import atexit

def on_elapsed(source, event_args):
    # execute timer related code here
    print(event_args.time)

@atexit.register
def clean_up():
    global l_timer
    l_timer.on_time_elapsed -= on_elapsed
    l_timer.stop()

# clean up function is already registered with the decorator
# keep in mind the decorator only works with parameter-less functions
l_timer = andi.create_timer()
l_timer.on_time_elapsed += on_elapsed
l_timer.start()
sleep(5)

Sample script using register function

from time import sleep
import atexit

def on_elapsed(source, event_args):
    # execute timer related code here
    print(event_args.time)

def clean_up():
    global l_timer
    l_timer.on_time_elapsed -= on_elapsed
    l_timer.stop()

def clean_up_args(local_timer):
    local_timer.on_time_elapsed -= on_elapsed
    local_timer.stop()

l_timer = andi.create_timer()
l_timer_2 = andi.create_timer()
# register clean up functions
# register function with argument
atexit.register(clean_up_args, l_timer_2)
# register function without argument
atexit.register(clean_up)
# code to run
l_timer.on_time_elapsed += on_elapsed
l_timer_2.on_time_elapsed += on_elapsed
l_timer.start()
l_timer_2.start()
sleep(5)

Using tc_return_foo methods 

tc_return_success(): informs about the success of the test case.
tc_return_failure(): informs about the failure of the test case.
tc_wait_for_return(optional_timeout): blocks execution until a tc_return_ function is called or the timeout, if mentioned, is elapsed.
tc_return_continue(): unblocks tc_wait_for_return without setting any status.
tc_return_skipped(): marks the script as skipped. To mark the script as skipped and stop it immediately, use current_script.skip().

Sample script

import random

RANDOM_NB = 0

# Create SOME/IP message
msg_someip = message_builder.create_someip_message("channel_1" , "channel_1")

# Callback function, invoked on each incoming SOME/IP message
def on_message_received(msg):
    global RANDOM_NB
    # A random number (takes 1 or 2) to test both results *Success* and *Failure*
    RANDOM_NB = random.randint(1, 2)
    # Unblock the execution caused by tc_wait_for_return
    tc_return_continue()
 
# Register callback function
msg_someip.on_message_received += on_message_received

# Start live capturing for SOME/IP message
msg_someip.start_capture()

# Send one message
msg_someip.send()

# Block the test case execution until the message is received 
# since on_message_received function calls tc_return_continue()
tc_wait_for_return()

## test result
print("--  RESULT  --")
if RANDOM_NB == 1:
    # Inform about the success of the test case.
    tc_return_success('** Success **')             
else:
    # Inform about the failure of the test case.
    tc_return_failure('** Failure **')

# De-register callback function
msg_someip.on_message_received -= on_message_received

# Stop capturing
msg_someip.stop_capture()

Writing clean scripts

Dereferencing event callback functions

Using try/except/finally blocks

Using atexit library

Using tc_return_foo methods

Writing clean scripts 

Dereferencing event callback functions 

Using try/except/finally blocks 

Using atexit library 

Using tc_return_foo methods 