Commit e60af9fa authored by Tomek Mrugalski's avatar Tomek Mrugalski 🛰
Browse files

[1503] -p portnumber option implemented for b10-dhcp6

 -p portnumber command-line option implemeneted for b10-dhcp6
 Corresponding tests implemented as well.
parent b35d3159
......@@ -62,8 +62,8 @@ usage() {
int
main(int argc, char* argv[]) {
int ch;
int port_number = DHCP4_SERVER_PORT; // the default. any other values are
// useful for testing only
int port_number = DHCP4_SERVER_PORT; // The default. any other values are
// useful for testing only.
while ((ch = getopt(argc, argv, "vp:")) != -1) {
switch (ch) {
......
# Copyright (C) 2011 Internet Systems Consortium.
# Copyright (C) 2012 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
......@@ -27,7 +27,6 @@ import fcntl
class TestDhcpv4Daemon(unittest.TestCase):
def setUp(self):
# don't redirect stdout/stderr here as we want to print out things
# during the test
pass
......@@ -155,7 +154,7 @@ class TestDhcpv4Daemon(unittest.TestCase):
self.assertEqual( str(error).count("option requires an argument"), 1)
def test_portnumber_nonroot(self):
print("Check that specifying unprivilidged port number will work.")
print("Check that specifying unprivileged port number will work.")
(returncode, output, error) = self.runDhcp4(['../b10-dhcp4', '-p', '10057'])
......
......@@ -40,7 +40,7 @@ const uint32_t HARDCODED_VALID_LIFETIME = 7200; // in seconds
const std::string HARDCODED_DNS_SERVER = "2001:db8:1::1";
Dhcpv6Srv::Dhcpv6Srv(uint16_t port) {
cout << "Initialization" << endl;
cout << "Initialization: opening sockets on port " << port << endl;
// first call to instance() will create IfaceMgr (it's a singleton)
// it may throw something if things go wrong
......
// Copyright (C) 2009-2011 Internet Systems Consortium, Inc. ("ISC")
// Copyright (C) 2011-2012 Internet Systems Consortium, Inc. ("ISC")
//
// Permission to use, copy, modify, and/or distribute this software for any
// purpose with or without fee is hereby granted, provided that the above
......@@ -53,20 +53,31 @@ usage() {
cerr << "Usage: b10-dhcp6 [-v]"
<< endl;
cerr << "\t-v: verbose output" << endl;
exit(1);
cerr << "\t-p number: specify non-standard port number 1-65535 (useful for testing only)" << endl;
exit(EXIT_FAILURE);
}
} // end of anonymous namespace
int
main(int argc, char* argv[]) {
int ch;
int port_number = DHCP6_SERVER_PORT; // The default. Any other values are
// useful for testing only.
while ((ch = getopt(argc, argv, ":v")) != -1) {
while ((ch = getopt(argc, argv, "vp:")) != -1) {
switch (ch) {
case 'v':
verbose_mode = true;
isc::log::denabled = true;
break;
case 'p':
port_number = strtol(optarg, NULL, 10);
if (port_number == 0) {
cerr << "Failed to parse port number: [" << optarg
<< "], 1-65535 allowed." << endl;
usage();
}
break;
case ':':
default:
usage();
......@@ -79,7 +90,7 @@ main(int argc, char* argv[]) {
usage();
}
int ret = 0;
int ret = EXIT_SUCCESS;
// TODO remainder of auth to dhcp6 code copy. We need to enable this in
// dhcp6 eventually
......@@ -99,13 +110,13 @@ main(int argc, char* argv[]) {
cout << "[b10-dhcp6] Initiating DHCPv6 operation." << endl;
Dhcpv6Srv* srv = new Dhcpv6Srv();
Dhcpv6Srv* srv = new Dhcpv6Srv(port_number);
srv->run();
} catch (const std::exception& ex) {
cerr << "[b10-dhcp6] Server failed: " << ex.what() << endl;
ret = 1;
ret = EXIT_FAILURE;
}
return (ret);
......
# Copyright (C) 2011 Internet Systems Consortium.
# Copyright (C) 2011,2012 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
......@@ -23,52 +23,150 @@ import socket
from isc.net.addr import IPAddr
import time
import isc
import fcntl
class TestDhcpv6Daemon(unittest.TestCase):
def setUp(self):
# don't redirect stdout/stderr here as we want to print out things
# during the test
pass
def tearDown(self):
pass
def runCommand(self, params, wait=1):
"""
This method runs a command and returns a touple: (returncode, stdout, stderr)
"""
## @todo: Convert this into generic method and reuse it in dhcp4 and dhcp6
print("Running command: %s" % (" ".join(params)))
# redirect stdout to a pipe so we can check that our
# process spawning is doing the right thing with stdout
self.old_stdout = os.dup(sys.stdout.fileno())
self.pipes = os.pipe()
os.dup2(self.pipes[1], sys.stdout.fileno())
os.close(self.pipes[1])
self.stdout_old = os.dup(sys.stdout.fileno())
self.stdout_pipes = os.pipe()
os.dup2(self.stdout_pipes[1], sys.stdout.fileno())
os.close(self.stdout_pipes[1])
# do the same trick for stderr:
self.stderr_old = os.dup(sys.stderr.fileno())
self.stderr_pipes = os.pipe()
os.dup2(self.stderr_pipes[1], sys.stderr.fileno())
os.close(self.stderr_pipes[1])
# note that we use dup2() to restore the original stdout
# to the main program ASAP in each test... this prevents
# hangs reading from the child process (as the pipe is only
# open in the child), and also insures nice pretty output
print("Please ignore any socket errors. Purpose of this test is to")
print("verify that DHCPv6 process could be started, not that socket")
print("could be bound. Binding fails when run as non-root user.")
def tearDown(self):
# clean up our stdout munging
os.dup2(self.old_stdout, sys.stdout.fileno())
os.close(self.pipes[0])
pi = ProcessInfo('Test Process', params)
pi.spawn()
time.sleep(wait)
os.dup2(self.stdout_old, sys.stdout.fileno())
os.dup2(self.stderr_old, sys.stderr.fileno())
self.assertNotEqual(pi.process, None)
self.assertTrue(type(pi.pid) is int)
# Set non-blocking read on pipes. Process may not print anything
# on specific output and the we would hang without this.
fd = self.stdout_pipes[0]
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
fd = self.stderr_pipes[0]
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
# There's potential problem if b10-dhcp4 prints out more
# than 4k of text
try:
output = os.read(self.stdout_pipes[0], 4096)
except OSError:
print("No data available from stdout")
output = ""
# read can return None. Make sure we have a string
if (output is None):
output = ""
try:
error = os.read(self.stderr_pipes[0], 4096)
except OSError:
print("No data available on stderr")
error = ""
# read can return None. Make sure we have a string
if (error is None):
error = ""
try:
if (not pi.process.poll()):
# let's be nice at first...
pi.process.terminate()
except OSError:
print("Ignoring failed kill attempt. Process is dead already.")
# call this to get returncode, process should be dead by now
rc = pi.process.wait()
# Clean up our stdout/stderr munging.
os.dup2(self.stdout_old, sys.stdout.fileno())
os.close(self.stdout_pipes[0])
os.dup2(self.stderr_old, sys.stderr.fileno())
os.close(self.stderr_pipes[0])
print ("Process finished, return code=%d, stdout=%d bytes, stderr=%d bytes"
% (rc, len(output), len(error)) )
return (rc, output, error)
def test_alive(self):
"""
Simple test. Checks that b10-dhcp6 can be started and prints out info
about starting DHCPv6 operation.
"""
pi = ProcessInfo('Test Process', [ '../b10-dhcp6' , '-v' ])
pi.spawn()
time.sleep(1)
os.dup2(self.old_stdout, sys.stdout.fileno())
self.assertNotEqual(pi.process, None)
self.assertTrue(type(pi.pid) is int)
output = os.read(self.pipes[0], 4096)
print("Note: Purpose of some of the tests is to check if DHCPv6 server can be started,")
print(" not that is can bind sockets correctly. Please ignore binding errors.")
(returncode, output, error) = self.runCommand(["../b10-dhcp6", "-v"])
self.assertEqual( str(output).count("[b10-dhcp6] Initiating DHCPv6 operation."), 1)
# kill this process
# XXX: b10-dhcp6 is too dumb to understand 'shutdown' command for now,
# so let's just kill the bastard
def test_portnumber_0(self):
print("Check that specifying port number 0 is not allowed.")
# TODO: Ignore errors for now. This test will be more thorough once ticket #1503
# (passing port number to b10-dhcp6 daemon) is implemented.
try:
os.kill(pi.pid, signal.SIGTERM)
except OSError:
print("Ignoring failed kill attempt. Process is dead already.")
(returncode, output, error) = self.runCommand(['../b10-dhcp6', '-p', '0'])
# When invalid port number is specified, return code must not be success
self.assertTrue(returncode != 0)
# Check that there is an error message about invalid port number printed on stderr
self.assertEqual( str(error).count("Failed to parse port number"), 1)
def test_portnumber_missing(self):
print("Check that -p option requires a parameter.")
(returncode, output, error) = self.runCommand(['../b10-dhcp6', '-p'])
# When invalid port number is specified, return code must not be success
self.assertTrue(returncode != 0)
# Check that there is an error message about invalid port number printed on stderr
self.assertEqual( str(error).count("option requires an argument"), 1)
def test_portnumber_nonroot(self):
print("Check that specifying unprivileged port number will work.")
(returncode, output, error) = self.runCommand(['../b10-dhcp6', '-p', '10057'])
# When invalid port number is specified, return code must not be success
# TODO: Temporarily commented out as socket binding on systems that do not have
# interface detection implemented currently fails.
# self.assertTrue(returncode == 0)
# Check that there is a message on stdout about opening proper port
self.assertEqual( str(output).count("opening sockets on port 10057"), 1)
if __name__ == '__main__':
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