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
|
import asyncio
import logging
from binascii import a2b_hex
from unittest.mock import call
import pytest
from tests.helper import MulticastTestingNetworkTarget, UnicastNetworkTestingTarget
from pyartnet import SacnNode
async def test_sacn() -> None:
sacn = SacnNode(
UnicastNetworkTestingTarget(('ip', 9999999)),
cid=b'\x41\x68\xf5\x2b\x1a\x7b\x2d\xe1\x17\x12\xe9\xee\x38\x3d\x22\x58',
source_name='default source name',
)
async with sacn:
channel = sacn.add_universe(1).add_channel(1, 10)
channel.set_values(range(1, 11))
data = ('001000004153432d45312e31370000007078000000044168f52b1a7b2de11712e9ee383d225870620000000264656661756c7420'
'736f75726365206e616d650000000000000000000000000000000000000000000000000000000000000000000000000000000000'
'0000000064000000000001701502a100000001000b000102030405060708090a')
await channel
await sacn._process_task.task
await asyncio.sleep(0.3)
m = sacn._socket
m.sendto.assert_called_once_with(bytearray(a2b_hex(data)), ('ip', 9999999))
await channel
@pytest.mark.parametrize('multicast', [False, True])
async def test_sacn_with_sync(caplog, multicast) -> None:
caplog.set_level(logging.DEBUG)
if multicast:
network = MulticastTestingNetworkTarget(('ip', 9999999))
else:
network = UnicastNetworkTestingTarget(('ip', 9999999))
sacn = SacnNode(
network,
cid=b'\x41\x68\xf5\x2b\x1a\x7b\x2d\xe1\x17\x12\xe9\xee\x38\x3d\x22\x58',
source_name='default source name',
name='device1'
)
async with sacn:
sacn.set_synchronous_mode(True, 2)
channel = sacn.add_universe(1).add_channel(1, 10)
channel.set_values(range(1, 11))
data = ('001000004153432d45312e31370000007078000000044168f52b1a7b2de11712e9ee383d225870620000000264656661756c7420'
'736f75726365206e616d650000000000000000000000000000000000000000000000000000000000000000000000000000000000'
'0000000064000200000001701502a100000001000b000102030405060708090a')
sync_data = '001000004153432d45312e31370000007021000000084168f52b1a7b2de11712e9ee383d2258700b000000010000020000'
await channel
await sacn._process_task.task
await asyncio.sleep(0.3)
data_dst = ('ip', 9999999) if not multicast else ('239.255.0.1', 5568)
sync_dst = ('ip', 9999999) if not multicast else ('239.255.0.2', 5568)
data_msg = 'ip:9999999' if not multicast else '239.255.0.1:5568'
sync_msg = 'ip:9999999' if not multicast else '239.255.0.2:5568'
m = sacn._socket
assert m.sendto.call_args_list == [
call(bytearray(a2b_hex(data)), data_dst),
call(bytearray(a2b_hex(sync_data)), sync_dst),
]
assert caplog.record_tuples == [
('pyartnet.Universe', 10, 'Added channel "1/10": start: 1, stop: 10'),
('pyartnet.Task', 10, 'Started Refresh task device1'),
('pyartnet.Task', 10, 'Started Process task device1'),
('pyartnet.SacnNode', 10, f'Sending sACN frame to {data_msg:s}: {data:s}'),
('pyartnet.SacnNode', 10, f'Sending sACN Synchronization Packet to {sync_msg}: {sync_data:s}'),
('pyartnet.Task', 10, 'Stopped Process task device1')
]
|
