ExternalVendorCode/trick
1
0
Fork 0
mirror of https://github.com/nasa/trick.git synced 2024-12-18 12:56:26 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

TrickOps: Update send_hs.py to work with latest Trick output (#1810)

Browse Source 
* Update send_hs.py to work with the new fields Trick's output produces
* Add unit tests and doctests for send_hs module
* Update TrickOps documentation to include info on send_hs usage
* Remove MonteCarloGenerationhelper.py's unecessary dependency on send_hs

Unrelated:

* Update pip when testing TrickOps per pip best practices

Closes #1807

Co-authored-by: Dan Jordan <daniel.d.jordan@nasa.gov>
This commit is contained in:
ddj116 2024-12-12 10:55:18 -06:00 committed by GitHub
parent 1330c424ad
commit 16828a9000
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 573 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/trickops.yml vendored
View File

@ -20,7 +20,7 @@ jobs:
- name: create virtual environment - name: create virtual environment
run: | run: |
cd share/trick/trickops/ cd share/trick/trickops/
python3 -m venv .venv && source .venv/bin/activate && pip3 install -r requirements.txt python3 -m venv .venv && source .venv/bin/activate && pip3 install --upgrade pip && pip3 install -r requirements.txt
- name: get and build koviz - name: get and build koviz
run: | run: |
cd /tmp/ && wget -q https://github.com/nasa/koviz/archive/refs/heads/master.zip && unzip master.zip cd /tmp/ && wget -q https://github.com/nasa/koviz/archive/refs/heads/master.zip && unzip master.zip

51
docs/documentation/miscellaneous_trick_tools/TrickOps.md
View File

@ -423,6 +423,57 @@ if ret == 0: # Successful generation
Note that the number of runs to-be-generated is configured somewhere in the `input.py` code and this module cannot robustly know that information for any particular use-case. This is why `monte_dir` is a required input to several functions - this directory is processed by the module to understand how many runs were generated. Note that the number of runs to-be-generated is configured somewhere in the `input.py` code and this module cannot robustly know that information for any particular use-case. This is why `monte_dir` is a required input to several functions - this directory is processed by the module to understand how many runs were generated.
## `send_hs` - TrickOps Helper Class for Parsing Simulation Diagnostics
Each Trick simulation run directory contains a number of Trick-generated metadata files, one of which is the `send_hs` file which represents all output during the simulation run sent through the Trick "(h)ealth and (s)tatus" messaging system. At the end of the `send_hs` message, internal diagnostic information is printed by Trick that looks a lot like this:
```
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
REALTIME SHUTDOWN STATS:
ACTUAL INIT TIME: 42.606
ACTUAL ELAPSED TIME: 55.551
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
SIMULATION TERMINATED IN
PROCESS: 0
ROUTINE: Executive_loop_single_thread.cpp:98
DIAGNOSTIC: Reached termination time
SIMULATION START TIME: 0.000
SIMULATION STOP TIME: 68.955
SIMULATION ELAPSED TIME: 68.955
USER CPU TIME USED: 55.690
SYSTEM CPU TIME USED: 0.935
SIMULATION / CPU TIME: 1.218
INITIALIZATION USER CPU TIME: 42.783
INITIALIZATION SYSTEM CPU TIME: 0.901
SIMULATION RAM USAGE: 1198.867MB
(External program RAM usage not included!)
VOLUNTARY CONTEXT SWITCHES (INIT): 792
INVOLUNTARY CONTEXT SWITCHES (INIT): 187
VOLUNTARY CONTEXT SWITCHES (RUN): 97
INVOLUNTARY CONTEXT SWITCHES (RUN): 14
```
The information provided here is a summary of how long the simulation ran, in both wall-clock time, and cpu time, both for initialization and run-time, and other useful metrics like how much peak RAM was used during run-time execution. Tracking this information can be useful for Trick-using groups so TrickOps provides a utility class for parsing this data. Here's an example of how you might use the `send_hs` module:
```python
import send_hs # Import the module
# Instantiate a send_hs instance, reading the given send_hs file
shs = send_hs.send_hs("path/to/SIM_A/RUN_01/send_hs")
start_time = shs.get('SIMULATION START TIME') # Get the value of sim start time
stop_time = shs.get('SIMULATION STOP TIME') # Get the value of sim stop time
realtime_ratio = shs.get('SIMULATION / CPU TIME') # Get the realtime ratio (how fast the sim ran)
# Instead of getting diagnostics individually, you can ask for the full dictionary
diagnostics = shs.get_diagnostics()
# Print the RAM usage from the dictionary
print(diagnostics['SIMULATION RAM USAGE'])
```
Plotting this data for regression scenarios can be quite useful - a lot of groups would want to know if the sim slowed down significantly and if so, when in the history this occurred. If you are already invested in Jenkins CI, you might be interested in using the `send_hs` module in conjunction with the [Jenkins plot plugin](https://plugins.jenkins.io/plot/). Here's an example of what tracking sim realtime ratio over time looks like in a workflow with about 15 regression scenarios all shown on a single plot:
![ExampleWorkflow In Action](images/sim_speed_example.png)
## More Information ## More Information
A lot of time was spent adding `python` docstrings to the modules in the `trickops/` directory and tests under the `trickops/tests/`. This README does not cover all functionality, so please see the in-code documentation and unit tests for more detailed information on the framework capabilities. A lot of time was spent adding `python` docstrings to the modules in the `trickops/` directory and tests under the `trickops/tests/`. This README does not cover all functionality, so please see the in-code documentation and unit tests for more detailed information on the framework capabilities.

BIN
docs/documentation/miscellaneous_trick_tools/images/sim_speed_example.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 1.2 MiB

1
share/trick/trickops/MonteCarloGenerationHelper.py
View File

@ -30,7 +30,6 @@ if ret == 0: # Successful generation
import sys, os import sys, os
import send_hs
import argparse, glob import argparse, glob
import subprocess, errno import subprocess, errno

295
share/trick/trickops/send_hs.py
View File

@ -1,67 +1,254 @@
import re, os import re, os
import pdb import pdb
# This global is the result of hours of frustration and debugging. This is only used by doctest
# but appears to be the only solution to the problem of __file__ not being an absolute path in
# some cases for some python versions and how that interacts with this class's os.chdir() when its
# base class initializes. If you attempt to define this_trick locally in the doctest block,
# which was my original attempt, you will find that the value of this_trick is different inside
# the __init__ doctest evaluation compared to any other member function. I believe this is only
# the case when using python version < 3.9, according to the information found here:
# https://note.nkmk.me/en/python-script-file-path/
# I do not like adding globals to "production code" just to facilitate a testing mechanism, but
# I don't know of any cleaner way way to do this. -Jordan 12/2024
this_trick = os.path.normpath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '../../..'))
class send_hs(object): class send_hs(object):
""" """
Reads a file containing the send_hs output and returns a send_hs Utility class for parsing simulation diagnostic data at the end of a
object containing the values from that output Trick-generated send_hs output file.
""" """
def __init__(self, hs_file): def __init__(self, hs_file=None):
self.hs_file = hs_file """
self.actual_init_time = None Initialize this instance.
self.actual_elapsed_time = None
self.start_time = None
self.stop_time = None
self.elapsed_time = None
self.actual_cpu_time_used = None
self.sim_cpu_time = None
self.init_cpu_time = None
self.parse()
def parse(self):
f = open(self.hs_file, 'r')
lines = f.readlines()
for line in lines:
self.actual_init_time = self.attempt_hs_match('ACTUAL INIT TIME',self.actual_init_time, line)
self.actual_elapsed_time = self.attempt_hs_match('ACTUAL ELAPSED TIME',self.actual_elapsed_time, line)
self.start_time = self.attempt_hs_match('SIMULATION START TIME',self.start_time, line)
self.stop_time = self.attempt_hs_match('SIMULATION STOP TIME',self.stop_time, line)
self.elapsed_time = self.attempt_hs_match('SIMULATION ELAPSED TIME',self.elapsed_time, line)
self.actual_cpu_time_used = self.attempt_hs_match('ACTUAL CPU TIME USED',self.actual_cpu_time_used, line)
self.sim_cpu_time = self.attempt_hs_match('SIMULATION / CPU TIME',self.sim_cpu_time, line)
self.init_cpu_time = self.attempt_hs_match('INITIALIZATION CPU TIME',self.init_cpu_time, line)
# TODO add capture of blade and DIAGNOSTIC: Reached termination time as success criteria
def attempt_hs_match(self, name, var, text): >>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
Parameters
----------
hs_file : str
Path to the send_hs output to read and parse
""" """
name: pattern to match (e.g. SIMULATION START TIME) self.hs_file = hs_file
var: variable to assign value if match found self.num_lines = None
text: text to search for pattern self._diagnostics = {} # Internal dict of diagnostic keys and values
returns: var if not found, found value if found self._missing_diagnostics = [] # List of diagnostics we failed to find
for k in self.keys():
self._diagnostics[k] = None
if hs_file:
self.parse()
def is_valid(self):
""" """
Check for validity of the parsed send_hs file. If any expected internal members
were not able to be parsed, store them in missing_diagnostics and return False.
If everything was found, return True
>>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh.is_valid()
True
Returns
-------
True if all expected internal members were parsed. False if any member wasn't found.
"""
self._missing_diagnostics = [] # Reset the internal list
for d in self._diagnostics:
if self._diagnostics[d] is None:
self._missing_diagnostics.append(d)
if self._missing_diagnostics:
return False
else:
return True
def missing_diagnostics(self):
'''
Check for validity and return a list of any missing diagnostics that we were
unable to find in the send_hs output
>>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh.missing_diagnostics()
[]
Returns
-------
list
All diagnotics that were unable to be parsed
'''
self.is_valid()
return self._missing_diagnostics
def parse(self, hs_file=None):
'''
Parse the content of self.hs_file and assign internal variables for each field found
if hs_file is given, overwrite self.hs_file with it's value
>>> sh = send_hs()
>>> sh.parse(os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh2 = send_hs()
>>> sh2.parse()
Traceback (most recent call last):
...
RuntimeError: send_hs file: 'None' cannot be read. You must provide a valid send_hs output file.
Parameters
----------
hs_file : str
Path to hs_file to parse. If None, self.hs_file is used instead.
Raises
------
RuntimeError
If no send_hs file was provided at construction or when calling this function
'''
if hs_file:
self.hs_file = hs_file
if not self.hs_file or not os.path.isfile(self.hs_file):
raise RuntimeError(f"send_hs file: '{self.hs_file}' cannot be read."
" You must provide a valid send_hs output file.")
self.num_lines = 0
with open(self.hs_file) as fp:
# Efficiency guard - we don't do an re pattern match until we've starting parsing the diagnostic msg
start_reading_diagnostics = False
for line in fp:
self.num_lines += 1
# NOTE this "attempt match" approach is less efficient but it should be
# robust to a future ordering change of the lines from Trick's output.
if start_reading_diagnostics:
for d in self._diagnostics:
if self._diagnostics[d] is None:
self._diagnostics[d] = self._attempt_hs_match(d, line)
# This text precedes the diagnostics output, use it as the trigger to start parsing
if 'REALTIME SHUTDOWN STATS:' in line:
start_reading_diagnostics = True
continue
# Set validity status immediately after parsing
self.is_valid()
def _attempt_hs_match(self, name, text):
"""
Parameters
----------
hs_file : str
Path to hs_file to parse. If None, self.hs_file is used instead.
name : str
Pattern to match (e.g. SIMULATION START TIME)
text : str
Text to search for pattern within
Returns
----------
float or None
Value of name if found, else: None
"""
name = name.replace('(', '\(').replace(')', '\)')
m = re.match(name+': +([-]?[0-9]*\.?[0-9]+)', text.strip()) m = re.match(name+': +([-]?[0-9]*\.?[0-9]+)', text.strip())
if m: if m:
return(float(m.group(1))) return(float(m.group(1)))
return(var) else:
return(None)
def get(self,name): def keys(self) -> list:
""" """
Get a value by the name that appears in the send_hs message Return a list of all possible keys/names that get() can accept This is
""" the master list of all diagnostics we search for when we parse the send_hs
if 'ACTUAL INIT TIME' in name: file, and naming should match 1:1 with the output of a Trick simulation
return self.actual_init_time
if 'ACTUAL ELAPSED TIME' in name: >>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
return self.actual_elapsed_time >>> sh.keys() #doctest: +ELLIPSIS
if 'SIMULATION START TIME' in name: ['ACTUAL INIT TIME', ... 'INVOLUNTARY CONTEXT SWITCHES (RUN)']
return self.start_time
if 'SIMULATION STOP TIME' in name: Returns
return self.stop_time -------
if 'SIMULATION ELAPSED TIME' in name: list
return self.elapsed_time All possible diagnostic names that get() can accept
if 'ACTUAL CPU TIME USED' in name: """
return self.actual_cpu_time_used return (['ACTUAL INIT TIME',
if 'SIMULATION / CPU TIME' in name: 'ACTUAL ELAPSED TIME',
return self.sim_cpu_time 'SIMULATION START TIME',
if 'INITIALIZATION CPU TIME' in name: 'SIMULATION STOP TIME',
return self.init_cpu_time 'SIMULATION ELAPSED TIME',
else: 'USER CPU TIME USED',
return None 'SYSTEM CPU TIME USED',
'SIMULATION / CPU TIME',
'INITIALIZATION USER CPU TIME',
'INITIALIZATION SYSTEM CPU TIME',
'SIMULATION RAM USAGE',
'VOLUNTARY CONTEXT SWITCHES (INIT)',
'INVOLUNTARY CONTEXT SWITCHES (INIT)',
'VOLUNTARY CONTEXT SWITCHES (RUN)',
'INVOLUNTARY CONTEXT SWITCHES (RUN)'
])
def get_diagnostic (self, name: str) -> float:
"""
Get a diagnostic value by it's name or partial name
The first name matched in the self._diagnostics dict will be returned
>>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh.get_diagnostic('ACTUAL INIT TIME')
42.606
>>> sh.get_diagnostic('SIMULATION RAM USAGE')
1198.867
Parameters
----------
name : str
Name or partial name of diagnostic to retrieve
Returns
-------
float
Value of diagnostic name given
Raises
------
LookupError
If name cannot be found in self._diagnostics
"""
for d in self._diagnostics:
if name in d:
return(self._diagnostics[d])
raise LookupError(f"Unable to get diagnostic '{name}'. Is it spelled correctly?")
def get(self, name: str) -> float:
"""
Get a diagnostic value by it's name or partial name. Convienence function that
calls self.get() directly.
>>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh.get('ACTUAL INIT TIME')
42.606
>>> sh.get('SIMULATION RAM USAGE')
1198.867
Returns
-------
float
Value of diagnostic name given
"""
return self.get_diagnostic(name)
@property
def diagnostics (self) -> dict:
return dict(self._diagnostics)
def get_diagnostics (self) -> dict:
"""
Get all diagnostics stored in internal self._diagnostics dictionary
>>> sh = send_hs(hs_file=os.path.join(this_trick,"share/trick/trickops/tests/send_hs.nominal"))
>>> sh.get_diagnostics() #doctest: +ELLIPSIS
{'ACTUAL INIT TIME': 42.606, 'ACTUAL ELAPSED TIME': 55.551, ... 14.0}
Returns
-------
dict
A copy of the internal self._diagnostics dictionary
"""
return (self.diagnostics)

2
share/trick/trickops/tests/run_tests.py
View File

@ -34,7 +34,7 @@ def run_tests(args):
# Run all doc tests by eating our own dogfood # Run all doc tests by eating our own dogfood
doctest_files = ['TrickWorkflow.py', 'WorkflowCommon.py', 'TrickWorkflowYamlVerifier.py', doctest_files = ['TrickWorkflow.py', 'WorkflowCommon.py', 'TrickWorkflowYamlVerifier.py',
'MonteCarloGenerationHelper.py'] 'MonteCarloGenerationHelper.py', 'send_hs.py']
wc = WorkflowCommon(this_dir, quiet=True) wc = WorkflowCommon(this_dir, quiet=True)
jobs = [] jobs = []
log_prepend = '_doctest_log.txt' log_prepend = '_doctest_log.txt'

5
share/trick/trickops/tests/send_hs.missing_all Normal file
View File

@ -0,0 +1,5 @@
|L 3|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Failed to resolve parameter 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable'
|L 2|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Mismatched data types for 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable' -> 'another_structure.myvariable'
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.flexMode.
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.myarray[0].
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable foo.bar.yippy.skippy.axial_sep_distance.

26
share/trick/trickops/tests/send_hs.missing_some Normal file
View File

@ -0,0 +1,26 @@
|L 3|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Failed to resolve parameter 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable'
|L 2|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Mismatched data types for 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable' -> 'another_structure.myvariable'
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.flexMode.
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.myarray[0].
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable foo.bar.yippy.skippy.axial_sep_distance.
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
REALTIME SHUTDOWN STATS:
ACTUAL INIT TIME: 32.306
ACTUAL ELAPSED TIME: 35.551
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
SIMULATION TERMINATED IN
PROCESS: 0
ROUTINE: Executive_loop_single_thread.cpp:98
DIAGNOSTIC: Reached termination time
SIMULATION START TIME: -10.000
SIMULATION STOP TIME: 78.955
SIMULATION ELAPSED TIME: 88.955
SYSTEM CPU TIME USED: 3.235
SIMULATION / CPU TIME: 9.218
INITIALIZATION USER CPU TIME: 72.763
SIMULATION RAM USAGE: 193.867MB
(External program RAM usage not included!)
VOLUNTARY CONTEXT SWITCHES (INIT): 292
VOLUNTARY CONTEXT SWITCHES (RUN): 37
INVOLUNTARY CONTEXT SWITCHES (RUN): 17

29
share/trick/trickops/tests/send_hs.nominal Normal file
View File

@ -0,0 +1,29 @@
|L 3|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Failed to resolve parameter 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable'
|L 2|2024/11/21,15:54:06|myworkstation| |T 0|0.000000| TrickParamCopy(UserCodeExample): Mismatched data types for 'fsw_out.fswOut_block.fswOut.fsw1HzOut.domain.subdomain.variable' -> 'another_structure.myvariable'
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.flexMode.
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable flex.flex_Obj.myarray[0].
|L 2|2024/11/21,15:54:07|myworkstation| |T 0|0.000000| Could not find Data Record variable foo.bar.yippy.skippy.axial_sep_distance.
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
REALTIME SHUTDOWN STATS:
ACTUAL INIT TIME: 42.606
ACTUAL ELAPSED TIME: 55.551
|L 0|2024/11/21,15:54:20|myworkstation| |T 0|68.955000|
SIMULATION TERMINATED IN
PROCESS: 0
ROUTINE: Executive_loop_single_thread.cpp:98
DIAGNOSTIC: Reached termination time
SIMULATION START TIME: 0.000
SIMULATION STOP TIME: 68.955
SIMULATION ELAPSED TIME: 68.955
USER CPU TIME USED: 55.690
SYSTEM CPU TIME USED: 0.935
SIMULATION / CPU TIME: 1.218
INITIALIZATION USER CPU TIME: 42.783
INITIALIZATION SYSTEM CPU TIME: 0.901
SIMULATION RAM USAGE: 1198.867MB
(External program RAM usage not included!)
VOLUNTARY CONTEXT SWITCHES (INIT): 792
INVOLUNTARY CONTEXT SWITCHES (INIT): 187
VOLUNTARY CONTEXT SWITCHES (RUN): 97
INVOLUNTARY CONTEXT SWITCHES (RUN): 14

3
share/trick/trickops/tests/test.py
View File

@ -5,6 +5,7 @@
import os, sys, pdb import os, sys, pdb
import unittest import unittest
import ut_send_hs
import ut_WorkflowCommon import ut_WorkflowCommon
import ut_TrickWorkflowYamlVerifier import ut_TrickWorkflowYamlVerifier
import ut_TrickWorkflow import ut_TrickWorkflow
@ -14,6 +15,7 @@ import ut_MonteCarloGenerationHelper
def load_tests(*args): def load_tests(*args):
passed_args = locals() passed_args = locals()
suite = unittest.TestSuite() suite = unittest.TestSuite()
suite.addTests(ut_send_hs.suite())
suite.addTests(ut_TrickWorkflowYamlVerifier.suite()) suite.addTests(ut_TrickWorkflowYamlVerifier.suite())
suite.addTests(ut_TrickWorkflow.suite()) suite.addTests(ut_TrickWorkflow.suite())
suite.addTests(ut_WorkflowCommon.suite()) suite.addTests(ut_WorkflowCommon.suite())
@ -23,6 +25,7 @@ def load_tests(*args):
# Local module level execution only # Local module level execution only
if __name__ == '__main__': if __name__ == '__main__':
suites = unittest.TestSuite() suites = unittest.TestSuite()
suites.addTests(ut_send_hs.suite())
suites.addTests(ut_TrickWorkflowYamlVerifier.suite()) suites.addTests(ut_TrickWorkflowYamlVerifier.suite())
suites.addTests(ut_TrickWorkflow.suite()) suites.addTests(ut_TrickWorkflow.suite())
suites.addTests(ut_WorkflowCommon.suite()) suites.addTests(ut_WorkflowCommon.suite())

216
share/trick/trickops/tests/ut_send_hs.py Normal file
View File

@ -0,0 +1,216 @@
import os, sys, glob
import unittest, shutil
import pdb
from testconfig import this_trick, tests_dir
import send_hs
def suite():
"""Create test suite from test cases here and return"""
suites = []
suites.append(unittest.TestLoader().loadTestsFromTestCase(SendHsTestCase))
suites.append(unittest.TestLoader().loadTestsFromTestCase(SendHsNoFileTestCase))
suites.append(unittest.TestLoader().loadTestsFromTestCase(SendHsMissingAllTestCase))
suites.append(unittest.TestLoader().loadTestsFromTestCase(SendHsMissingSomeTestCase))
suites.append(unittest.TestLoader().loadTestsFromTestCase(SendHsVerifyDiagNames))
return (suites)
class SendHsTestCase(unittest.TestCase):
def setUp(self):
# Nominal - reading a valid send_hs output file
self.instance = send_hs.send_hs(hs_file=os.path.join(tests_dir, 'send_hs.nominal'))
self.assertEqual(self.instance.missing_diagnostics(), [])
def tearDown(self):
if self.instance:
del self.instance
self.instance = None
def assert_values(self, instance):
self.assertEqual(instance.hs_file, os.path.join(tests_dir, 'send_hs.nominal'))
self.assertEqual(instance.num_lines, 29)
# Test the values of each individual diagnostic name
self.assertEqual(instance.get('ACTUAL INIT TIME'), 42.606)
self.assertEqual(instance.get('ACTUAL INIT'), 42.606) # Partial name test
self.assertEqual(instance.get('ACTUAL ELAPSED TIME'), 55.551)
self.assertEqual(instance.get('SIMULATION START TIME'), 0.0)
self.assertEqual(instance.get('SIMULATION STOP TIME'), 68.955)
self.assertEqual(instance.get('SIMULATION ELAPSED TIME'), 68.955)
self.assertEqual(instance.get('USER CPU TIME USED'), 55.690 )
self.assertEqual(instance.get('SYSTEM CPU TIME USED'), 0.935)
self.assertEqual(instance.get('SIMULATION / CPU TIME'), 1.218)
self.assertEqual(instance.get('INITIALIZATION USER CPU TIME'), 42.783 )
self.assertEqual(instance.get('INITIALIZATION SYSTEM CPU TIME'), 0.901)
self.assertEqual(instance.get('SIMULATION RAM USAGE'), 1198.867)
self.assertEqual(instance.get('VOLUNTARY CONTEXT SWITCHES (INIT)'), 792)
self.assertEqual(instance.get('VOLUNTARY CONTEXT SWITCHES'), 792) # Partial name test
self.assertEqual(instance.get('INVOLUNTARY CONTEXT SWITCHES (INIT)'), 187)
self.assertEqual(instance.get('VOLUNTARY CONTEXT SWITCHES (RUN)'), 97)
self.assertEqual(instance.get('INVOLUNTARY CONTEXT SWITCHES (RUN)'), 14 )
self.assertEqual(instance.is_valid(), True)
self.assertEqual(instance.missing_diagnostics(), [])
# Test the values of the full dict
diags = self.instance.get_diagnostics()
self.assertEqual(diags['ACTUAL INIT TIME'], 42.606)
self.assertEqual(diags['ACTUAL ELAPSED TIME'], 55.551)
self.assertEqual(diags['SIMULATION START TIME'], 0.0)
self.assertEqual(diags['SIMULATION STOP TIME'], 68.955)
self.assertEqual(diags['SIMULATION ELAPSED TIME'], 68.955)
self.assertEqual(diags['USER CPU TIME USED'], 55.690 )
self.assertEqual(diags['SYSTEM CPU TIME USED'], 0.935)
self.assertEqual(diags['SIMULATION / CPU TIME'], 1.218)
self.assertEqual(diags['INITIALIZATION USER CPU TIME'], 42.783 )
self.assertEqual(diags['INITIALIZATION SYSTEM CPU TIME'], 0.901)
self.assertEqual(diags['SIMULATION RAM USAGE'], 1198.867)
self.assertEqual(diags['VOLUNTARY CONTEXT SWITCHES (INIT)'], 792)
self.assertEqual(diags['INVOLUNTARY CONTEXT SWITCHES (INIT)'], 187)
self.assertEqual(diags['VOLUNTARY CONTEXT SWITCHES (RUN)'], 97)
self.assertEqual(diags['INVOLUNTARY CONTEXT SWITCHES (RUN)'], 14 )
# Test the other way to get diagnostics and ensure we get the same result
diags2 = self.instance.diagnostics
self.assertEqual(diags, diags2 )
def test_get_unknown_diag(self):
with self.assertRaises(LookupError):
self.instance.get('fake_diag_name')
def test_init(self):
# Init happened in setUp, make sure values are right
self.assert_values(self.instance)
def test_twostep_init(self):
# Construct without giving the instance a file to parse
self.tsi = send_hs.send_hs()
# Everything is None before the file has been parsed
for d in self.tsi._diagnostics:
self.assertEqual(self.tsi._diagnostics[d], None)
# Parse the file after construction
self.tsi = send_hs.send_hs('send_hs.nominal')
# Make sure values are right
self.assert_values(self.instance)
class SendHsNoFileTestCase(unittest.TestCase):
def setUp(self):
# No-file constructions
self.instance = send_hs.send_hs()
def test_init(self):
with self.assertRaises(RuntimeError):
self.instance.parse() # Nothing to parse, no file given
class SendHsMissingAllTestCase(unittest.TestCase):
def setUp(self):
# Invalid - reading a send_hs file missing all expected output
self.instance = send_hs.send_hs(hs_file=os.path.join(tests_dir, 'send_hs.missing_all'))
def tearDown(self):
if self.instance:
del self.instance
self.instance = None
def test_init(self):
self.assertEqual(self.instance.num_lines, 5)
self.assertEqual(self.instance.is_valid(), False)
self.assertEqual(self.instance.missing_diagnostics(), [
'ACTUAL INIT TIME',
'ACTUAL ELAPSED TIME',
'SIMULATION START TIME',
'SIMULATION STOP TIME',
'SIMULATION ELAPSED TIME',
'USER CPU TIME USED',
'SYSTEM CPU TIME USED',
'SIMULATION / CPU TIME',
'INITIALIZATION USER CPU TIME',
'INITIALIZATION SYSTEM CPU TIME',
'SIMULATION RAM USAGE',
'VOLUNTARY CONTEXT SWITCHES (INIT)',
'INVOLUNTARY CONTEXT SWITCHES (INIT)',
'VOLUNTARY CONTEXT SWITCHES (RUN)',
'INVOLUNTARY CONTEXT SWITCHES (RUN)'
])
class SendHsMissingSomeTestCase(unittest.TestCase):
def setUp(self):
# Invalid - reading a send_hs file missing all expected output
self.instance = send_hs.send_hs(hs_file=os.path.join(tests_dir, 'send_hs.missing_some'))
def tearDown(self):
if self.instance:
del self.instance
self.instance = None
def test_init(self):
self.assertEqual(self.instance.num_lines, 26)
self.assertEqual(self.instance.is_valid(), False)
self.assertEqual(self.instance.missing_diagnostics(), [
'USER CPU TIME USED',
'INITIALIZATION SYSTEM CPU TIME',
'INVOLUNTARY CONTEXT SWITCHES (INIT)',
])
self.assertEqual(self.instance.get('ACTUAL INIT TIME'), 32.306)
self.assertEqual(self.instance.get('ACTUAL INIT'), 32.306) # Partial name test
self.assertEqual(self.instance.get('ACTUAL ELAPSED TIME'), 35.551)
self.assertEqual(self.instance.get('SIMULATION START TIME'), -10.00)
self.assertEqual(self.instance.get('SIMULATION STOP TIME'), 78.955)
self.assertEqual(self.instance.get('SIMULATION ELAPSED TIME'), 88.955)
self.assertEqual(self.instance.get('SYSTEM CPU TIME USED'), 3.235)
self.assertEqual(self.instance.get('SIMULATION / CPU TIME'), 9.218)
self.assertEqual(self.instance.get('INITIALIZATION USER CPU TIME'), 72.763 )
self.assertEqual(self.instance.get('SIMULATION RAM USAGE'), 193.867)
self.assertEqual(self.instance.get('VOLUNTARY CONTEXT SWITCHES (INIT)'), 292)
self.assertEqual(self.instance.get('VOLUNTARY CONTEXT SWITCHES'), 292) # Partial name test
self.assertEqual(self.instance.get('VOLUNTARY CONTEXT SWITCHES (RUN)'), 37)
self.assertEqual(self.instance.get('INVOLUNTARY CONTEXT SWITCHES (RUN)'), 17 )
class SendHsVerifyDiagNames(unittest.TestCase):
'''
This test is bonkers. Here we search for the keys in the self._diagnostics (which
contain the text patterns we expect in the send_hs message) in the tracked source
code of Trick. This test should fail if the diagnostics are ever changed in name.
The other way to do this would be to ingest a real send_hs generated from a
sim built and run before these tests run, but that adds significant complexity
and a new dependency to this unit testing -Jordan 12/2024
'''
def setUp(self):
# An empty instance is all we need since the diag keys are baked into the class
self.instance = send_hs.send_hs()
self.files_to_search = [
os.path.join(this_trick, "trick_source/sim_services/Executive/Executive_shutdown.cpp"),
os.path.join(this_trick, "trick_source/sim_services/RealtimeSync/RealtimeSync.cpp"),
]
self.the_code = []
for source_file in self.files_to_search:
with open(source_file, 'r') as f:
self.the_code += f.readlines()
def tearDown(self):
if self.instance:
del self.instance
self.instance = None
def test_verify_diags(self):
# Look for the diagnostic text in the files to search, and note
# that they are FOUND if they are
for d in self.instance.keys():
if self.instance._diagnostics[d] == 'FOUND':
continue
for line in self.the_code:
if d in line:
self.instance._diagnostics[d] = 'FOUND'
break
diags = self.instance.diagnostics
# Assert that all expected text was found in the source code
for d in diags:
self.assertEqual(diags[d], 'FOUND')