testing the message parsing and template data structures. Made some changes due to the tests
This commit is contained in:
locklainn.linden
committed by
Salad Dais
Salad Dais
parent
8d80bf3c10
commit
034de0a1a9
@@ -67,7 +67,7 @@ class MessageTemplateBlock():
|
||||
self.name = header[0]
|
||||
self.blockType = header[1]
|
||||
if self.blockType == 'Multiple':
|
||||
self.number = header[2]
|
||||
self.number = int(header[2])
|
||||
else:
|
||||
self.number = 0
|
||||
|
||||
@@ -80,7 +80,7 @@ class MessageTemplateBlock():
|
||||
def get_name(self):
|
||||
return self.name
|
||||
|
||||
def add_var(self, var):
|
||||
def add_variable(self, var):
|
||||
self.vars[var.get_name()] = var
|
||||
|
||||
def get_variables(self):
|
||||
@@ -99,15 +99,15 @@ class MessageTemplate():
|
||||
self.msg_num = string.atoi(header[2],0)
|
||||
if self.frequency == 'Fixed':
|
||||
#have to do this because Fixed messages are stored as a long in the template
|
||||
binTemp = struct.pack('>l', string.atol(header[2],0))
|
||||
self.msg_num_hex = repr(binTemp)
|
||||
binTemp = struct.pack('>L', string.atol(header[2],0))
|
||||
self.msg_num_hex = binTemp
|
||||
self.msg_num = struct.unpack('>h','\x00' + binTemp[3])[0]
|
||||
elif self.frequency == 'Low':
|
||||
self.msg_num_hex = repr(struct.pack('>bbh',0xff,0xff, self.msg_num))
|
||||
self.msg_num_hex = struct.pack('>BBh',0xff,0xff, self.msg_num)
|
||||
elif self.frequency == 'Medium':
|
||||
self.msg_num_hex = repr(struct.pack('>bb',0xff, self.msg_num))
|
||||
self.msg_num_hex = struct.pack('>BB',0xff, self.msg_num)
|
||||
elif self.frequency == 'High':
|
||||
self.msg_num_hex = repr(struct.pack('>b', self.msg_num))
|
||||
self.msg_num_hex = struct.pack('>B', self.msg_num)
|
||||
|
||||
self.msg_trust = header[3]
|
||||
self.msg_encoding = header[4]
|
||||
@@ -125,10 +125,10 @@ class MessageTemplate():
|
||||
def get_message_hex_num(self):
|
||||
return self.msg_num_hex
|
||||
|
||||
def get_message_trust(self):
|
||||
def get_trust(self):
|
||||
return self.msg_trust
|
||||
|
||||
def get_message_encoding(self):
|
||||
def get_encoding(self):
|
||||
return self.msg_encoding
|
||||
|
||||
def get_deprecation(self):
|
||||
|
||||
@@ -7,9 +7,12 @@ import message_template
|
||||
from pyogp.lib.base.data import msg_tmpl
|
||||
|
||||
class MessageTemplateParser():
|
||||
def __init__(self):
|
||||
def __init__(self, template_file):
|
||||
self.message_templates = []
|
||||
self.version = ''
|
||||
self.template_file = template_file
|
||||
self.count = 0
|
||||
self.__parse_template_file()
|
||||
|
||||
def get_version(self):
|
||||
return self.version
|
||||
@@ -17,17 +20,18 @@ class MessageTemplateParser():
|
||||
def get_template_list(self):
|
||||
return self.message_templates
|
||||
|
||||
def get_template(self, name):
|
||||
return self.message_templates[name]
|
||||
def get_count(self):
|
||||
return self.count
|
||||
|
||||
def add_template(self, new_template):
|
||||
def __add_template(self, new_template):
|
||||
self.count += 1
|
||||
#self.message_templates[new_template.get_name()] = new_template
|
||||
self.message_templates.append(new_template)
|
||||
|
||||
def parse_template_file(self, template_file):
|
||||
def __parse_template_file(self):
|
||||
count = 0
|
||||
template_file.seek(0)
|
||||
lines = template_file
|
||||
self.template_file.seek(0)
|
||||
lines = self.template_file
|
||||
#results = re.match("^\t([^\t{}]+.+)",line) #gets packet headers
|
||||
#results = re.match("^\t\t([^{}]+.+)",line) #gets packet blocks
|
||||
#results = re.match("^\t\t([{}]+.+)",line) #gets block data
|
||||
@@ -35,8 +39,6 @@ class MessageTemplateParser():
|
||||
current_packet = None
|
||||
current_block = None
|
||||
|
||||
print lines
|
||||
|
||||
#we have to go through all the packets and parse them
|
||||
while(True):
|
||||
try:
|
||||
@@ -61,7 +63,7 @@ class MessageTemplateParser():
|
||||
parts = parts.split()
|
||||
|
||||
current_packet = message_template.MessageTemplate(parts)
|
||||
self.add_template(current_packet)
|
||||
self.__add_template(current_packet)
|
||||
|
||||
block_header = re.match("^\t\t([^{}]+.+)",line) #gets packet block header
|
||||
if block_header != None:
|
||||
@@ -78,7 +80,9 @@ class MessageTemplateParser():
|
||||
parts.remove('{')
|
||||
parts.remove('}')
|
||||
current_var = message_template.MessageTemplateVariable(parts[0], parts[1])
|
||||
current_block.add_var(current_var)
|
||||
current_block.add_variable(current_var)
|
||||
|
||||
self.template_file.seek(0)
|
||||
|
||||
def print_packet_list(packet_list):
|
||||
for packet in packet_list:
|
||||
|
||||
135
pyogp/lib/base/tests/test_packets.py
Normal file
135
pyogp/lib/base/tests/test_packets.py
Normal file
@@ -0,0 +1,135 @@
|
||||
#standard libraries
|
||||
import unittest, doctest
|
||||
|
||||
#local libraries
|
||||
from pyogp.lib.base.data import msg_tmpl
|
||||
from pyogp.lib.base.message_template import MessageTemplate, MessageTemplateBlock, MessageTemplateVariable
|
||||
from pyogp.lib.base.message_template_parser import MessageTemplateParser
|
||||
|
||||
|
||||
class TestTemplates(unittest.TestCase):
|
||||
|
||||
def tearDown(self):
|
||||
pass
|
||||
|
||||
def setUp(self):
|
||||
self.template_file = msg_tmpl
|
||||
|
||||
def test_parser(self):
|
||||
parser = MessageTemplateParser(self.template_file)
|
||||
|
||||
|
||||
def test_parser_version(self):
|
||||
parser = MessageTemplateParser(self.template_file)
|
||||
version = parser.get_version()
|
||||
assert version == 2.0, "Version not correct, expected 2.0, got " + str(version)
|
||||
|
||||
def test_template(self):
|
||||
template = MessageTemplate(('CompletePingCheck', 'High', '2', 'NotTrusted', 'Unencoded'))
|
||||
name = template.get_name()
|
||||
freq = template.get_frequency()
|
||||
num = template.get_message_number()
|
||||
hx = template.get_message_hex_num()
|
||||
trust = template.get_trust()
|
||||
enc = template.get_encoding()
|
||||
dep = template.get_deprecation()
|
||||
assert name == 'CompletePingCheck', "Expected: CompletePingCheck Returned: " + name
|
||||
assert freq == 'High', "Expected: High Returned: " + freq
|
||||
assert num == 2, "Expected: 2 Returned: " + str(num)
|
||||
assert hx == '\x02', "Expected: '\x02' Returned: " + repr(hx)
|
||||
assert trust == 'NotTrusted', "Expected: NotTrusted Returned: " + trust
|
||||
assert enc == 'Unencoded', "Expected: Unencoded Returned: " + enc
|
||||
assert dep == '', "Expected: Returned: " + dep
|
||||
|
||||
def test_template_low_deprecated(self):
|
||||
template = MessageTemplate(('CompletePingCheck', 'Low', '2', 'NotTrusted', 'Unencoded', 'Deprecated'))
|
||||
hx = template.get_message_hex_num()
|
||||
dep = template.get_deprecation()
|
||||
assert hx == '\xff\xff\x00\x02', "Expected: '\xff\xff\x00\x02' Returned: " + repr(hx)
|
||||
assert dep == 'Deprecated', "Expected: Deprecated Returned: " + dep
|
||||
|
||||
def test_template_medium(self):
|
||||
template = MessageTemplate(('CompletePingCheck', 'Medium', '2', 'NotTrusted', 'Unencoded', 'Deprecated'))
|
||||
hx = template.get_message_hex_num()
|
||||
dep = template.get_deprecation()
|
||||
assert hx == '\xff\x02', "Expected: " + r'\xff\x02' + " Returned: " + hx
|
||||
|
||||
def test_template_fixed(self):
|
||||
template = MessageTemplate(('CompletePingCheck', 'Fixed', '0xFFFFFFFB', 'NotTrusted', 'Unencoded'))
|
||||
num = template.get_message_number()
|
||||
hx = template.get_message_hex_num()
|
||||
assert num == 251, "Expected: 251 Returned: " + str(num)
|
||||
assert hx == '\xff\xff\xff\xfb', "Expected: " + r'\xff\xff\xff\xfb' + " Returned: " + repr(hx)
|
||||
|
||||
def test_block(self):
|
||||
block = MessageTemplateBlock(('PingID', 'Single'))
|
||||
name = block.get_name()
|
||||
tp = block.get_block_type()
|
||||
num = block.get_block_number()
|
||||
assert name == 'PingID', "Expected: PingID Returned" + name
|
||||
assert tp == 'Single', "Expected: Single Returned: " + tp
|
||||
assert num == 0, "Expected: 0 Returned: " + str(num)
|
||||
|
||||
def test_block_multiple(self):
|
||||
block = MessageTemplateBlock(('CircuitCode', 'Multiple', '2'))
|
||||
name = block.get_name()
|
||||
tp = block.get_block_type()
|
||||
num = block.get_block_number()
|
||||
assert name == 'CircuitCode', "Expected: CircuitCode Returned" + name
|
||||
assert tp == 'Multiple', "Expected: Multiple Returned: " + tp
|
||||
assert num == 2, "Expected: 2 Returned: " + str(num)
|
||||
|
||||
def test_variable(self):
|
||||
variable = MessageTemplateVariable("PingID", "U8")
|
||||
name = variable.get_name()
|
||||
tp = variable.get_type()
|
||||
assert name == 'PingID', "Expected: PingID Returned: " + name
|
||||
assert tp == 'U8', "Expected: U8 Returned: " + tp
|
||||
|
||||
def test_add_get_block(self):
|
||||
template = MessageTemplate(('CompletePingCheck', 'High', '2', 'NotTrusted', 'Unencoded'))
|
||||
block = MessageTemplateBlock(('CircuitCode', 'Multiple', '2'))
|
||||
template.add_block(block)
|
||||
blocks = template.get_blocks()
|
||||
count = len(blocks)
|
||||
assert blocks[0].get_name() == 'CircuitCode', "Add block failed"
|
||||
assert template.get_block('CircuitCode') != None, "Get block failed"
|
||||
|
||||
def test_add_variable(self):
|
||||
block = MessageTemplateBlock(('CircuitCode', 'Multiple', '2'))
|
||||
variable = MessageTemplateVariable("PingID", "U8")
|
||||
block.add_variable(variable)
|
||||
var_list = block.get_variables()
|
||||
assert var_list[0].get_name() == 'PingID', "Add variable failed"
|
||||
assert block.get_variable('PingID') != None, "Get variable failed"
|
||||
|
||||
def test_StartPingCheck(self):
|
||||
parser = MessageTemplateParser(self.template_file)
|
||||
template_list = parser.get_template_list()
|
||||
my_template = template_list[4]
|
||||
name = my_template.get_name()
|
||||
frequency = my_template.get_frequency()
|
||||
num = my_template.get_message_number()
|
||||
hx = my_template.get_message_hex_num()
|
||||
trust = my_template.get_trust()
|
||||
code = my_template.get_encoding()
|
||||
deprecate = my_template.get_deprecation()
|
||||
block_count = len(my_template.get_blocks())
|
||||
my_block = my_template.get_block('PingID')
|
||||
p_id = my_block.get_variable('PingID')
|
||||
assert name == 'StartPingCheck', "Expected: StartPingCheck Returned: " + name
|
||||
assert frequency == 'High', "Expected: High Returned: " + frequency
|
||||
assert num == 1, "Expected: 1 Returned: " + str(num)
|
||||
assert hx == '\x01', "Expected: " + r'\x01' + " Returned: " + repr(hx)
|
||||
assert trust == 'NotTrusted', "Expected: NotTrusted Returned: " + trust
|
||||
assert code == 'Unencoded', "Expected: Unencoded Returned: " + code
|
||||
assert deprecate == '', "Expected: Returned: " + str(deprecate)
|
||||
assert block_count == 1, "Expected: 1 Returned: " + str(block_count)
|
||||
assert my_block != None, "Getting block PingID failed"
|
||||
assert p_id != None, "Getting variable PingID failed"
|
||||
|
||||
def test_suite():
|
||||
from unittest import TestSuite, makeSuite
|
||||
suite = TestSuite()
|
||||
suite.addTest(makeSuite(TestTemplates))
|
||||
return suite
|
||||
Reference in New Issue
Block a user