Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_as_kcd_string(self):
"""Test the KCD dump functionality.
"""
filename = os.path.join('tests', 'files', 'dump.kcd')
db = cantools.database.load_file(filename)
with open(filename, 'rb') as fin:
self.assertEqual(db.as_kcd_string().encode('ascii'), fin.read())
def test_vehicle(self):
filename = os.path.join('tests', 'files', 'vehicle.dbc')
db = cantools.database.load_file(filename)
self.assertEqual(len(db.nodes), 1)
self.assertEqual(db.nodes[0].name, 'UnusedNode')
self.assertEqual(len(db.messages), 217)
self.assertEqual(db.messages[216].name, 'RT_SB_Gyro_Rates')
self.assertEqual(db.messages[216].frame_id, 155872546)
self.assertEqual(db.messages[216].senders, [])
self.assertEqual(str(db.messages[0]),
"message('RT_SB_INS_Vel_Body_Axes', 0x9588322, True, 8, None)")
self.assertEqual(repr(db.messages[0].signals[0]),
"signal('Validity_INS_Vel_Forwards', 0, 1, 'little_endian', "
"False, 1, 0, 0, 1, 'None', False, None, None, 'Valid when "
"bit is set, invalid when bit is clear.')")
self.assertEqual(db.messages[0].signals[0].receivers, [])
self.assertEqual(db.messages[0].cycle_time, None)
self.assertEqual(db.messages[0].send_type, None)
self.assertEqual(repr(db.nodes[0]), "node('UnusedNode', None)")
def test_database_signals_check_failure(self):
signal = cantools.database.can.Signal('S',
7,
33,
'big_endian')
message = cantools.database.can.Message(37,
'M',
4,
[signal],
strict=False)
with self.assertRaises(cantools.database.errors.Error) as cm:
cantools.database.can.Database([message])
self.assertEqual(str(cm.exception),
'The signal S does not fit in message M.')
def test_strict_load(self):
filenames = [
os.path.join('tests', 'files', 'bad_message_length.kcd'),
os.path.join('tests', 'files', 'bad_message_length.dbc'),
os.path.join('tests', 'files', 'bad_message_length.sym')
]
for filename in filenames:
# Strict true.
with self.assertRaises(cantools.database.Error) as cm:
cantools.database.load_file(filename, strict=True)
self.assertEqual(str(cm.exception),
'The signal Signal1 does not fit in message Message1.')
# Strict false.
db = cantools.database.load_file(filename, strict=False)
# Sanity checks.
message_1 = db.get_message_by_frame_id(1)
self.assertEqual(message_1.length, 1)
self.assertEqual(message_1.signals[0].start, 8)
self.assertEqual(message_1.signals[0].length, 1)
(
'Message4',
1.9,
"Expected signal 'Signal1' value greater than or equal to 2 in "
"message 'Message4', but got 1.9."
),
(
'Message4',
8.1,
"Expected signal 'Signal1' value less than or equal to 8 in "
"message 'Message4', but got 8.1."
)
]
for message_name, signal_value, error_message in datas:
with self.assertRaises(cantools.database.EncodeError) as cm:
db.encode_message(message_name, {'Signal1': signal_value})
self.assertEqual(str(cm.exception), error_message)
# Values out of range, but range checks disabled.
datas = [
('Message1', 0),
('Message1', 3)
]
for message_name, signal_value in datas:
db.encode_message(message_name,
{'Signal1': signal_value},
strict=False)
# Missing value.
return
if len(data) != message.length:
self._discarded += 1
return
name = message.name
if message.is_multiplexed():
# Handle the case where a multiplexer index is used that isn't
# specified in the DBC file (ie. outside of the range). In this
# case, we just discard the message, like we do when the CAN
# message ID or length doesn't match what's specified in the DBC.
try:
name = format_multiplexed_name(message, data, True)
except database.DecodeError:
self._discarded += 1
return
if self._single_line:
formatted = format_message(message, data, True, True)
self._formatted_messages[name] = [
'{:12.3f} {}'.format(timestamp, formatted)
]
else:
formatted = format_message(message, data, True, False)
lines = formatted.splitlines()
formatted = ['{:12.3f} {}'.format(timestamp, lines[1])]
formatted += [14 * ' ' + line for line in lines[2:]]
self._formatted_messages[name] = formatted
if name not in self._filtered_sorted_message_names:
def _do_generate_c_source(args):
dbase = database.load_file(args.infile,
encoding=args.encoding,
strict=not args.no_strict)
if args.database_name is None:
basename = os.path.basename(args.infile)
database_name = os.path.splitext(basename)[0]
database_name = camel_to_snake_case(database_name)
else:
database_name = args.database_name
filename_h = database_name + '.h'
filename_c = database_name + '.c'
fuzzer_filename_c = database_name + '_fuzzer.c'
fuzzer_filename_mk = database_name + '_fuzzer.mk'
header, source, fuzzer_source, fuzzer_makefile = generate(