1#!/usr/bin python3
2#
3#       Copyright 2019 - The Android Open Source Project
4#
5#       Licensed under the Apache License, Version 2.0 (the "License");
6#       you may not use this file except in compliance with the License.
7#       You may obtain a copy of the License at
8#
9#               http://www.apache.org/licenses/LICENSE-2.0
10#
11#       Unless required by applicable law or agreed to in writing, software
12#       distributed under the License is distributed on an "AS IS" BASIS,
13#       WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#       See the License for the specific language governing permissions and
15#       limitations under the License.
16"""Python unittest module for GNSS Abstract Instrument Library."""
17
18import socket
19import unittest
20from unittest.mock import Mock
21from unittest.mock import patch
22import acts.controllers.abstract_inst as pyinst
23
24
25class SocketInstrumentTest(unittest.TestCase):
26    """A class for unit-testing acts.controllers.gnssinst_lib.abstract_inst"""
27
28    @patch('socket.create_connection')
29    def test__connect_socket(self, mock_connect):
30        """test socket connection normal completion."""
31        mock_connect.return_value.recv.return_value = b'Dummy Instrument\n'
32
33        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
34        test_inst._connect_socket()
35
36        mock_connect.assert_called_with(('192.168.1.11', '5050'), timeout=120)
37
38    @patch('socket.create_connection')
39    def test__connect_socket_timeout(self, mock_connect):
40        """test socket connection with timeout."""
41        mock_connect.side_effect = socket.timeout
42
43        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
44
45        with self.assertRaises(pyinst.SocketInstrumentError):
46            test_inst._connect_socket()
47
48    @patch('socket.create_connection')
49    def test__connect_socket_error(self, mock_connect):
50        """test socket connection with socket error."""
51        mock_connect.side_effect = socket.error
52
53        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
54
55        with self.assertRaises(pyinst.SocketInstrumentError):
56            test_inst._connect_socket()
57
58    def test__send(self):
59        """test send function with normal completion."""
60        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
61
62        test_inst._socket = Mock()
63
64        test_inst._send('TestCommand')
65
66        test_inst._socket.sendall.assert_called_with(b'TestCommand\n')
67
68    def test__send_timeout(self):
69        """test send function with timeout."""
70        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
71
72        test_inst._socket = Mock()
73        test_inst._socket.sendall.side_effect = socket.timeout
74
75        with self.assertRaises(pyinst.SocketInstrumentError):
76            test_inst._send('TestCommand')
77
78    def test__send_error(self):
79        """test send function with error."""
80        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
81
82        test_inst._socket = Mock()
83        test_inst._socket.sendall.side_effect = socket.error
84
85        with self.assertRaises(pyinst.SocketInstrumentError):
86            test_inst._send('TestCommand')
87
88    def test__recv(self):
89        """test recv function with normal completion."""
90        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
91
92        test_inst._socket = Mock()
93        test_inst._socket.recv.return_value = b'TestResponse\n'
94
95        mock_resp = test_inst._recv()
96
97        self.assertEqual(mock_resp, 'TestResponse')
98
99    def test__recv_timeout(self):
100        """test recv function with timeout."""
101        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
102
103        test_inst._socket = Mock()
104        test_inst._socket.recv.side_effect = socket.timeout
105
106        with self.assertRaises(pyinst.SocketInstrumentError):
107            test_inst._recv()
108
109    def test__recv_error(self):
110        """test recv function with error."""
111        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
112
113        test_inst._socket = Mock()
114        test_inst._socket.recv.side_effect = socket.error
115
116        with self.assertRaises(pyinst.SocketInstrumentError):
117            test_inst._recv()
118
119    @patch('socket.create_connection')
120    def test__close_socket(self, mock_connect):
121        """test socket close normal completion."""
122        mock_connect.return_value.recv.return_value = b'Dummy Instrument\n'
123
124        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
125        test_inst._connect_socket()
126        test_inst._close_socket()
127
128        mock_connect.return_value.shutdown.assert_called_with(socket.SHUT_RDWR)
129        mock_connect.return_value.close.assert_called_with()
130
131    def test__query(self):
132        """test query function with normal completion."""
133        test_inst = pyinst.SocketInstrument('192.168.1.11', '5050')
134
135        test_inst._socket = Mock()
136        test_inst._socket.recv.return_value = b'TestResponse\n'
137
138        mock_resp = test_inst._query('TestCommand')
139
140        test_inst._socket.sendall.assert_called_with(b'TestCommand;*OPC?\n')
141        self.assertEqual(mock_resp, 'TestResponse')
142
143
144if __name__ == '__main__':
145    unittest.main()
146