1# Copyright (C) 2024 The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""MAP proxy module."""
15
16from typing import Optional
17
18from mmi2grpc._helpers import assert_description
19from mmi2grpc._proxy import ProfileProxy
20from pandora.host_grpc import Host
21from pandora.host_pb2 import Connection
22from pandora_experimental.map_grpc import Map as MapProfile
23from pandora_experimental.os_grpc import Os
24from pandora_experimental.os_pb2 import ACCESS_MESSAGE
25
26
27class MAPProxy(ProfileProxy):
28    """MAP proxy.
29
30    Implements MAP PTS MMIs.
31    """
32
33    connection: Optional[Connection] = None
34
35    def __init__(self, channel):
36        super().__init__(channel)
37
38        self.host = Host(channel)
39        self.os = Os(channel)
40        self.map_profile = MapProfile(channel)
41
42        self.connection = None
43        self._init_send_sms()
44
45    @assert_description
46    def TSC_MMI_iut_connectable(self, **kwargs):
47        """
48        Click OK when the IUT becomes connectable.
49        """
50
51        return "OK"
52
53    @assert_description
54    def TSC_OBEX_MMI_iut_accept_slc_connect_l2cap(self, pts_addr: bytes, **kwargs):
55        """
56        Please accept the l2cap channel connection for an OBEX connection.
57        """
58
59        self.os.SetAccessPermission(address=pts_addr, access_type=ACCESS_MESSAGE)
60        self.connection = self.host.WaitConnection(address=pts_addr).connection
61
62        return "OK"
63
64    @assert_description
65    def TSC_OBEX_MMI_iut_accept_connect(self, test: str, pts_addr: bytes, **kwargs):
66        """
67        Please accept the OBEX CONNECT REQ.
68        """
69
70        if test in {"MAP/MSE/GOEP/BC/BV-01-I", "MAP/MSE/GOEP/BC/BV-03-I", "MAP/MSE/MMN/BV-02-I"}:
71            if self.connection is None:
72                self.os.SetAccessPermission(address=pts_addr, access_type=ACCESS_MESSAGE)
73                self.connection = self.host.WaitConnection(address=pts_addr).connection
74
75        return "OK"
76
77    @assert_description
78    def TSC_OBEX_MMI_iut_initiate_slc_connect(self, **kwargs):
79        """
80        Take action to create an l2cap channel or rfcomm channel for an OBEX
81        connection.
82
83        Note:
84        Service Name: MAP-MNS
85        """
86
87        return "OK"
88
89    @assert_description
90    def TSC_OBEX_MMI_iut_initiate_connect_MAP(self, **kwargs):
91        """
92        Take action to initiate an OBEX CONNECT REQ for MAP.
93        """
94
95        return "OK"
96
97    @assert_description
98    def TSC_OBEX_MMI_iut_initiate_disconnect(self, **kwargs):
99        """
100        Take action to initiate an OBEX DISCONNECT REQ.
101        """
102
103        return "OK"
104
105    @assert_description
106    def TSC_OBEX_MMI_iut_accept_disconnect(self, **kwargs):
107        """
108        Please accept the OBEX DISCONNECT REQ command.
109        """
110
111        return "OK"
112
113    @assert_description
114    def TSC_OBEX_MMI_iut_accept_set_path(self, **kwargs):
115        """
116        Please accept the SET_PATH command.
117        """
118
119        return "OK"
120
121    @assert_description
122    def TSC_OBEX_MMI_iut_accept_get_srm(self, **kwargs):
123        """
124        Please accept the GET REQUEST with an SRM ENABLED header.
125        """
126
127        return "OK"
128
129    @assert_description
130    def TSC_OBEX_MMI_iut_accept_browse_folders(self, **kwargs):
131        """
132        Please accept the browse folders (GET) command.
133        """
134
135        return "OK"
136
137    @assert_description
138    def TSC_MMI_iut_send_set_event_message_MessageRemoved_request(self, **kwargs):
139        """
140        Send Set Event Report with MessageRemoved Message.
141        """
142        return "OK"
143
144    @assert_description
145    def TSC_MMI_iut_verify_message_have_send(self, **kwargs):
146        """
147        Verify that the message has been successfully delivered via the network,
148        then click OK.  Otherwise click Cancel.
149        """
150
151        return "OK"
152
153    @assert_description
154    def TSC_OBEX_MMI_iut_reject_action(self, **kwargs):
155        """
156        Take action to reject the ACTION command sent by PTS.
157        """
158        return "OK"
159
160    @assert_description
161    def TSC_MMI_iut_send_set_event_message_gsm_request(self, **kwargs):
162        """
163        Send Set Event Report with New GSM Message.
164        """
165
166        self.map_profile.SendSMS()
167
168        return "OK"
169
170    @assert_description
171    def TSC_MMI_iut_send_set_event_1_2_request(self, **kwargs):
172        """
173        Send 1.2 Event Report .
174        """
175        return "OK"
176
177    @assert_description
178    def TSC_OBEX_MMI_iut_reject_session(self, **kwargs):
179        """
180        Take action to reject the SESSION command sent by PTS.
181        """
182
183        return "OK"
184
185    def _init_send_sms(self):
186
187        min_sms_count = 2  # Few test cases requires minimum 2 sms to pass
188        for _ in range(min_sms_count):
189            self.map_profile.SendSMS()
190