1#!/usr/bin/env python3
2#
3#   Copyright 2016 - 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.
16import mock
17import unittest
18
19from acts.controllers.sl4a_lib import rpc_client, rpc_connection
20
21MOCK_RESP = b'{"id": 0, "result": 123, "error": null, "status": 1, "uid": 1}'
22MOCK_RESP_UNKNOWN_UID = b'{"id": 0, "result": 123, "error": null, "status": 0}'
23MOCK_RESP_WITH_ERROR = b'{"id": 0, "error": 1, "status": 1, "uid": 1}'
24
25
26class MockSocketFile(object):
27    def __init__(self, resp):
28        self.resp = resp
29        self.last_write = None
30
31    def write(self, msg):
32        self.last_write = msg
33
34    def readline(self):
35        return self.resp
36
37    def flush(self):
38        pass
39
40
41class RpcConnectionTest(unittest.TestCase):
42    """This test class has unit tests for the implementation of everything
43    under acts.controllers.android, which is the RPC client module for sl4a.
44    """
45
46    @staticmethod
47    def mock_rpc_connection(response=MOCK_RESP,
48                            uid=rpc_connection.UNKNOWN_UID):
49        """Sets up a faked socket file from the mock connection."""
50        fake_file = MockSocketFile(response)
51        fake_conn = mock.MagicMock()
52        fake_conn.makefile.return_value = fake_file
53        adb = mock.Mock()
54        ports = mock.Mock()
55
56        return rpc_connection.RpcConnection(
57            adb, ports, fake_conn, fake_file, uid=uid)
58
59    def test_open_chooses_init_on_unknown_uid(self):
60        """Tests rpc_connection.RpcConnection.open().
61
62        Tests that open uses the init start command when the uid is unknown.
63        """
64
65        def pass_on_init(start_command):
66            if not start_command == rpc_connection.Sl4aConnectionCommand.INIT:
67                self.fail(
68                    'Must call "init". Called "%s" instead.' % start_command)
69
70        connection = self.mock_rpc_connection()
71        connection._initiate_handshake = pass_on_init
72        connection.open()
73
74    def test_open_chooses_continue_on_known_uid(self):
75        """Tests rpc_connection.RpcConnection.open().
76
77        Tests that open uses the continue start command when the uid is known.
78        """
79
80        def pass_on_continue(start_command):
81            if start_command != rpc_connection.Sl4aConnectionCommand.CONTINUE:
82                self.fail('Must call "continue". Called "%s" instead.' %
83                          start_command)
84
85        connection = self.mock_rpc_connection(uid=1)
86        connection._initiate_handshake = pass_on_continue
87        connection.open()
88
89    def test_initiate_handshake_returns_uid(self):
90        """Tests rpc_connection.RpcConnection._initiate_handshake().
91
92        Test that at the end of a handshake with no errors the client object
93        has the correct parameters.
94        """
95        connection = self.mock_rpc_connection()
96        connection._initiate_handshake(
97            rpc_connection.Sl4aConnectionCommand.INIT)
98
99        self.assertEqual(connection.uid, 1)
100
101    def test_initiate_handshake_returns_unknown_status(self):
102        """Tests rpc_connection.RpcConnection._initiate_handshake().
103
104        Test that when the handshake is given an unknown uid then the client
105        will not be given a uid.
106        """
107        connection = self.mock_rpc_connection(MOCK_RESP_UNKNOWN_UID)
108        connection._initiate_handshake(
109            rpc_connection.Sl4aConnectionCommand.INIT)
110
111        self.assertEqual(connection.uid, rpc_client.UNKNOWN_UID)
112
113    def test_initiate_handshake_no_response(self):
114        """Tests rpc_connection.RpcConnection._initiate_handshake().
115
116        Test that if a handshake receives no response then it will give a
117        protocol error.
118        """
119        connection = self.mock_rpc_connection(b'')
120
121        with self.assertRaises(
122                rpc_client.Sl4aProtocolError,
123                msg=rpc_client.Sl4aProtocolError.NO_RESPONSE_FROM_HANDSHAKE):
124            connection._initiate_handshake(
125                rpc_connection.Sl4aConnectionCommand.INIT)
126
127    def test_cmd_properly_formatted(self):
128        """Tests rpc_connection.RpcConnection._cmd().
129
130        Tests that the command sent is properly formatted.
131        """
132        connection = self.mock_rpc_connection(MOCK_RESP)
133        connection._cmd('test')
134        self.assertIn(
135            connection._socket_file.last_write,
136            [b'{"cmd": "test", "uid": -1}\n', b'{"uid": -1, "cmd": "test"}\n'])
137
138    def test_get_new_ticket(self):
139        """Tests rpc_connection.RpcConnection.get_new_ticket().
140
141        Tests that a new number is always given for get_new_ticket().
142        """
143        connection = self.mock_rpc_connection(MOCK_RESP)
144        self.assertEqual(connection.get_new_ticket() + 1,
145                         connection.get_new_ticket())
146
147
148if __name__ == "__main__":
149    unittest.main()
150