send_message_received_signal()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 7

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 7
dl 0
loc 7
rs 10
c 0
b 0
f 0
cc 1
nop 2
1
"""
2
:Copyright: 2007-2025 Jochen Kupperschmidt
3
:License: MIT, see LICENSE for details.
4
"""
5
6
import pytest
7
8
from weitersager.config import Config, HttpConfig, IrcConfig
9
from weitersager.processor import Processor
10
from weitersager.signals import irc_channel_joined, message_received
11
12
13
@pytest.fixture
14
def processor():
15
    http_config = HttpConfig(
16
        'localhost',
17
        8080,
18
        api_tokens=set(),
19
        channel_tokens_to_channel_names={},
20
    )
21
22
    irc_config = IrcConfig(
23
        server=None,
24
        nickname='Nick',
25
        realname='Nick',
26
        commands=[],
27
        channels=set(),
28
    )
29
30
    config = Config(log_level='debug', http=http_config, irc=irc_config)
31
32
    return Processor(config)
33
34
35
def test_message_handled(processor):
36
    channel_name = '#foo'
37
    text = 'Knock, knock.'
38
39
    received_signal_data = []
40
41
    def announce(channel_name, text):
42
        received_signal_data.append((channel_name, text))
43
44
    processor.announcer.announce = announce
45
46
    fake_channel_join(channel_name)
47
48
    send_message_received_signal(channel_name, text)
49
50
    processor.process_queue(timeout_seconds=1)
51
52
    assert received_signal_data == [
53
        (channel_name, text),
54
    ]
55
56
57
def fake_channel_join(channel_name):
58
    irc_channel_joined.send(channel_name=channel_name)
59
60
61
def send_message_received_signal(channel_name, text):
62
    source_ip_address = '127.0.0.1'
63
    message_received.send(
64
        None,
65
        channel_name=channel_name,
66
        text=text,
67
        source_ip_address=source_ip_address,
68
    )
69