1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
|
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Tests for the Coturn app component.
"""
import json
from unittest.mock import call, patch
import pytest
from plinth.utils import random_string
from .. import notify_configuration_change
from ..components import (TurnConfiguration, TurnConsumer,
TurnTimeLimitedConsumer, UserTurnConfiguration)
@pytest.fixture(name='turn_configuration')
def fixture_turn_configuration():
"""Return test Coturn configuration."""
return TurnConfiguration('test-domain.example', [], random_string(64))
@pytest.fixture(name='empty_component_list', autouse=True)
def fixture_empty_component_list():
"""Remove all entries in component list before starting a test."""
TurnConsumer._all = {}
def test_turn_configuration_init():
"""Test creating configuration object."""
config = TurnConfiguration('test-domain.example', [], 'test-shared-secret')
assert config.domain == 'test-domain.example'
assert config.shared_secret == 'test-shared-secret'
assert config.uris == [
'stun:test-domain.example:3478',
'turn:test-domain.example:3478?transport=tcp',
'turn:test-domain.example:3478?transport=udp',
]
config = TurnConfiguration(None, ['test-uri1', 'test-uri2'],
'test-shared-secret')
assert config.domain is None
assert config.uris == ['test-uri1', 'test-uri2']
config = TurnConfiguration('test-domain.example',
['test-uri1', 'test-uri2'],
'test-shared-secret')
assert config.domain == 'test-domain.example'
assert config.uris == ['test-uri1', 'test-uri2']
config = UserTurnConfiguration('test-domain.example',
['test-uri1', 'test-uri2'], None,
'test-username', 'test-credential')
assert config.domain == 'test-domain.example'
assert config.uris == ['test-uri1', 'test-uri2']
assert config.shared_secret is None
assert config.username == 'test-username'
assert config.credential == 'test-credential'
def test_turn_configuration_to_json():
"""Test exporting configuration to JSON."""
config = TurnConfiguration('test-domain.example', [], 'test-shared-secret')
assert json.loads(config.to_json()) == {
'domain': 'test-domain.example',
'uris': [
'stun:test-domain.example:3478',
'turn:test-domain.example:3478?transport=tcp',
'turn:test-domain.example:3478?transport=udp',
],
'shared_secret': 'test-shared-secret'
}
def test_turn_configuration_validate_turn_uris():
"""Test validation method to check for STUN/TURN URIs."""
valid_uris = [
'stun:test-domain.example:3478',
'turn:test-domain.example:3478?transport=tcp',
'turn:test-domain.example:3478?transport=udp',
]
invalid_uris = [
'x-invalid-uri',
'stun:',
'stun:domain-port-missing.example',
'stun:testing.example:1234invalid-append',
'turn:testing.example:1234',
'turn:testing.example:1234?invalid-param=value',
'turn:testing.example:1234?transport=invalid-value',
'turn:testing.example:1234?transport=tcp-invalid-append',
]
assert TurnConfiguration().validate_turn_uris(valid_uris)
for uri in invalid_uris:
assert not TurnConfiguration().validate_turn_uris([uri])
def test_component_init_and_list():
"""Test initializing and listing all the components."""
component1 = TurnConsumer('component1')
component2 = TurnConsumer('component2')
component3 = TurnTimeLimitedConsumer('component3')
assert component1.component_id == 'component1'
assert [component1, component2, component3] == list(TurnConsumer.list())
@patch('plinth.modules.coturn.get_config')
def test_notify_on_configuration_changed(get_config, turn_configuration):
"""Test configuration change notifications."""
component = TurnConsumer('component')
get_config.return_value = turn_configuration
with patch.object(component, 'on_config_change') as mock_method:
notify_configuration_change()
mock_method.assert_has_calls([call(turn_configuration)])
@patch('plinth.modules.coturn.get_config')
def test_get_configuration(get_config, turn_configuration):
"""Test coturn configuration retrieval using component."""
get_config.return_value = turn_configuration
component = TurnConsumer('component')
assert component.get_configuration() == turn_configuration
@patch('plinth.modules.coturn.get_config')
def test_get_user_configuration(get_config, turn_configuration):
"""Test coturn user configuration retrieval using component."""
get_config.return_value = turn_configuration
component = TurnTimeLimitedConsumer('component')
user_config = component.get_configuration()
assert user_config.domain == turn_configuration.domain
assert user_config.uris == turn_configuration.uris
assert user_config.shared_secret is None
assert user_config.username is not None
assert user_config.credential is not None
|
