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