diff options
Diffstat (limited to 'src/target/trx_toolkit/test_data_msg.py')
| -rw-r--r-- | src/target/trx_toolkit/test_data_msg.py | 193 |
1 files changed, 193 insertions, 0 deletions
diff --git a/src/target/trx_toolkit/test_data_msg.py b/src/target/trx_toolkit/test_data_msg.py new file mode 100644 index 00000000..62542031 --- /dev/null +++ b/src/target/trx_toolkit/test_data_msg.py @@ -0,0 +1,193 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +# TRX Toolkit +# Unit test for TRXD message codec +# +# (C) 2019 by Vadim Yanitskiy <axilirator@gmail.com> +# +# All Rights Reserved +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +import unittest + +from data_msg import DATAMSG, DATAMSG_L12TRX, DATAMSG_TRX2L1 + +class DATAMSG_Test(unittest.TestCase): + # Compare message a with message b + def _compare_msg(self, a, b): + # Make sure we're comparing messages of the same type + self.assertEqual(a.__class__, b.__class__) + + # Compare common header fields + self.assertEqual(a.ver, b.ver) + self.assertEqual(a.fn, b.fn) + self.assertEqual(a.tn, b.tn) + + # Burst bits (if present) + self.assertEqual(a.burst, b.burst) + + # TRX2L1 specific fields + if isinstance(a, DATAMSG_L12TRX): + self.assertEqual(a.pwr, b.pwr) + + # L12TRX specific fields + if isinstance(a, DATAMSG_TRX2L1): + # Version independent fields + self.assertEqual(a.toa256, b.toa256) + self.assertEqual(a.rssi, b.rssi) + + # Version specific fields + if a.ver >= 1: + self.assertEqual(a.nope_ind, b.nope_ind) + self.assertEqual(a.mod_type, b.mod_type) + self.assertEqual(a.tsc_set, b.tsc_set) + self.assertEqual(a.tsc, b.tsc) + self.assertEqual(a.ci, b.ci) + + # Make sure that message validation throws a ValueError + def test_validate(self): + # Unknown version + with self.assertRaises(ValueError): + msg = DATAMSG(fn = 0, tn = 0, ver = 100) + msg.validate() + + # Uninitialized field + with self.assertRaises(ValueError): + msg = DATAMSG() + msg.validate() + with self.assertRaises(ValueError): + msg = DATAMSG(fn = None, tn = 0) + msg.validate() + + # Out-of-range value(s) + with self.assertRaises(ValueError): + msg = DATAMSG(fn = -1, tn = 0) + msg.validate() + with self.assertRaises(ValueError): + msg = DATAMSG(fn = 0, tn = 10) + msg.validate() + + # Validate header and burst randomization + def test_rand_hdr_burst(self): + msg_l12trx = DATAMSG_L12TRX() + msg_trx2l1 = DATAMSG_TRX2L1() + + for i in range(100): + msg_l12trx.rand_burst() + msg_trx2l1.rand_burst() + msg_l12trx.rand_hdr() + msg_trx2l1.rand_hdr() + + msg_l12trx.validate() + msg_trx2l1.validate() + + def _test_enc_dec(self, msg, legacy = False, nope_ind = False): + # Prepare a given message (randomize) + msg.rand_hdr() + + # NOPE.ind contains no burst + if not nope_ind: + msg.rand_burst() + else: + msg.nope_ind = True + msg.mod_type = None + msg.tsc_set = None + msg.tsc = None + + # Encode a given message to bytes + msg_enc = msg.gen_msg(legacy) + + # Decode a new message from bytes + msg_dec = msg.__class__() + msg_dec.parse_msg(msg_enc) + + # Compare decoded vs the original + self._compare_msg(msg, msg_dec) + + # Validate encoding and decoding + def test_enc_dec(self): + for ver in DATAMSG.known_versions: + with self.subTest("L1 -> TRX message", ver = ver): + msg = DATAMSG_L12TRX(ver = ver) + self._test_enc_dec(msg) + + with self.subTest("TRX -> L1 message", ver = ver): + msg = DATAMSG_TRX2L1(ver = ver) + self._test_enc_dec(msg) + + if ver >= 1: + with self.subTest("TRX -> L1 NOPE.ind", ver = ver): + msg = DATAMSG_TRX2L1(ver = ver) + self._test_enc_dec(msg, nope_ind = True) + + with self.subTest("TRX -> L1 message (legacy)"): + msg = DATAMSG_TRX2L1(ver = 0) + self._test_enc_dec(msg, legacy = True) + + # Validate bit conversations + def test_bit_conv(self): + usbits_ref = list(range(0, 256)) + sbits_ref = list(range(-127, 128)) + + # Test both usbit2sbit() and sbit2usbit() + sbits = DATAMSG.usbit2sbit(usbits_ref) + usbits = DATAMSG.sbit2usbit(sbits) + self.assertEqual(usbits[:255], usbits_ref[:255]) + self.assertEqual(usbits[255], 254) + + # Test both sbit2ubit() and ubit2sbit() + ubits = DATAMSG.sbit2ubit(sbits_ref) + self.assertEqual(ubits, ([1] * 127 + [0] * 128)) + + sbits = DATAMSG.ubit2sbit(ubits) + self.assertEqual(sbits, ([-127] * 127 + [127] * 128)) + + def _test_transform(self, msg): + # Prepare given messages + msg.rand_hdr() + msg.rand_burst() + + # Perform message transformation + if isinstance(msg, DATAMSG_L12TRX): + msg_trans = msg.gen_trx2l1() + else: + msg_trans = msg.gen_l12trx() + + self.assertEqual(msg_trans.ver, msg.ver) + self.assertEqual(msg_trans.fn, msg.fn) + self.assertEqual(msg_trans.tn, msg.tn) + + if isinstance(msg, DATAMSG_TRX2L1): + burst = DATAMSG.sbit2ubit(msg.burst) + self.assertEqual(msg_trans.burst, burst) + else: + burst = DATAMSG.ubit2sbit(msg.burst) + self.assertEqual(msg_trans.burst, burst) + + # Validate message transformation + def test_transform(self): + for ver in DATAMSG.known_versions: + with self.subTest("L1 -> TRX message", ver = ver): + msg = DATAMSG_L12TRX(ver = ver) + self._test_transform(msg) + + with self.subTest("TRX -> L1 message", ver = ver): + msg = DATAMSG_TRX2L1(ver = ver) + self._test_transform(msg) + +if __name__ == '__main__': + unittest.main() |