1# -*- coding: utf-8 -*- 2from __future__ import absolute_import 3import unittest 4import tempfile 5import mock 6import socket 7import os 8import port_for 9from port_for.utils import ranges_to_set 10 11 12def test_common_ports(): 13 assert port_for.is_available(80) == False 14 assert port_for.is_available(11211) == False 15 16 17def test_good_port_ranges(): 18 ranges = [ 19 (10, 15), # too short 20 (100, 200), # good 21 (220, 245), # a bit short 22 (300, 330), # good 23 (440, 495), # also good 24 ] 25 26 ports = ranges_to_set(ranges) 27 good_ranges = port_for.good_port_ranges(ports, 20, 3) 28 assert good_ranges == [(103, 197), (443, 492), (303, 327)], good_ranges 29 30 31def test_something_works(): 32 assert len(port_for.good_port_ranges()) > 10 33 assert len(port_for.available_good_ports()) > 1000 34 35 36def test_binding(): 37 # low ports are not available 38 assert port_for.port_is_used(10) == True 39 40 41def test_binding_high(): 42 s = socket.socket() 43 s.bind(("", 0)) 44 port = s.getsockname()[1] 45 assert port_for.port_is_used(port) == True 46 s.close() 47 assert port_for.port_is_used(port) == False 48 49 50class SelectPortTest(unittest.TestCase): 51 52 @mock.patch('port_for.api.port_is_used') 53 def test_all_used(self, port_is_used): 54 port_is_used.return_value = True 55 self.assertRaises(port_for.PortForException, port_for.select_random) 56 57 @mock.patch('port_for.api.port_is_used') 58 def test_random_port(self, port_is_used): 59 ports = set([1, 2, 3]) 60 used = {1: True, 2:False, 3:True} 61 port_is_used.side_effect = lambda port: used[port] 62 63 for x in range(100): 64 self.assertEqual(port_for.select_random(ports), 2) 65 66 67class StoreTest(unittest.TestCase): 68 69 def setUp(self): 70 fd, self.fname = tempfile.mkstemp() 71 self.store = port_for.PortStore(self.fname) 72 73 def tearDown(self): 74 os.remove(self.fname) 75 76 def test_store(self): 77 assert self.store.bound_ports() == [] 78 79 port = self.store.bind_port('foo') 80 self.assertTrue(port) 81 self.assertEqual(self.store.bound_ports(), [('foo', port)]) 82 self.assertEqual(port, self.store.bind_port('foo')) 83 84 port2 = self.store.bind_port('aar') 85 self.assertNotEqual(port, port2) 86 self.assertEqual(self.store.bound_ports(), [('foo', port), ('aar', port2)]) 87 88 self.store.unbind_port('aar') 89 self.assertEqual(self.store.bound_ports(), [('foo', port)]) 90 91 def test_rebind(self): 92 # try to rebind an used port for an another app 93 port = self.store.bind_port('foo') 94 self.assertRaises(port_for.PortForException, self.store.bind_port, 'baz', port) 95 96 def test_change_port(self): 97 # changing app ports is not supported. 98 port = self.store.bind_port('foo') 99 another_port = port_for.select_random() 100 assert port != another_port 101 self.assertRaises(port_for.PortForException, self.store.bind_port, 'foo', another_port) 102 103 def test_bind_unavailable(self): 104 # it is possible to explicitly bind currently unavailable port 105 port = self.store.bind_port('foo', 80) 106 self.assertEqual(port, 80) 107 self.assertEqual(self.store.bound_ports(), [('foo', 80)]) 108 109 def test_bind_non_auto(self): 110 # it is possible to pass a port 111 port = port_for.select_random() 112 res_port = self.store.bind_port('foo', port) 113 self.assertEqual(res_port, port) 114