Commit cc99c8a3 authored by Jelte Jansen's avatar Jelte Jansen

initial JSON usage on command channel in python.

to_wire simply returns an byte-encoded string representation of the objects in json format
from_wire translates that back to actual objects



git-svn-id: svn://bind10.isc.org/svn/bind10/branches/trac172@2104 e5f2f494-b856-4b98-b285-d166d9295462
parent 48867a1d
......@@ -272,41 +272,42 @@ class BoB:
if self.verbose:
sys.stdout.write("[bind10] ccsession started\n")
# start the xfrout before auth-server, to make sure every xfr-query can
# be processed properly.
xfrout_args = ['b10-xfrout']
if self.verbose:
sys.stdout.write("[bind10] Starting b10-xfrout\n")
xfrout_args += ['-v']
try:
xfrout = ProcessInfo("b10-xfrout", xfrout_args,
c_channel_env )
except Exception as e:
c_channel.process.kill()
bind_cfgd.process.kill()
return "Unable to start b10-xfrout; " + str(e)
self.processes[xfrout.pid] = xfrout
if self.verbose:
sys.stdout.write("[bind10] Started b10-xfrout (PID %d)\n" % xfrout.pid)
# start b10-auth
# XXX: this must be read from the configuration manager in the future
authargs = ['b10-auth', '-p', str(self.auth_port)]
if self.verbose:
sys.stdout.write("[bind10] Starting b10-auth using port %d\n" %
self.auth_port)
authargs += ['-v']
try:
auth = ProcessInfo("b10-auth", authargs,
c_channel_env)
except Exception as e:
c_channel.process.kill()
bind_cfgd.process.kill()
xfrout.process.kill()
return "Unable to start b10-auth; " + str(e)
self.processes[auth.pid] = auth
if self.verbose:
sys.stdout.write("[bind10] Started b10-auth (PID %d)\n" % auth.pid)
## TODO: UNCOMMENT (commented out, doing json for python first)
# # start the xfrout before auth-server, to make sure every xfr-query can
# # be processed properly.
# xfrout_args = ['b10-xfrout']
# if self.verbose:
# sys.stdout.write("[bind10] Starting b10-xfrout\n")
# xfrout_args += ['-v']
# try:
# xfrout = ProcessInfo("b10-xfrout", xfrout_args,
# c_channel_env )
# except Exception as e:
# c_channel.process.kill()
# bind_cfgd.process.kill()
# return "Unable to start b10-xfrout; " + str(e)
# self.processes[xfrout.pid] = xfrout
# if self.verbose:
# sys.stdout.write("[bind10] Started b10-xfrout (PID %d)\n" % xfrout.pid)
#
# # start b10-auth
# # XXX: this must be read from the configuration manager in the future
# authargs = ['b10-auth', '-p', str(self.auth_port)]
# if self.verbose:
# sys.stdout.write("[bind10] Starting b10-auth using port %d\n" %
# self.auth_port)
# authargs += ['-v']
# try:
# auth = ProcessInfo("b10-auth", authargs,
# c_channel_env)
# except Exception as e:
# c_channel.process.kill()
# bind_cfgd.process.kill()
# xfrout.process.kill()
# return "Unable to start b10-auth; " + str(e)
# self.processes[auth.pid] = auth
# if self.verbose:
# sys.stdout.write("[bind10] Started b10-auth (PID %d)\n" % auth.pid)
# start b10-xfrin
xfrin_args = ['b10-xfrin']
......
......@@ -35,7 +35,15 @@ _ITEM_LENGTH_16 = 0x10
_ITEM_LENGTH_8 = 0x20
_ITEM_LENGTH_MASK = 0x30
import json
def to_wire(items):
return json.dumps(items).encode('utf8')
def from_wire(data):
return json.loads(data.decode('utf8'))
def old_to_wire(items):
"""Encode a dict into wire format.
>>> wire_format = Message.to_wire({"a": "b"})
"""
......@@ -140,7 +148,7 @@ def _encode_hash(item):
# decode methods
#
def from_wire(data):
def old_from_wire(data):
if len(data) < 5:
raise DecodeError("Data is too short to decode")
wire_version, data = data[0:4], data[4:]
......
PYTESTS = data_test.py session_test.py test.py
PYTESTS = message_test.py
#data_test.py session_test.py test.py
# NOTE: test_session.py is to be run manually, so not automated.
EXTRA_DIST = $(PYTESTS)
EXTRA_DIST += sendcmd.py
......
# Copyright (C) 2010 Internet Systems Consortium.
#
# Permission to use, copy, modify, and 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 INTERNET SYSTEMS CONSORTIUM
# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# INTERNET SYSTEMS CONSORTIUM 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.
#
# Tests for the functions in data.py
#
import unittest
import isc.cc
#
# Umz, do we need anything more than abstraction from JSON?
# (i.e. a encode and decode function?)
#
class MessageTest(unittest.TestCase):
def setUp(self):
self.msg1 = { "just": [ "an", "arbitrary", "structure" ] }
self.msg1_str = "{\"just\": [\"an\", \"arbitrary\", \"structure\"]}";
self.msg1_wire = self.msg1_str.encode()
self.msg2 = { "aaa": [ 1, 1.1, True, False, None ] }
self.msg2_str = "{\"aaa\": [1, 1.1, true, false, null]}";
self.msg2_wire = self.msg2_str.encode()
def test_encode_json(self):
self.assertEqual(self.msg1_wire, isc.cc.message.to_wire(self.msg1))
self.assertEqual(self.msg2_wire, isc.cc.message.to_wire(self.msg2))
def test_decode_json(self):
self.assertEqual(self.msg1, isc.cc.message.from_wire(self.msg1_wire))
self.assertEqual(self.msg2, isc.cc.message.from_wire(self.msg2_wire))
if __name__ == '__main__':
#if not 'CONFIG_TESTDATA_PATH' in os.environ:
# print("You need to set the environment variable CONFIG_TESTDATA_PATH to point to the directory containing the test data files")
# exit(1)
unittest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment