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
20
21from pandora.host_grpc import Host
22from pandora.host_pb2 import Connection
23from pandora_experimental.map_grpc import Map as MapProfile
24from pandora_experimental.os_grpc import Os
25from pandora_experimental.os_pb2 import ACCESS_MESSAGE
26
27
28class MAPProxy(ProfileProxy):
29    """MAP proxy.
30
31    Implements MAP PTS MMIs.
32    """
33
34    connection: Optional[Connection] = None
35
36    def __init__(self, channel):
37        super().__init__(channel)
38
39        self.host = Host(channel)
40        self.os = Os(channel)
41        self.map_profile = MapProfile(channel)
42
43        self.connection = None
44        self._init_send_sms()
45
46    @assert_description
47    def TSC_MMI_iut_connectable(self, **kwargs):
48        """
49        Click OK when the IUT becomes connectable.
50        """
51
52        return "OK"
53
54    @assert_description
55    def TSC_OBEX_MMI_iut_accept_slc_connect_l2cap(self, pts_addr: bytes, **kwargs):
56        """
57        Please accept the l2cap channel connection for an OBEX connection.
58        """
59
60        self.os.SetAccessPermission(address=pts_addr, access_type=ACCESS_MESSAGE)
61        self.connection = self.host.WaitConnection(address=pts_addr).connection
62
63        return "OK"
64
65    @assert_description
66    def TSC_OBEX_MMI_iut_accept_connect(self, test: str, pts_addr: bytes, **kwargs):
67        """
68        Please accept the OBEX CONNECT REQ.
69        """
70
71        if test in {"MAP/MSE/GOEP/BC/BV-01-I", "MAP/MSE/GOEP/BC/BV-03-I", "MAP/MSE/MMN/BV-02-I"}:
72            if self.connection is None:
73                self.os.SetAccessPermission(address=pts_addr, access_type=ACCESS_MESSAGE)
74                self.connection = self.host.WaitConnection(address=pts_addr).connection
75
76        return "OK"
77
78    @assert_description
79    def TSC_OBEX_MMI_iut_initiate_slc_connect(self, **kwargs):
80        """
81        Take action to create an l2cap channel or rfcomm channel for an OBEX
82        connection.
83
84        Note:
85        Service Name: MAP-MNS
86        """
87
88        return "OK"
89
90    @assert_description
91    def TSC_OBEX_MMI_iut_initiate_connect_MAP(self, **kwargs):
92        """
93        Take action to initiate an OBEX CONNECT REQ for MAP.
94        """
95
96        return "OK"
97
98    @assert_description
99    def TSC_OBEX_MMI_iut_initiate_disconnect(self, **kwargs):
100        """
101        Take action to initiate an OBEX DISCONNECT REQ.
102        """
103
104        return "OK"
105
106    @assert_description
107    def TSC_OBEX_MMI_iut_accept_disconnect(self, **kwargs):
108        """
109        Please accept the OBEX DISCONNECT REQ command.
110        """
111
112        return "OK"
113
114    @assert_description
115    def TSC_OBEX_MMI_iut_accept_set_path(self, **kwargs):
116        """
117        Please accept the SET_PATH command.
118        """
119
120        return "OK"
121
122    @assert_description
123    def TSC_OBEX_MMI_iut_accept_get_srm(self, **kwargs):
124        """
125        Please accept the GET REQUEST with an SRM ENABLED header.
126        """
127
128        return "OK"
129
130    @assert_description
131    def TSC_OBEX_MMI_iut_accept_browse_folders(self, **kwargs):
132        """
133        Please accept the browse folders (GET) command.
134        """
135
136        return "OK"
137
138    @assert_description
139    def TSC_MMI_iut_send_set_event_message_MessageRemoved_request(self, **kwargs):
140        """
141        Send Set Event Report with MessageRemoved Message.
142        """
143        return "OK"
144
145    @assert_description
146    def TSC_MMI_iut_verify_message_have_send(self, **kwargs):
147        """
148        Verify that the message has been successfully delivered via the network,
149        then click OK.  Otherwise click Cancel.
150        """
151
152        return "OK"
153
154    @assert_description
155    def TSC_OBEX_MMI_iut_reject_action(self, **kwargs):
156        """
157        Take action to reject the ACTION command sent by PTS.
158        """
159        return "OK"
160
161    @assert_description
162    def TSC_MMI_iut_send_set_event_message_gsm_request(self, **kwargs):
163        """
164        Send Set Event Report with New GSM Message.
165        """
166
167        self.map_profile.SendSMS()
168
169        return "OK"
170
171    @assert_description
172    def TSC_MMI_iut_send_set_event_1_2_request(self, **kwargs):
173        """
174        Send 1.2 Event Report .
175        """
176        return "OK"
177
178    @assert_description
179    def TSC_OBEX_MMI_iut_reject_session(self, **kwargs):
180        """
181        Take action to reject the SESSION command sent by PTS.
182        """
183
184        return "OK"
185
186    def _init_send_sms(self):
187
188        min_sms_count = 2  # Few test cases requires minimum 2 sms to pass
189        for _ in range(min_sms_count):
190            self.map_profile.SendSMS()
191