Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
BC
public
liblinphone
Commits
32c1c8b5
Commit
32c1c8b5
authored
Sep 01, 2014
by
Ghislain MARY
Browse files
Add some message sending unit tests in Python.
parent
cbe15d33
Changes
5
Hide whitespace changes
Inline
Side-by-side
tools/python/unittests/linphonetester.py
View file @
32c1c8b5
from
datetime
import
timedelta
,
datetime
from
nose.tools
import
assert_equals
from
copy
import
deepcopy
import
linphone
import
logging
import
os
...
...
@@ -180,6 +181,78 @@ class CoreManager:
def
wait_for
(
cls
,
manager1
,
manager2
,
func
):
return
cls
.
wait_for_until
(
manager1
,
manager2
,
func
,
10000
)
@
classmethod
def
call
(
cls
,
caller_manager
,
callee_manager
,
caller_params
=
None
,
callee_params
=
None
,
build_callee_params
=
False
):
initial_caller_stats
=
deepcopy
(
caller_manager
.
stats
)
initial_callee_stats
=
deepcopy
(
callee_manager
.
stats
)
# Use playfile for callee to avoid locking on capture card
callee_manager
.
lc
.
use_files
=
True
callee_manager
.
lc
.
play_file
=
os
.
path
.
join
(
tester_resources_path
,
'sounds'
,
'hello8000.wav'
)
if
caller_params
is
None
:
call
=
caller_manager
.
lc
.
invite_address
(
callee_manager
.
identity
)
else
:
call
=
caller_manager
.
lc
.
invite_address_with_params
(
callee_manager
.
identity
,
caller_params
)
assert
call
is
not
None
assert_equals
(
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
callee_manager
.
stats
.
number_of_LinphoneCallIncomingReceived
==
initial_callee_stats
.
number_of_LinphoneCallIncomingReceived
+
1
),
True
)
assert_equals
(
callee_manager
.
lc
.
incoming_invite_pending
,
True
)
assert_equals
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingProgress
,
initial_caller_stats
.
number_of_LinphoneCallOutgoingProgress
+
1
)
retry
=
0
while
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingRinging
!=
initial_caller_stats
.
number_of_LinphoneCallOutgoingRinging
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingEarlyMedia
!=
initial_caller_stats
.
number_of_LinphoneCallOutgoingEarlyMedia
+
1
)
and
\
retry
<
20
:
retry
+=
1
caller_manager
.
lc
.
iterate
()
callee_manager
.
lc
.
iterate
()
time
.
sleep
(
0.1
)
assert
((
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingRinging
==
initial_caller_stats
.
number_of_LinphoneCallOutgoingRinging
+
1
)
or
\
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingEarlyMedia
==
initial_caller_stats
.
number_of_LinphoneCallOutgoingEarlyMedia
+
1
))
==
True
assert
callee_manager
.
lc
.
current_call_remote_address
is
not
None
if
caller_manager
.
lc
.
current_call
is
None
or
callee_manager
.
lc
.
current_call
is
None
or
callee_manager
.
lc
.
current_call_remote_address
is
None
:
return
False
callee_from_address
=
caller_manager
.
identity
.
clone
()
callee_from_address
.
port
=
0
# Remove port because port is never present in from header
assert_equals
(
callee_from_address
.
weak_equal
(
callee_manager
.
lc
.
current_call_remote_address
),
True
)
if
callee_params
is
not
None
:
callee_manager
.
lc
.
accept_call_with_params
(
callee_manager
.
lc
.
current_call
,
callee_params
)
elif
build_callee_params
:
default_params
=
callee_manager
.
lc
.
create_call_params
(
callee_manager
.
lc
.
current_call
)
callee_manager
.
lc
.
accept_call_with_params
(
callee_manager
.
lc
.
current_call
,
default_params
)
else
:
callee_manager
.
lc
.
accept_call
(
callee_manager
.
lc
.
current_call
)
assert_equals
(
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
(
callee_manager
.
stats
.
number_of_LinphoneCallConnected
==
initial_callee_stats
.
number_of_LinphoneCallConnected
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallConnected
==
initial_caller_stats
.
number_of_LinphoneCallConnected
+
1
)),
True
)
# Just to sleep
result
=
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
(
callee_manager
.
stats
.
number_of_LinphoneCallStreamsRunning
==
initial_callee_stats
.
number_of_LinphoneCallStreamsRunning
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallStreamsRunning
==
initial_caller_stats
.
number_of_LinphoneCallStreamsRunning
+
1
))
if
caller_manager
.
lc
.
media_encryption
!=
linphone
.
MediaEncryption
.
MediaEncryptionNone
and
callee_manager
.
lc
.
media_encryption
!=
linphone
.
MediaEncryption
.
MediaEncryptionNone
:
# Wait for encryption to be on, in case of zrtp, it can take a few seconds
if
caller_manager
.
lc
.
media_encryption
==
linphone
.
MediaEncryption
.
MediaEncryptionZRTP
:
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
caller_manager
.
stats
.
number_of_LinphoneCallEncryptedOn
==
initial_caller_stats
.
number_of_LinphoneCallEncryptedOn
+
1
)
if
callee_manager
.
lc
.
media_encryption
==
linphone
.
MediaEncryption
.
MediaEncryptionZRTP
:
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
callee_manager
.
stats
.
number_of_LinphoneCallEncryptedOn
==
initial_callee_stats
.
number_of_LinphoneCallEncryptedOn
+
1
)
assert_equals
(
callee_manager
.
lc
.
current_call
.
current_params
.
media_encryption
,
caller_manager
.
lc
.
media_encryption
)
assert_equals
(
caller_manager
.
lc
.
current_call
.
current_params
.
media_encryption
,
callee_manager
.
lc
.
media_encryption
)
return
result
@
classmethod
def
end_call
(
cls
,
caller_manager
,
callee_manager
):
caller_manager
.
lc
.
terminate_all_calls
()
assert_equals
(
CoreManager
.
wait_for
(
caller_manager
,
callee_manager
,
lambda
caller_manager
,
callee_manager
:
caller_manager
.
stats
.
number_of_LinphoneCallEnd
==
1
and
callee_manager
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
@
classmethod
def
registration_state_changed
(
cls
,
lc
,
cfg
,
state
,
message
):
manager
=
lc
.
user_data
...
...
@@ -257,6 +330,20 @@ class CoreManager:
else
:
raise
Exception
(
"Unexpected call state"
)
@
classmethod
def
message_received
(
cls
,
lc
,
room
,
message
):
manager
=
lc
.
user_data
from_str
=
message
.
from_address
.
as_string
()
text_str
=
message
.
text
external_body_url
=
message
.
external_body_url
if
manager
.
logger
is
not
None
:
manager
.
logger
.
info
(
"[TESTER] Message from [{from_str}] is [{text_str}], external URL [{external_body_url}]"
.
format
(
from_str
=
from_str
,
text_str
=
text_str
,
external_body_url
=
external_body_url
))
manager
.
stats
.
number_of_LinphoneMessageReceived
+=
1
if
message
.
external_body_url
is
not
None
:
manager
.
stats
.
number_of_LinphoneMessageExtBodyReceived
+=
1
def
__init__
(
self
,
rc_file
=
None
,
check_for_proxies
=
True
,
vtable
=
{},
logger
=
None
):
self
.
logger
=
logger
if
not
vtable
.
has_key
(
'registration_state_changed'
):
...
...
@@ -265,10 +352,8 @@ class CoreManager:
vtable
[
'auth_info_requested'
]
=
CoreManager
.
auth_info_requested
if
not
vtable
.
has_key
(
'call_state_changed'
):
vtable
[
'call_state_changed'
]
=
CoreManager
.
call_state_changed
#if not vtable.has_key('text_received'):
#vtable['text_received'] = CoreManager.text_received
#if not vtable.has_key('message_received'):
#vtable['message_received'] = CoreManager.message_received
if
not
vtable
.
has_key
(
'message_received'
):
vtable
[
'message_received'
]
=
CoreManager
.
message_received
#if not vtable.has_key('file_transfer_recv'):
#vtable['file_transfer_recv'] = CoreManager.file_transfer_recv
#if not vtable.has_key('file_transfer_send'):
...
...
tools/python/unittests/test_call.py
View file @
32c1c8b5
from
nose.tools
import
assert_equals
from
copy
import
deepcopy
import
linphone
import
linphonetester
from
linphonetester
import
*
import
os
import
time
...
...
@@ -11,86 +10,16 @@ class TestCall:
@
classmethod
def
setup_class
(
cls
):
base
,
ext
=
os
.
path
.
splitext
(
os
.
path
.
basename
(
__file__
))
cls
.
logger
=
linphonetester
.
Logger
(
base
+
'.log'
)
def
call
(
self
,
caller_manager
,
callee_manager
,
caller_params
=
None
,
callee_params
=
None
,
build_callee_params
=
False
):
initial_caller_stats
=
deepcopy
(
caller_manager
.
stats
)
initial_callee_stats
=
deepcopy
(
callee_manager
.
stats
)
# Use playfile for callee to avoid locking on capture card
callee_manager
.
lc
.
use_files
=
True
callee_manager
.
lc
.
play_file
=
os
.
path
.
join
(
linphonetester
.
tester_resources_path
,
'sounds'
,
'hello8000.wav'
)
if
caller_params
is
None
:
call
=
caller_manager
.
lc
.
invite_address
(
callee_manager
.
identity
)
else
:
call
=
caller_manager
.
lc
.
invite_address_with_params
(
callee_manager
.
identity
,
caller_params
)
assert
call
is
not
None
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
callee_manager
.
stats
.
number_of_LinphoneCallIncomingReceived
==
initial_callee_stats
.
number_of_LinphoneCallIncomingReceived
+
1
),
True
)
assert_equals
(
callee_manager
.
lc
.
incoming_invite_pending
,
True
)
assert_equals
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingProgress
,
initial_caller_stats
.
number_of_LinphoneCallOutgoingProgress
+
1
)
retry
=
0
while
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingRinging
!=
initial_caller_stats
.
number_of_LinphoneCallOutgoingRinging
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingEarlyMedia
!=
initial_caller_stats
.
number_of_LinphoneCallOutgoingEarlyMedia
+
1
)
and
\
retry
<
20
:
retry
+=
1
caller_manager
.
lc
.
iterate
()
callee_manager
.
lc
.
iterate
()
time
.
sleep
(
0.1
)
assert
((
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingRinging
==
initial_caller_stats
.
number_of_LinphoneCallOutgoingRinging
+
1
)
or
\
(
caller_manager
.
stats
.
number_of_LinphoneCallOutgoingEarlyMedia
==
initial_caller_stats
.
number_of_LinphoneCallOutgoingEarlyMedia
+
1
))
==
True
assert
callee_manager
.
lc
.
current_call_remote_address
is
not
None
if
caller_manager
.
lc
.
current_call
is
None
or
callee_manager
.
lc
.
current_call
is
None
or
callee_manager
.
lc
.
current_call_remote_address
is
None
:
return
False
callee_from_address
=
caller_manager
.
identity
.
clone
()
callee_from_address
.
port
=
0
# Remove port because port is never present in from header
assert_equals
(
callee_from_address
.
weak_equal
(
callee_manager
.
lc
.
current_call_remote_address
),
True
)
if
callee_params
is
not
None
:
callee_manager
.
lc
.
accept_call_with_params
(
callee_manager
.
lc
.
current_call
,
callee_params
)
elif
build_callee_params
:
default_params
=
callee_manager
.
lc
.
create_call_params
(
callee_manager
.
lc
.
current_call
)
callee_manager
.
lc
.
accept_call_with_params
(
callee_manager
.
lc
.
current_call
,
default_params
)
else
:
callee_manager
.
lc
.
accept_call
(
callee_manager
.
lc
.
current_call
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
(
callee_manager
.
stats
.
number_of_LinphoneCallConnected
==
initial_callee_stats
.
number_of_LinphoneCallConnected
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallConnected
==
initial_caller_stats
.
number_of_LinphoneCallConnected
+
1
)),
True
)
# Just to sleep
result
=
linphonetester
.
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
(
callee_manager
.
stats
.
number_of_LinphoneCallStreamsRunning
==
initial_callee_stats
.
number_of_LinphoneCallStreamsRunning
+
1
)
and
\
(
caller_manager
.
stats
.
number_of_LinphoneCallStreamsRunning
==
initial_caller_stats
.
number_of_LinphoneCallStreamsRunning
+
1
))
if
caller_manager
.
lc
.
media_encryption
!=
linphone
.
MediaEncryption
.
MediaEncryptionNone
and
callee_manager
.
lc
.
media_encryption
!=
linphone
.
MediaEncryption
.
MediaEncryptionNone
:
# Wait for encryption to be on, in case of zrtp, it can take a few seconds
if
caller_manager
.
lc
.
media_encryption
==
linphone
.
MediaEncryption
.
MediaEncryptionZRTP
:
linphonetester
.
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
caller_manager
.
stats
.
number_of_LinphoneCallEncryptedOn
==
initial_caller_stats
.
number_of_LinphoneCallEncryptedOn
+
1
)
if
callee_manager
.
lc
.
media_encryption
==
linphone
.
MediaEncryption
.
MediaEncryptionZRTP
:
linphonetester
.
CoreManager
.
wait_for
(
callee_manager
,
caller_manager
,
lambda
callee_manager
,
caller_manager
:
callee_manager
.
stats
.
number_of_LinphoneCallEncryptedOn
==
initial_callee_stats
.
number_of_LinphoneCallEncryptedOn
+
1
)
assert_equals
(
callee_manager
.
lc
.
current_call
.
current_params
.
media_encryption
,
caller_manager
.
lc
.
media_encryption
)
assert_equals
(
caller_manager
.
lc
.
current_call
.
current_params
.
media_encryption
,
callee_manager
.
lc
.
media_encryption
)
return
result
def
end_call
(
self
,
caller_manager
,
callee_manager
):
caller_manager
.
lc
.
terminate_all_calls
()
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
caller_manager
,
callee_manager
,
lambda
caller_manager
,
callee_manager
:
caller_manager
.
stats
.
number_of_LinphoneCallEnd
==
1
and
callee_manager
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
cls
.
logger
=
Logger
(
base
+
'.log'
)
def
test_early_declined_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
.
lc
.
max_calls
=
0
out_call
=
pauline
.
lc
.
invite
(
'marie'
)
# Wait until flexisip transfers the busy...
assert_equals
(
linphonetester
.
CoreManager
.
wait_for_until
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallError
==
1
,
33000
),
True
)
assert_equals
(
CoreManager
.
wait_for_until
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallError
==
1
,
33000
),
True
)
assert_equals
(
pauline
.
stats
.
number_of_LinphoneCallError
,
1
)
assert_equals
(
out_call
.
reason
,
linphone
.
Reason
.
ReasonBusy
)
if
len
(
pauline
.
lc
.
call_logs
)
>
0
:
...
...
@@ -101,16 +30,16 @@ class TestCall:
pauline
.
stop
()
def
test_declined_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
out_call
=
pauline
.
lc
.
invite_address
(
marie
.
identity
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
in_call
=
marie
.
lc
.
current_call
assert
in_call
is
not
None
if
in_call
is
not
None
:
marie
.
lc
.
terminate_call
(
in_call
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
marie
.
stats
.
number_of_LinphoneCallEnd
,
1
)
assert_equals
(
pauline
.
stats
.
number_of_LinphoneCallEnd
,
1
)
assert_equals
(
in_call
.
reason
,
linphone
.
Reason
.
ReasonDeclined
)
...
...
@@ -119,69 +48,69 @@ class TestCall:
pauline
.
stop
()
def
test_cancelled_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
out_call
=
pauline
.
lc
.
invite
(
'marie'
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
pauline
.
lc
.
terminate_call
(
out_call
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
pauline
.
stats
.
number_of_LinphoneCallEnd
,
1
)
assert_equals
(
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
,
0
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
marie
.
stop
()
pauline
.
stop
()
def
test_early_cancelled_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'empty_rc'
,
check_for_proxies
=
False
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'empty_rc'
,
check_for_proxies
=
False
,
logger
=
TestCall
.
logger
)
out_call
=
pauline
.
lc
.
invite_address
(
marie
.
identity
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
pauline
.
lc
.
terminate_call
(
out_call
)
# Since everything is executed in a row, no response can be received from the server, thus the CANCEL cannot be sent.
# It will ring at Marie's side.
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
pauline
.
stats
.
number_of_LinphoneCallEnd
,
1
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
# Now the CANCEL should have been sent and the the call at marie's side should terminate
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallEnd
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
),
True
)
marie
.
stop
()
pauline
.
stop
()
def
test_cancelled_ringing_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
out_call
=
pauline
.
lc
.
invite
(
'marie'
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
==
1
),
True
)
pauline
.
lc
.
terminate_call
(
out_call
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
(
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
)
and
(
marie
.
stats
.
number_of_LinphoneCallReleased
==
1
)),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
(
pauline
.
stats
.
number_of_LinphoneCallReleased
==
1
)
and
(
marie
.
stats
.
number_of_LinphoneCallReleased
==
1
)),
True
)
assert_equals
(
marie
.
stats
.
number_of_LinphoneCallEnd
,
1
)
assert_equals
(
pauline
.
stats
.
number_of_LinphoneCallEnd
,
1
)
marie
.
stop
()
pauline
.
stop
()
def
test_call_failed_because_of_codecs
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
marie
.
disable_all_audio_codecs_except_one
(
'pcmu'
)
pauline
.
disable_all_audio_codecs_except_one
(
'pcma'
)
out_call
=
pauline
.
lc
.
invite
(
'marie'
)
assert_equals
(
linphonetester
.
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallOutgoingInit
==
1
),
True
)
# flexisip will retain the 488 until the "urgent reply" timeout arrives.
assert_equals
(
linphonetester
.
CoreManager
.
wait_for_until
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallError
==
1
,
6000
),
True
)
assert_equals
(
CoreManager
.
wait_for_until
(
pauline
,
marie
,
lambda
pauline
,
marie
:
pauline
.
stats
.
number_of_LinphoneCallError
==
1
,
6000
),
True
)
assert_equals
(
out_call
.
reason
,
linphone
.
Reason
.
ReasonNotAcceptable
)
assert_equals
(
marie
.
stats
.
number_of_LinphoneCallIncomingReceived
,
0
)
marie
.
stop
()
pauline
.
stop
()
def
test_simple_call
(
self
):
marie
=
linphonetester
.
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
linphonetester
.
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
assert_equals
(
self
.
call
(
pauline
,
marie
),
True
)
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestCall
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestCall
.
logger
)
assert_equals
(
CoreManager
.
call
(
pauline
,
marie
),
True
)
#liblinphone_tester_check_rtcp(marie,pauline);
self
.
end_call
(
marie
,
pauline
)
CoreManager
.
end_call
(
marie
,
pauline
)
marie
.
stop
()
pauline
.
stop
()
tools/python/unittests/test_message.py
0 → 100644
View file @
32c1c8b5
from
nose.tools
import
assert_equals
from
copy
import
deepcopy
import
linphone
from
linphonetester
import
*
import
os
import
time
class
TestMessage
:
@
classmethod
def
setup_class
(
cls
):
base
,
ext
=
os
.
path
.
splitext
(
os
.
path
.
basename
(
__file__
))
cls
.
logger
=
Logger
(
base
+
'.log'
)
def
wait_for_server_to_purge_messages
(
self
,
manager1
,
manager2
):
# Wait a little bit just to have time to purge message stored in the server
CoreManager
.
wait_for_until
(
manager1
,
manager2
,
lambda
manager1
,
manager2
:
False
,
100
)
manager1
.
stats
.
reset
()
manager2
.
stats
.
reset
()
def
test_text_message
(
self
):
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestMessage
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestMessage
.
logger
)
chat_room
=
pauline
.
lc
.
get_chat_room
(
marie
.
identity
)
self
.
wait_for_server_to_purge_messages
(
marie
,
pauline
)
msg
=
chat_room
.
create_message
(
"Bla bla bla bla"
)
chat_room
.
send_message2
(
msg
,
None
,
None
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneMessageReceived
==
1
),
True
)
assert
marie
.
lc
.
get_chat_room
(
pauline
.
identity
)
is
not
None
marie
.
stop
()
pauline
.
stop
()
def
test_text_message_within_dialog
(
self
):
marie
=
CoreManager
(
'marie_rc'
,
logger
=
TestMessage
.
logger
)
pauline
=
CoreManager
(
'pauline_rc'
,
logger
=
TestMessage
.
logger
)
pauline
.
lc
.
config
.
set_int
(
'sip'
,
'chat_use_call_dialogs'
,
1
)
chat_room
=
pauline
.
lc
.
get_chat_room
(
marie
.
identity
)
self
.
wait_for_server_to_purge_messages
(
marie
,
pauline
)
assert_equals
(
CoreManager
.
call
(
marie
,
pauline
),
True
)
msg
=
chat_room
.
create_message
(
"Bla bla bla bla"
)
chat_room
.
send_message2
(
msg
,
None
,
None
)
assert_equals
(
CoreManager
.
wait_for
(
pauline
,
marie
,
lambda
pauline
,
marie
:
marie
.
stats
.
number_of_LinphoneMessageReceived
==
1
),
True
)
assert
marie
.
lc
.
get_chat_room
(
pauline
.
identity
)
is
not
None
marie
.
stop
()
pauline
.
stop
()
tools/python/unittests/test_register.py
View file @
32c1c8b5
from
nose.tools
import
assert_equals
import
linphone
import
linphonetester
from
linphonetester
import
*
import
os
import
time
class
RegisterCoreManager
(
linphonetester
.
CoreManager
):
class
RegisterCoreManager
(
CoreManager
):
@
classmethod
def
auth_info_requested
(
cls
,
lc
,
realm
,
username
,
domain
):
linphonetester
.
CoreManager
.
auth_info_requested
(
cls
,
lc
,
realm
,
username
,
domain
)
CoreManager
.
auth_info_requested
(
cls
,
lc
,
realm
,
username
,
domain
)
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
...
...
@@ -16,13 +16,13 @@ class RegisterCoreManager(linphonetester.CoreManager):
vtable
=
{}
if
with_auth
:
vtable
[
'auth_info_requested'
]
=
RegisterCoreManager
.
auth_info_requested
linphonetester
.
CoreManager
.
__init__
(
self
,
vtable
=
vtable
)
CoreManager
.
__init__
(
self
,
vtable
=
vtable
)
def
register_with_refresh
(
self
,
refresh
,
domain
,
route
,
late_auth_info
=
False
,
expected_final_state
=
linphone
.
RegistrationState
.
RegistrationOk
):
assert
self
.
lc
is
not
None
self
.
stats
.
reset
()
proxy_cfg
=
self
.
lc
.
create_proxy_config
()
from_address
=
linphonetester
.
create_address
(
domain
)
from_address
=
create_address
(
domain
)
proxy_cfg
.
identity
=
from_address
.
as_string
()
server_addr
=
from_address
.
domain
proxy_cfg
.
register_enabled
=
True
...
...
@@ -50,7 +50,7 @@ class RegisterCoreManager(linphonetester.CoreManager):
if
self
.
stats
.
number_of_auth_info_requested
>
0
and
proxy_cfg
.
state
==
linphone
.
RegistrationState
.
RegistrationFailed
and
late_auth_info
:
if
len
(
self
.
lc
.
auth_info_list
)
==
0
:
assert_equals
(
proxy_cfg
.
error
,
linphone
.
Reason
.
ReasonUnauthorized
)
info
=
linphone
.
AuthInfo
.
new
(
linphonetester
.
test_username
,
None
,
linphonetester
.
test_password
,
None
,
None
,
None
)
# Create authentication structure from identity
info
=
linphone
.
AuthInfo
.
new
(
test_username
,
None
,
test_password
,
None
,
None
,
None
)
# Create authentication structure from identity
self
.
lc
.
add_auth_info
(
info
)
if
proxy_cfg
.
error
==
linphone
.
Reason
.
ReasonForbidden
or
\
(
self
.
stats
.
number_of_auth_info_requested
>
2
and
proxy_cfg
.
error
==
linphone
.
Reason
.
ReasonUnauthorized
):
...
...
@@ -78,7 +78,7 @@ class TestRegister:
@
classmethod
def
setup_class
(
cls
):
base
,
ext
=
os
.
path
.
splitext
(
os
.
path
.
basename
(
__file__
))
cls
.
logger
=
linphonetester
.
Logger
(
base
+
'.log'
)
cls
.
logger
=
Logger
(
base
+
'.log'
)
def
test_simple_register
(
self
):
cm
=
RegisterCoreManager
()
...
...
tools/python/unittests/test_setup.py
View file @
32c1c8b5
from
nose.tools
import
assert_equals
import
linphone
import
linphonetester
from
linphonetester
import
*
import
os
class
TestSetup
:
...
...
@@ -8,10 +8,10 @@ class TestSetup:
@
classmethod
def
setup_class
(
cls
):
base
,
ext
=
os
.
path
.
splitext
(
os
.
path
.
basename
(
__file__
))
cls
.
logger
=
linphonetester
.
Logger
(
base
+
'.log'
)
cls
.
logger
=
Logger
(
base
+
'.log'
)
def
test_address
(
self
):
linphonetester
.
create_address
(
None
)
create_address
(
None
)
def
test_core_init
(
self
):
lc
=
linphone
.
Core
.
new
({},
None
,
None
)
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment