eblot/pybootd
1
1
Fork 0
mirror of https://github.com/eblot/pybootd.git synced 2024-10-05 19:49:04 +03:00
Code Issues Projects Releases Wiki Activity

Code cleanup

Browse Source
This commit is contained in:
Emmanuel Blot 2020-04-28 13:57:33 +02:00
parent c904b04644
commit 0a10b95b33
3 changed files with 24 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pybootd/pxed.py
View File

@ -52,7 +52,6 @@ from .util import hexline, to_bool, iptoint, inttoip, get_iface_config
BOOTP_PORT_REQUEST = 67
BOOTP_PORT_REPLY = 68
BOOTREQUEST = 1
BOOTREPLY = 2
@ -858,8 +857,6 @@ class BootpServer:
self.log.info('Force global broadcast for buggy client %s',
mac_str)
addr = ('255.255.255.255', addr[1])
else:
self.log.debug('Not buggy')
# send the response
sock.sendto(pkt, addr)

32
pybootd/tests/simple.py
View File

@ -25,8 +25,9 @@
from binascii import hexlify, unhexlify
from io import StringIO
from socket import (socket, timeout, AF_INET, SOCK_DGRAM, IPPROTO_UDP,
SOL_SOCKET, SO_BROADCAST, SO_REUSEADDR)
from socket import (if_nametoindex, socket, timeout, AF_INET, SOCK_DGRAM,
IPPROTO_IP, IPPROTO_UDP, SOL_SOCKET, SO_BROADCAST,
SO_REUSEADDR)
from sys import modules
from textwrap import fill
from time import sleep
@ -45,8 +46,13 @@ class PxeSimpleTestCase(TestCase):
level='DEBUG')
cfgparser = EasyConfigParser()
config = StringIO("""
[logger]
type = stdout
level = debug
[bootpd]
address = 0.0.0.0
port = 67
pool_start = 127.0.0.100
pool_count = 5
servername = localhost
@ -65,6 +71,7 @@ default = pxelinux.0
""")
cfgparser.read_file(config)
cls.config = cfgparser
cls.server_port = int(cls.config.get('bootpd', 'port', '67'))
def setUp(self):
self.server = BootpDaemon(logger=self.logger, config=self.config,
@ -73,7 +80,7 @@ default = pxelinux.0
self.sock.setsockopt(SOL_SOCKET, SO_BROADCAST, 1)
self.sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
self.sock.settimeout(1)
self.sock.bind(('', 68))
self.sock.bind(('', self.server_port+1))
self.server.start()
# be sure the server can be scheduled and started before resuming
sleep(0.1)
@ -117,17 +124,22 @@ default = pxelinux.0
00000000000000000000000000000000000000000000000000000000000000000000
00000000
"""
for hexdata in (discover, request):
for step, hexdata in enumerate((discover, request), start=1):
req = unhexlify(hexdata.replace(' ', '').replace('\n', ''))
self.sock.sendto(req, ('<broadcast>', 67))
address = '<broadcast>'
self.sock.sendto(req, (address, self.server_port))
try:
fail = False
resp = self.sock.recv(1024)
# print('response:\n',
# fill(hexlify(resp).decode(),
# initial_indent=' ',
# subsequent_indent=' '))
self.logger.debug('response:\n%s',
fill(hexlify(resp).decode(),
initial_indent=' ',
subsequent_indent=' '))
except timeout:
self.assertFalse(True, 'No response from response')
fail = True
if fail:
self.assertFalse(True,
'No response from server @ step %s' % step)
def suite():
suite_ = TestSuite()

4
pybootd/util.py
View File

@ -125,7 +125,7 @@ def hexline(data, sep=' '):
def logger_factory(logtype='syslog', logfile=None, level='WARNING',
logid='PXEd', format=None):
logid='PXE', format=None):
# this code has been copied from Trac (MIT modified license)
logger = getLogger(logid)
logtype = logtype.lower()
@ -144,7 +144,7 @@ def logger_factory(logtype='syslog', logfile=None, level='WARNING',
hdlr = BufferingHandler(0)
if not format:
format = 'PXEd[%(module)s] %(levelname)s: %(message)s'
format = 'PXE[%(module)s] %(levelname)s: %(message)s'
if logtype in ('file', 'stderr'):
format = '%(asctime)s ' + format
datefmt = ''