diff options
author | Roman Khassraf <roman@khassraf.at> | 2015-08-05 12:51:33 +0200 |
---|---|---|
committer | Roman Khassraf <roman@khassraf.at> | 2015-08-05 12:51:33 +0200 |
commit | 64c74463ba818857138ad9b1286d41866d5ee7f2 (patch) | |
tree | 0dcde6a742d005004ba61de31a317f7784bd8fe0 /python | |
parent | 1fc0565aa9b0b02ab6cbcb9dcb7d7524bdceddd9 (diff) | |
parent | 1ce4469fd20a466984607ce0d075cb213e9c86f9 (diff) |
Merged origin/test-utils-98 into dev, fixed conflicts
Diffstat (limited to 'python')
-rw-r--r-- | python/CMakeLists.txt | 1 | ||||
-rwxr-xr-x | python/qa_decryption.py | 16 | ||||
-rwxr-xr-x | python/qa_message_printer.py | 166 |
3 files changed, 175 insertions, 8 deletions
diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt index 47f4bfe..e5190c8 100644 --- a/python/CMakeLists.txt +++ b/python/CMakeLists.txt @@ -48,5 +48,6 @@ set(GR_TEST_TARGET_DEPS gr-gsm) set(GR_TEST_PYTHON_DIRS ${CMAKE_BINARY_DIR}/swig) GR_ADD_TEST(qa_decryption ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_decryption.py) GR_ADD_TEST(qa_burst_printer ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_burst_printer.py) +GR_ADD_TEST(qa_message_printer ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_message_printer.py) #GR_ADD_TEST(qa_receiver ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_receiver.py) #GR_ADD_TEST(qa_receiver_hier ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/qa_receiver_hier.py) diff --git a/python/qa_decryption.py b/python/qa_decryption.py index cf228ae..08f0915 100755 --- a/python/qa_decryption.py +++ b/python/qa_decryption.py @@ -56,9 +56,9 @@ class qa_decryption (gr_unittest.TestCase): key = [0x32,0xE5,0x45,0x53,0x20,0x8C,0xE0,0x00] a5_version = 1 - src = grgsm.burst_source_qa(framenumbers_input, timeslots_input, bursts_input) + src = grgsm.burst_source(framenumbers_input, timeslots_input, bursts_input) decryption = grgsm.decryption((key), a5_version) - dst = grgsm.burst_sink_qa() + dst = grgsm.burst_sink() self.tb.msg_connect(src, "out", decryption, "bursts") self.tb.msg_connect(decryption, "bursts", dst, "in") @@ -97,9 +97,9 @@ class qa_decryption (gr_unittest.TestCase): key = [0xAD,0x6A,0x3E,0xC2,0xB4,0x42,0xE4,0x00] a5_version = 1 - src = grgsm.burst_source_qa(framenumbers_input, timeslots_input, bursts_input) + src = grgsm.burst_source(framenumbers_input, timeslots_input, bursts_input) decryption = grgsm.decryption((key), a5_version) - dst = grgsm.burst_sink_qa() + dst = grgsm.burst_sink() self.tb.msg_connect(src, "out", decryption, "bursts") self.tb.msg_connect(decryption, "bursts", dst, "in") @@ -138,9 +138,9 @@ class qa_decryption (gr_unittest.TestCase): key = [0x41,0xBC,0x19,0x30,0xB6,0x31,0x8A,0xC8] a5_version = 3 - src = grgsm.burst_source_qa(framenumbers_input, timeslots_input, bursts_input) + src = grgsm.burst_source(framenumbers_input, timeslots_input, bursts_input) decryption = grgsm.decryption((key), a5_version) - dst = grgsm.burst_sink_qa() + dst = grgsm.burst_sink() self.tb.msg_connect(src, "out", decryption, "bursts") self.tb.msg_connect(decryption, "bursts", dst, "in") @@ -179,9 +179,9 @@ class qa_decryption (gr_unittest.TestCase): key = [0xAD,0x2C,0xB3,0x83,0x2F,0x4A,0x6C,0xF1] a5_version = 3 - src = grgsm.burst_source_qa(framenumbers_input, timeslots_input, bursts_input) + src = grgsm.burst_source(framenumbers_input, timeslots_input, bursts_input) decryption = grgsm.decryption((key), a5_version) - dst = grgsm.burst_sink_qa() + dst = grgsm.burst_sink() self.tb.msg_connect(src, "out", decryption, "bursts") self.tb.msg_connect(decryption, "bursts", dst, "in") diff --git a/python/qa_message_printer.py b/python/qa_message_printer.py new file mode 100755 index 0000000..4efaaef --- /dev/null +++ b/python/qa_message_printer.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @file +# @author Roman Khassraf <rkhassraf@gmail.com> +# @section LICENSE +# +# Gr-gsm 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 3, or (at your option) +# any later version. +# +# Gr-gsm 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 gr-gsm; see the file COPYING. If not, write to +# the Free Software Foundation, Inc., 51 Franklin Street, +# Boston, MA 02110-1301, USA. +# +# + +from gnuradio import gr, gr_unittest, blocks +import grgsm +import os +import pmt +import sys +import tempfile + +class qa_message_printer (gr_unittest.TestCase): + + def setUp (self): + self.tb = gr.top_block() + self.tmpfile = tempfile.NamedTemporaryFile() + self.prevfd = os.dup(sys.stdout.fileno()) + os.dup2(self.tmpfile.fileno(), sys.stdout.fileno()) + self.prev = sys.stdout + sys.stdout = os.fdopen(self.prevfd, "w") + + def tearDown (self): + self.tb = None + os.dup2(self.prevfd, self.prev.fileno()) + sys.stdout = self.prev + self.tmpfile.close() + + def getOutput(self): + self.tmpfile.seek(0) + return self.tmpfile.read() + + def getOutputExpected(self, expected_lines): + out = "" + for l in expected_lines: + out = out + l + "\n" + return out + + def test_001_no_prefix_no_header (self): + """ + Four messages, without any prefix, no gsmtap header + """ + msgs_input = [ + "02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + # there is a whitespace at the beginning of message_printer output + msgs_expected = [ + " 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + " 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + " 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + " 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + src = grgsm.message_source(msgs_input) + printer = grgsm.message_printer(pmt.intern(""), False) + self.tb.msg_connect(src, "msgs", printer, "msgs") + self.tb.run() + + self.assertEqual(self.getOutput(), self.getOutputExpected(msgs_expected)) + + + def test_002_prefix_no_header (self): + """ + Four messages, with prefix "test_002:", no gsmtap header + """ + msgs_input = [ + "02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + msgs_expected = [ + "test_002: 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "test_002: 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "test_002: 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "test_002: 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + src = grgsm.message_source(msgs_input) + printer = grgsm.message_printer(pmt.intern("test_002:"), False) + self.tb.msg_connect(src, "msgs", printer, "msgs") + self.tb.run() + + self.assertEqual(self.getOutput(), self.getOutputExpected(msgs_expected)) + + + def test_003_no_prefix_header (self): + """ + Four messages, without any prefix, with gsmtap header + """ + msgs_input = [ + "02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + # there is a whitespace at the beginning of message_printer output + msgs_expected = [ + " 02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + " 02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + " 02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + " 02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + src = grgsm.message_source(msgs_input) + printer = grgsm.message_printer(pmt.intern(""), True) + self.tb.msg_connect(src, "msgs", printer, "msgs") + self.tb.run() + + self.assertEqual(self.getOutput(), self.getOutputExpected(msgs_expected)) + + + def test_004_prefix_header (self): + """ + Four messages, with prefix "test_004:", with gsmtap header + """ + msgs_input = [ + "02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + # there is a whitespace at the beginning of message_printer output + msgs_expected = [ + "test_004: 02 04 01 00 00 00 c9 00 00 1d 3c e5 02 00 01 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "test_004: 02 04 01 00 00 00 ca 00 00 1d 3c e9 02 00 02 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b", + "test_004: 02 04 01 00 00 00 cb 00 00 1d 3d 0e 01 00 00 00 59 06 1a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ff e5 04 00", + "test_004: 02 04 01 00 00 00 cb 00 00 1d 3d 12 02 00 00 00 15 06 21 00 01 f0 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b 2b" + ] + + src = grgsm.message_source(msgs_input) + printer = grgsm.message_printer(pmt.intern("test_004:"), True) + self.tb.msg_connect(src, "msgs", printer, "msgs") + self.tb.run() + + self.assertEqual(self.getOutput(), self.getOutputExpected(msgs_expected)) + + +if __name__ == '__main__': + gr_unittest.run(qa_message_printer, "qa_message_printer.xml") + |