#
# ISC License
#
# Copyright (c) 2026, Autonomous Vehicle Systems Lab, University of Colorado Boulder
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
#
import numpy as np
from Basilisk.architecture import messaging
from Basilisk.moduleTemplates import autoCppModule as module_under_test
from Basilisk.utilities import SimulationBaseClass, macros
[docs]
def test_autoCppModule():
"""Run the draft module and check its output message publication.
**Validation Test Description**
Connect blank input messages and execute three scheduled updates.
Check the module call count, output written status, sample times,
and message write times. These checks verify scheduling and messaging;
add numerical payload checks when the module algorithm is implemented.
"""
task_name = "unitTask"
simulation = SimulationBaseClass.SimBaseClass()
time_step = macros.sec2nano(0.5) # [ns]
process = simulation.CreateNewProcess("TestProcess")
process.addTask(simulation.CreateNewTask(task_name, time_step))
# Set up the module to be tested.
module = module_under_test.AutoCppModule()
module.ModelTag = "autoCppModuleTag"
simulation.AddModelToTask(task_name, module)
# Configure and retain blank input messages, then subscribe the module.
input_messages = []
input_payload = messaging.AttRefMsgPayload()
input_message = messaging.AttRefMsg().write(input_payload)
module.someInMsg.subscribeTo(input_message)
input_messages.append(input_message)
input_payload = messaging.AttRefMsgPayload()
input_message = messaging.AttRefMsg().write(input_payload)
module.some2InMsg.subscribeTo(input_message)
input_messages.append(input_message)
input_payload = messaging.CSSConfigMsgPayload()
input_message = messaging.CSSConfigMsg().write(input_payload)
module.anotherInMsg.subscribeTo(input_message)
input_messages.append(input_message)
input_payload = messaging.CSSConfigLogMsgPayload()
input_message = messaging.CSSConfigLogMsg().write(input_payload)
module.anotherCppInMsg.subscribeTo(input_message)
input_messages.append(input_message)
# Record each output after the module runs in the same task.
output_readers = {
"some2OutMsg": messaging.AttRefMsgReader(),
"someOutMsg": messaging.SCStatesMsgReader(),
"anotherCppOutMsg": messaging.DataStorageStatusMsgReader(),
}
output_recorders = {}
for name, reader in output_readers.items():
reader.subscribeTo(getattr(module, name))
recorder = reader.recorder()
output_recorders[name] = recorder
simulation.AddModelToTask(task_name, recorder)
simulation.InitializeSimulation()
simulation.ConfigureStopTime(2 * time_step)
simulation.ExecuteSimulation()
expected_times = np.array([0, time_step, 2 * time_step], dtype=np.uint64) # [ns]
assert module.CallCounts == len(expected_times), "Module did not run three times"
for name, recorder in output_recorders.items():
assert output_readers[name].isWritten(), f"{name} was never written"
np.testing.assert_array_equal(
recorder.times(), expected_times, err_msg=f"{name} recording times"
)
np.testing.assert_array_equal(
recorder.timesWritten(), expected_times, err_msg=f"{name} write times"
)
# Add module-specific numerical payload checks here.
if __name__ == "__main__":
test_autoCppModule()