4
4
import unittest
5
5
from unittest import mock
6
6
7
- import snap7 .error
8
- import snap7 .server
9
- import snap7 .types
7
+ from snap7 .common import error_text
8
+ from snap7 .error import server_errors
9
+ from snap7 .server import Server
10
+ from snap7 .types import SrvEvent , mkEvent , mkLog , srvAreaDB , LocalPort , WorkInterval , MaxClients , RemotePort
10
11
11
12
logging .basicConfig (level = logging .WARNING )
12
13
13
14
14
15
@pytest .mark .server
15
16
class TestServer (unittest .TestCase ):
16
17
def setUp (self ):
17
- self .server = snap7 . server . Server ()
18
+ self .server = Server ()
18
19
self .server .start (tcpport = 1102 )
19
20
20
21
def tearDown (self ):
@@ -23,29 +24,29 @@ def tearDown(self):
23
24
24
25
def test_register_area (self ):
25
26
db1_type = ctypes .c_char * 1024
26
- self .server .register_area (snap7 . types . srvAreaDB , 3 , db1_type ())
27
+ self .server .register_area (srvAreaDB , 3 , db1_type ())
27
28
28
29
def test_error (self ):
29
- for error in snap7 . error . server_errors :
30
- snap7 . common . error_text (error , context = "client" )
30
+ for error in server_errors :
31
+ error_text (error , context = "client" )
31
32
32
33
def test_event (self ):
33
- event = snap7 . types . SrvEvent ()
34
+ event = SrvEvent ()
34
35
self .server .event_text (event )
35
36
36
37
def test_get_status (self ):
37
38
server , cpu , num_clients = self .server .get_status ()
38
39
39
40
def test_get_mask (self ):
40
- self .server .get_mask (snap7 . types . mkEvent )
41
- self .server .get_mask (snap7 . types . mkLog )
41
+ self .server .get_mask (mkEvent )
42
+ self .server .get_mask (mkLog )
42
43
# invalid kind
43
44
self .assertRaises (Exception , self .server .get_mask , 3 )
44
45
45
46
def test_lock_area (self ):
46
47
from threading import Thread
47
48
48
- area_code = snap7 . types . srvAreaDB
49
+ area_code = srvAreaDB
49
50
index = 1
50
51
db1_type = ctypes .c_char * 1024
51
52
# we need to register first
@@ -72,10 +73,10 @@ def test_set_cpu_status(self):
72
73
self .assertRaises (ValueError , self .server .set_cpu_status , - 1 )
73
74
74
75
def test_set_mask (self ):
75
- self .server .set_mask (kind = snap7 . types . mkEvent , mask = 10 )
76
+ self .server .set_mask (kind = mkEvent , mask = 10 )
76
77
77
78
def test_unlock_area (self ):
78
- area_code = snap7 . types . srvAreaDB
79
+ area_code = srvAreaDB
79
80
index = 1
80
81
db1_type = ctypes .c_char * 1024
81
82
@@ -87,7 +88,7 @@ def test_unlock_area(self):
87
88
self .server .unlock_area (area_code , index )
88
89
89
90
def test_unregister_area (self ):
90
- area_code = snap7 . types . srvAreaDB
91
+ area_code = srvAreaDB
91
92
index = 1
92
93
db1_type = ctypes .c_char * 1024
93
94
self .server .register_area (area_code , index , db1_type ())
@@ -107,7 +108,7 @@ def read_events_call_back(event):
107
108
108
109
def test_pick_event (self ):
109
110
event = self .server .pick_event ()
110
- self .assertEqual (type (event ), snap7 . types . SrvEvent )
111
+ self .assertEqual (type (event ), SrvEvent )
111
112
event = self .server .pick_event ()
112
113
self .assertFalse (event )
113
114
@@ -121,12 +122,12 @@ def test_start_to(self):
121
122
122
123
def test_get_param (self ):
123
124
# check the defaults
124
- self .assertEqual (self .server .get_param (snap7 . types . LocalPort ), 1102 )
125
- self .assertEqual (self .server .get_param (snap7 . types . WorkInterval ), 100 )
126
- self .assertEqual (self .server .get_param (snap7 . types . MaxClients ), 1024 )
125
+ self .assertEqual (self .server .get_param (LocalPort ), 1102 )
126
+ self .assertEqual (self .server .get_param (WorkInterval ), 100 )
127
+ self .assertEqual (self .server .get_param (MaxClients ), 1024 )
127
128
128
129
# invalid param for server
129
- self .assertRaises (Exception , self .server .get_param , snap7 . types . RemotePort )
130
+ self .assertRaises (Exception , self .server .get_param , RemotePort )
130
131
131
132
132
133
@pytest .mark .server
@@ -136,10 +137,10 @@ class TestServerBeforeStart(unittest.TestCase):
136
137
"""
137
138
138
139
def setUp (self ):
139
- self .server = snap7 . server . Server ()
140
+ self .server = Server ()
140
141
141
142
def test_set_param (self ):
142
- self .server .set_param (snap7 . types . LocalPort , 1102 )
143
+ self .server .set_param (LocalPort , 1102 )
143
144
144
145
145
146
@pytest .mark .server
@@ -161,11 +162,11 @@ def tearDown(self):
161
162
self .loadlib_patch .stop ()
162
163
163
164
def test_create (self ):
164
- snap7 . server . Server (log = False )
165
+ Server (log = False )
165
166
self .mocklib .Srv_Create .assert_called_once ()
166
167
167
168
def test_gc (self ):
168
- server = snap7 . server . Server (log = False )
169
+ server = Server (log = False )
169
170
del server
170
171
self .mocklib .Srv_Destroy .assert_called_once ()
171
172
0 commit comments