test_register.py 9.65 KB
Newer Older
1 2
from nose.tools import assert_equals
import linphone
3
from linphonetester import *
4 5 6
import os
import time

7
class RegisterCoreManager(CoreManager):
8 9 10

    @classmethod
    def auth_info_requested(cls, lc, realm, username, domain):
11
        CoreManager.auth_info_requested(lc, realm, username, domain)
12 13 14
        info = linphone.AuthInfo.new(test_username, None, test_password, None, realm, domain) # Create authentication structure from identity
        lc.add_auth_info(info) # Add authentication info to LinphoneCore

15
    def __init__(self, with_auth = False):
16 17 18
        vtable = {}
        if with_auth:
            vtable['auth_info_requested'] = RegisterCoreManager.auth_info_requested
19
        CoreManager.__init__(self, vtable=vtable)
20

21
    def register_with_refresh_base(self, refresh, domain, route, late_auth_info = False, transport = linphone.SipTransports(5070, 5070, 5071, 0), expected_final_state = linphone.RegistrationState.Ok):
22 23
        assert self.lc is not None
        self.stats.reset()
24
        self.lc.sip_transports = transport
25
        proxy_cfg = self.lc.create_proxy_config()
26
        from_address = create_address(domain)
27
        proxy_cfg.identity_address = from_address
28 29 30 31 32 33 34 35 36
        server_addr = from_address.domain
        proxy_cfg.register_enabled = True
        proxy_cfg.expires = 1
        if route is None:
            proxy_cfg.server_addr = server_addr
        else:
            proxy_cfg.route = route
            proxy_cfg.server_addr = route
        self.lc.add_proxy_config(proxy_cfg)
37
        self.lc.default_proxy_config = proxy_cfg
38 39 40 41 42 43

        retry = 0
        expected_count = 1
        if refresh:
            expected_count += 1
        max_retry = 110
Ghislain MARY's avatar
Ghislain MARY committed
44
        if expected_final_state == linphone.RegistrationState.Progress:
45 46 47 48
            max_retry += 200
        while self.stats.number_of_LinphoneRegistrationOk < expected_count and retry < max_retry:
            retry += 1
            self.lc.iterate()
Ghislain MARY's avatar
Ghislain MARY committed
49
            if self.stats.number_of_auth_info_requested > 0 and proxy_cfg.state == linphone.RegistrationState.Failed and late_auth_info:
50
                if len(self.lc.auth_info_list) == 0:
Ghislain MARY's avatar
Ghislain MARY committed
51
                    assert_equals(proxy_cfg.error, linphone.Reason.Unauthorized)
52
                    info = linphone.AuthInfo.new(test_username, None, test_password, None, None, None) # Create authentication structure from identity
53
                    self.lc.add_auth_info(info)
Ghislain MARY's avatar
Ghislain MARY committed
54 55
            if proxy_cfg.error == linphone.Reason.Forbidden or \
                (self.stats.number_of_auth_info_requested > 2 and proxy_cfg.error == linphone.Reason.Unauthorized):
56 57 58 59 60 61
                break
            time.sleep(0.1)

        assert_equals(proxy_cfg.state, expected_final_state)
        assert_equals(self.stats.number_of_LinphoneRegistrationNone, 0)
        assert self.stats.number_of_LinphoneRegistrationProgress >= 1
Ghislain MARY's avatar
Ghislain MARY committed
62
        if expected_final_state == linphone.RegistrationState.Ok:
63 64 65 66 67 68 69 70
            assert_equals(self.stats.number_of_LinphoneRegistrationOk, expected_count)
            expected_failed = 0
            if late_auth_info:
                expected_failed = 1
            assert_equals(self.stats.number_of_LinphoneRegistrationFailed, expected_failed)
        else:
            assert_equals(self.stats.number_of_LinphoneRegistrationCleared, 0)

71 72
    def register_with_refresh(self, refresh, domain, route, late_auth_info = False, transport = linphone.SipTransports(5070, 5070, 5071, 0), expected_final_state = linphone.RegistrationState.Ok):
        self.register_with_refresh_base(refresh, domain, route, late_auth_info, expected_final_state = expected_final_state)
73 74
        # Not testable as the callbacks can not be called once the core destruction has started
        #assert_equals(self.stats.number_of_LinphoneRegistrationCleared, 1)
75 76 77 78 79


class TestRegister:

    def test_simple_register(self):
80
        cm = RegisterCoreManager()
81 82
        cm.register_with_refresh(False, None, None)
        assert_equals(cm.stats.number_of_auth_info_requested, 0)
83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145

    def test_simple_unregister(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh_base(False, None, None)
        pc = cm.lc.default_proxy_config
        pc.edit()
        cm.stats.reset() # clear stats
        # nothing is supposed to arrive until done
        assert_equals(CoreManager.wait_for_until(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationCleared == 1, 3000), False)
        pc.register_enabled = False
        pc.done()
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationCleared == 1), True)

    def test_simple_tcp_register(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh(False, test_domain, "sip:{route};transport=tcp".format(route=test_route))

    def test_simple_tcp_register_compatibility_mode(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh(False, test_domain, "sip:{route}".format(route=test_route), transport=linphone.SipTransports(0, 5070, 0, 0))

    def test_simple_tls_register(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh(False, test_domain, "sip:{route};transport=tls".format(route=test_route))

    def test_tls_register_with_alt_name(self):
        cm = CoreManager('pauline_alt_rc', False)
        cm.lc.root_ca = os.path.join(tester_resources_path, 'certificates', 'cn', 'cafile.pem')
        cm.lc.refresh_registers()
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 1), True)
        assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 0)

    def test_tls_wildcard_register(self):
        cm = CoreManager('pauline_wild_rc', False)
        cm.lc.root_ca = os.path.join(tester_resources_path, 'certificates', 'cn', 'cafile.pem')
        cm.lc.refresh_registers()
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 2), True)
        assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 0)

    def test_tls_certificate_failure(self):
        cm = CoreManager('pauline_rc', False)
        cm.lc.root_ca = os.path.join(tester_resources_path, 'certificates', 'cn', 'agent.pem') # bad root ca
        cm.lc.network_reachable = True
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == 1), True)
        cm.lc.root_ca = None # no root ca
        cm.lc.refresh_registers()
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == 2), True)
        cm.lc.root_ca = os.path.join(tester_resources_path, 'certificates', 'cn', 'cafile.pem') # good root ca
        cm.lc.refresh_registers()
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 1), True)
        assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 2)

    def test_tls_with_non_tls_server(self):
        cm = CoreManager('marie_rc', False)
        cm.lc.sip_transport_timeout = 3000
        pc = cm.lc.default_proxy_config
        pc.edit()
        addr = linphone.Address.new(pc.server_addr)
        port = addr.port
        if port <= 0:
            port = 5060
        pc.server_addr = "sip:{domain}:{port};transport=tls".format(domain=addr.domain, port=port)
        pc.done()
146
        assert_equals(CoreManager.wait_for_until(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == 1, 10000), True)
147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175

    def test_simple_authenticated_register(self):
        cm = RegisterCoreManager()
        info = linphone.AuthInfo.new(test_username, None, test_password, None, auth_domain, None) # Create authentication structure from identity
        cm.lc.add_auth_info(info)
        cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route))
        assert_equals(cm.stats.number_of_auth_info_requested, 0)

    def test_digest_auth_without_initial_credentials(self):
        cm = RegisterCoreManager(with_auth=True)
        cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route))
        assert_equals(cm.stats.number_of_auth_info_requested, 1)

    def test_authenticated_register_with_late_credentials(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route), True, linphone.SipTransports(5070, 5070, 5071, 0))
        assert_equals(cm.stats.number_of_auth_info_requested, 1)

    def test_simple_register_with_refresh(self):
        cm = RegisterCoreManager()
        cm.register_with_refresh(True, None, None)
        assert_equals(cm.stats.number_of_auth_info_requested, 0)

    def test_simple_auth_register_with_refresh(self):
        cm = RegisterCoreManager(with_auth=True)
        cm.register_with_refresh(True, auth_domain, "sip:{route}".format(route=test_route))
        assert_equals(cm.stats.number_of_auth_info_requested, 1)

    def test_multiple_accounts(self):
Ghislain MARY's avatar
Ghislain MARY committed
176
        CoreManager('multi_account_rc', False)
177 178

    def test_transport_change(self):
Ghislain MARY's avatar
Ghislain MARY committed
179
        cm = CoreManager('multi_account_rc', True)
180 181
        number_of_udp_proxies = reduce(lambda x, y: x + int(y.transport == "udp"), cm.lc.proxy_config_list, 0)
        total_number_of_proxies = len(cm.lc.proxy_config_list)
182
        register_ok = cm.stats.number_of_LinphoneRegistrationOk
183
        # Keep only UDP
184 185
        tr = linphone.SipTransports(0, 0, 0, 0)
        tr.udp_port = cm.lc.sip_transports.udp_port
186
        cm.lc.sip_transports = tr
187
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == (register_ok + number_of_udp_proxies)), True)
188
        assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == (total_number_of_proxies - number_of_udp_proxies)), True)