FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
lib-python/test/test_blynk_protocol.py at master · unreliablecode/lib-python · GitHub
unreliablecode
lib-python
Repository navigation
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
lib-python
/
test
/
test_blynk_protocol.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
142 lines (114 loc) · 4.9 KB
Breadcrumbs
lib-python
/
test
/
test_blynk_protocol.py
Copy path
File metadata and controls
142 lines (114 loc) · 4.9 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
136
137
138
139
140
141
142
# -*- coding: utf-8 -*-
import
sys
import
pytest
from
blynklib
import
Protocol
,
BlynkError
class
TestBlynkProtocol
:
@
pytest
.
fixture
def
pb
(
self
):
protocol
=
Protocol
()
yield
protocol
def
test_get_msg_id
(
self
,
pb
):
pb
.
_msg_id
=
4
msg_id
=
pb
.
_get_msg_id
()
assert
msg_id
==
5
def
test_get_msg_id_before_loop
(
self
,
pb
):
pb
.
_msg_id
=
0xFFFE
msg_id
=
pb
.
_get_msg_id
()
assert
msg_id
==
0xFFFF
def
test_get_msg_id_after_loop
(
self
,
pb
):
pb
.
_msg_id
=
0xFFFF
msg_id
=
pb
.
_get_msg_id
()
assert
msg_id
==
1
def
test_get_msg_id_defined
(
self
,
pb
):
pb
.
_msg_id
=
0xFFFF
msg_id
=
pb
.
_get_msg_id
(
msg_id
=
17
)
assert
msg_id
==
17
def
test_pack_msg
(
self
,
pb
):
msg_type
=
20
args
=
[
'test'
,
1234
,
745
,
'abcde'
]
result
=
pb
.
_pack_msg
(
msg_type
,
*
args
)
assert
result
==
b'
\x14
\x00
\x01
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
def
test_pack_msg_no_args
(
self
,
pb
):
msg_type
=
15
args
=
[]
result
=
pb
.
_pack_msg
(
msg_type
,
*
args
)
assert
result
==
b'
\x0f
\x00
\x01
\x00
\x00
'
def
test_pack_msg_unicode
(
self
,
pb
):
if
sys
.
version_info
[
0
]
==
2
:
pytest
.
skip
(
'Python2 unicode compatibility issue'
)
msg_type
=
20
args
=
[
'ёж'
]
result
=
pb
.
_pack_msg
(
msg_type
,
*
args
)
assert
result
==
b'
\x14
\x00
\x01
\x00
\x04
\xd1
\x91
\xd0
\xb6
'
def
test_parse_response_msg_hw
(
self
,
pb
):
data
=
b'
\x14
\x00
\x02
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
msg_buffer
=
1024
result
=
pb
.
parse_response
(
data
,
msg_buffer
)
assert
result
==
(
20
,
2
,
19
, [
u'test'
,
u'1234'
,
u'745'
,
u'abcde'
])
def
test_parse_response_msg_id_0
(
self
,
pb
):
data
=
b'
\x14
\x00
\x00
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
msg_buffer
=
100
with
pytest
.
raises
(
BlynkError
)
as
b_err
:
pb
.
parse_response
(
data
,
msg_buffer
)
assert
'invalid msg_id == 0'
==
str
(
b_err
.
value
)
def
test_parse_response_more_data_than_buffer
(
self
,
pb
):
data
=
b'
\x14
\x00
\x02
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
msg_buffer
=
10
with
pytest
.
raises
(
BlynkError
)
as
b_err
:
pb
.
parse_response
(
data
,
msg_buffer
)
assert
'Command too long'
in
str
(
b_err
.
value
)
def
test_parse_response_msg_ping
(
self
,
pb
):
data
=
b'
\x06
\x00
\x04
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
msg_buffer
=
100
result
=
pb
.
parse_response
(
data
,
msg_buffer
)
assert
result
==
(
6
,
4
,
19
, [])
def
test_parse_response_corrupted_data
(
self
,
pb
):
data
=
b'
\xd1
\x91
\xd0
\xb6
'
msg_buffer
=
100
with
pytest
.
raises
(
Exception
)
as
exc
:
pb
.
parse_response
(
data
,
msg_buffer
)
assert
'Message parse error:'
in
str
(
exc
.
value
)
def
test_parse_response_wrong_msg_type
(
self
,
pb
):
data
=
b'
\x86
\x00
\x04
\x00
\x13
test
\x00
1234
\x00
745
\x00
abcde'
msg_buffer
=
100
with
pytest
.
raises
(
BlynkError
)
as
b_err
:
pb
.
parse_response
(
data
,
msg_buffer
)
assert
"Unknown message type: '134'"
in
str
(
b_err
.
value
)
def
test_parse_response_msg_hw_unicode
(
self
,
pb
):
data
=
b'
\x14
\x00
\x02
\x00
\x04
\xd1
\x91
\xd0
\xb6
'
msg_buffer
=
1024
result
=
pb
.
parse_response
(
data
,
msg_buffer
)
assert
result
==
(
20
,
2
,
4
, [
u'ёж'
])
def
test_heartbeat_msg
(
self
,
pb
):
result
=
pb
.
heartbeat_msg
(
20
,
2048
)
assert
result
==
b'
\x11
\x00
\x01
\x00
+ver
\x00
0.2.6
\x00
buff-in
\x00
2048
\x00
h-beat
\x00
20
\x00
dev
\x00
python'
def
test_login_msg
(
self
,
pb
):
result
=
pb
.
login_msg
(
'1234'
)
assert
result
==
b'
\x02
\x00
\x01
\x00
\x04
1234'
def
test_ping_msg
(
self
,
pb
):
result
=
pb
.
ping_msg
()
assert
result
==
b'
\x06
\x00
\x01
\x00
\x00
'
def
test_response_msg
(
self
,
pb
):
result
=
pb
.
response_msg
(
202
)
assert
result
==
b'
\x00
\x00
\x01
\x00
\x03
202'
def
test_virtual_write_msg
(
self
,
pb
):
result
=
pb
.
virtual_write_msg
(
127
,
'abc'
,
123
)
assert
result
==
b'
\x14
\x00
\x01
\x00
\x0e
vw
\x00
127
\x00
abc
\x00
123'
def
test_virtual_sync_msg
(
self
,
pb
):
result
=
pb
.
virtual_sync_msg
(
1
,
24
)
assert
result
==
b'
\x10
\x00
\x01
\x00
\x07
vr
\x00
1
\x00
24'
def
test_email_msg
(
self
,
pb
):
result
=
pb
.
email_msg
(
'a@b.com'
,
'Test'
,
'MSG'
)
assert
result
==
b'
\r
\x00
\x01
\x00
\x10
a@b.com
\x00
Test
\x00
MSG'
def
test_tweet_msg
(
self
,
pb
):
result
=
pb
.
tweet_msg
(
'tweet_msg_test'
)
assert
result
==
b'
\x0c
\x00
\x01
\x00
\x0e
tweet_msg_test'
def
test_notify_msg
(
self
,
pb
):
result
=
pb
.
notify_msg
(
'app_msg_test'
)
assert
result
==
b'
\x0e
\x00
\x01
\x00
\x0c
app_msg_test'
def
test_set_property_msg
(
self
,
pb
):
result
=
pb
.
set_property_msg
(
10
,
'color'
,
'#FF00EE'
)
assert
result
==
b'
\x13
\x00
\x01
\x00
\x10
10
\x00
color
\x00
#FF00EE'
def
test_internal_msg
(
self
,
pb
):
result
=
pb
.
internal_msg
(
'rtc'
,
'sync'
)
assert
result
==
b'
\x11
\x00
\x01
\x00
\x08
rtc
\x00
sync'
Back
|
FazBrowse Home
|
New Git URL