Commit 15b4c0c8 authored by Rafael Monnerat's avatar Rafael Monnerat

promise.is_packet_lost: New Promise

  Added a promise to test if there is a packet lost when ping to an IPv4/IPv6.
parent 0c750595
......@@ -81,6 +81,7 @@ setup(name=name,
'htpasswd = slapos.htpasswd:main',
'is-local-tcp-port-opened = slapos.promise.is_local_tcp_port_opened:main',
'is-process-older-than-dependency-set = slapos.promise.is_process_older_than_dependency_set:main',
'is-icmp-packet-lost = slapos.promise.is_icmp_packet_lost:main',
'killpidfromfile = slapos.systool:killpidfromfile', # BBB
'monitor.bootstrap = slapos.monitor.monitor:main',
'monitor.collect = slapos.monitor.collect:main',
......
......@@ -13,7 +13,7 @@ ping_re = re.compile(
date_reg_exp = re.compile('\d{4}[-/]\d{2}[-/]\d{2}')
def ping(host, timeout=10, protocol="4"):
def ping(host, timeout=10, protocol="4", count=10):
if protocol == '4':
ping_bin = 'ping'
test_title = 'PING'
......@@ -21,7 +21,7 @@ def ping(host, timeout=10, protocol="4"):
ping_bin = 'ping6'
test_title = 'PING6'
proc = subprocess.Popen((ping_bin, '-c', '10', '-w', str(timeout), host),
proc = subprocess.Popen((ping_bin, '-c', str(count), '-w', str(timeout), host),
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
......@@ -46,6 +46,6 @@ def ping(host, timeout=10, protocol="4"):
return info_list
def ping6(host, timeout=10):
return ping(host, timeout=10, protocol='6')
def ping6(host, timeout=10, count=10):
return ping(host, timeout=10, protocol='6', count=count)
import argparse
import re
import time
import sys
from slapos.networkbench.ping import ping, ping6
def test(address, ipv4, count):
if ipv4:
return ping(address, count=count)
return ping6(address, count=count)
def main():
parser = argparse.ArgumentParser()
# Address to ping to
parser.add_argument("-a", "--address", required=True)
# Force use ipv4 protocol
parser.add_argument("-4", "--ipv4", action="store_true" )
parser.add_argument("-c", "--count", metavar="COUNT", default=10 )
args = parser.parse_args()
result = test(args.address, args.ipv4, args.count)
print "%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s" % result
if result[4] != "0":
# Packet lost occurred
print "FAIL"
sys.exit(1)
print "OK"
##############################################################################
#
# Copyright (c) 2017 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program 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
# of the License, or (at your option) any later version.
#
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import unittest
import os.path
import socket
import time
from slapos.promise.is_icmp_packet_lost import test
class TestIsICMPPacketLost(unittest.TestCase):
def test_localhost(self):
result = test("localhost", True, 5)
self.assertEquals(result[4], '0')
def test_error(self):
result = test("couscous", True, 5)
self.assertEquals(result[4], -1)
def test_localhost6 (self):
result = test("::1", False, 5)
self.assertEquals(result[4], '0')
def test_error6(self):
result = test("couscous", False, 5)
self.assertEquals(result[4], -1)
def test_error_4_6(self):
result = test("::1", True, 5)
self.assertEquals(result[4], -1)
if __name__ == '__main__':
unittest.main()
##############################################################################
#
# Copyright (c) 2017 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program 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
# of the License, or (at your option) any later version.
#
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import unittest
import os.path
import socket
from slapos.promise.is_local_tcp_port_opened import isLocalTcpPortOpened
class TestLocalTcpPortOpened(unittest.TestCase):
def test_port_is_not_open(self):
self.assertEquals(isLocalTcpPortOpened("127.0.0.1",65550), False)
def test_port6_is_not_open(self):
self.assertEquals(isLocalTcpPortOpened("::1",65550), False)
def test_port_is_open(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.bind(("127.0.0.1", 0))
s.listen(1)
port = s.getsockname()[1]
self.assertEquals(isLocalTcpPortOpened("127.0.0.1",port), True)
finally:
s.close()
def test_port6_is_open(self):
s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
try:
s.bind(("::1", 0))
s.listen(1)
port = s.getsockname()[1]
self.assertEquals(isLocalTcpPortOpened("::1",port), True)
finally:
s.close()
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