1#
2#   Copyright 2018 - The Android Open Source Project
3#
4#   Licensed under the Apache License, Version 2.0 (the "License");
5#   you may not use this file except in compliance with the License.
6#   You may obtain a copy of the License at
7#
8#       http://www.apache.org/licenses/LICENSE-2.0
9#
10#   Unless required by applicable law or agreed to in writing, software
11#   distributed under the License is distributed on an "AS IS" BASIS,
12#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13#   See the License for the specific language governing permissions and
14#   limitations under the License.
15
16from acts import asserts
17from acts.test_utils.net import connectivity_const as cconst
18from queue import Empty
19
20def _listen_for_keepalive_event(ad, key, msg, ka_event):
21    """Listen for keepalive event and return status
22
23    Args:
24        ad: DUT object
25        key: keepalive key
26        msg: Error message
27        event: Keepalive event type
28    """
29    ad.droid.socketKeepaliveStartListeningForEvent(key, ka_event)
30    try:
31        event = ad.ed.pop_event("SocketKeepaliveCallback")
32        status = event["data"]["socketKeepaliveEvent"] == ka_event
33    except Empty:
34        asserts.fail(msg)
35    finally:
36        ad.droid.socketKeepaliveStopListeningForEvent(key, ka_event)
37    if ka_event != "Started":
38        ad.droid.removeSocketKeepaliveReceiverKey(key)
39    if status:
40        ad.log.info("'%s' keepalive event successful" % ka_event)
41    return status
42
43def start_natt_socket_keepalive(ad, udp_encap, src_ip, dst_ip, interval = 10):
44    """Start NATT SocketKeepalive on DUT
45
46    Args:
47        ad: DUT object
48        udp_encap: udp_encap socket key
49        src_ip: IP addr of the client
50        dst_ip: IP addr of the keepalive server
51        interval: keepalive time interval
52    """
53    ad.log.info("Starting Natt Socket Keepalive")
54    key = ad.droid.startNattSocketKeepalive(udp_encap, src_ip, dst_ip, interval)
55    msg = "Failed to receive confirmation of starting natt socket keeaplive"
56    status = _listen_for_keepalive_event(ad, key, msg, "Started")
57    return key if status else None
58
59def start_tcp_socket_keepalive(ad, socket, time_interval = 10):
60    """Start TCP socket keepalive on DUT
61
62    Args:
63        ad: DUT object
64        socket: TCP socket key
65        time_interval: Keepalive time interval
66    """
67    ad.log.info("Starting TCP Socket Keepalive")
68    key = ad.droid.startTcpSocketKeepalive(socket, time_interval)
69    msg = "Failed to receive confirmation of starting tcp socket keeaplive"
70    status = _listen_for_keepalive_event(ad, key, msg, "Started")
71    return key if status else None
72
73def socket_keepalive_error(ad, key):
74    """Verify Error callback
75
76    Args:
77        ad: DUT object
78        key: Keepalive key
79    """
80    ad.log.info("Verify Error callback on keepalive: %s" % key)
81    msg = "Failed to receive confirmation of Error callback"
82    return _listen_for_keepalive_event(ad, key, msg, "Error")
83
84def socket_keepalive_data_received(ad, key):
85    """Verify OnDataReceived callback
86
87    Args:
88        ad: DUT object
89        key: Keepalive key
90    """
91    ad.log.info("Verify OnDataReceived callback on keepalive: %s" % key)
92    msg = "Failed to receive confirmation of OnDataReceived callback"
93    return _listen_for_keepalive_event(ad, key, msg, "OnDataReceived")
94
95def stop_socket_keepalive(ad, key):
96    """Stop SocketKeepalive on DUT
97
98    Args:
99        ad: DUT object
100        key: Keepalive key
101    """
102    ad.log.info("Stopping Socket keepalive: %s" % key)
103    ad.droid.stopSocketKeepalive(key)
104    msg = "Failed to receive confirmation of stopping socket keepalive"
105    return _listen_for_keepalive_event(ad, key, msg, "Stopped")
106
107def set_private_dns(ad, dns_mode, hostname=None):
108    """ Set private DNS mode on dut """
109    if dns_mode == cconst.PRIVATE_DNS_MODE_OFF:
110        ad.droid.setPrivateDnsMode(False)
111    else:
112        ad.droid.setPrivateDnsMode(True, hostname)
113
114    mode = ad.droid.getPrivateDnsMode()
115    host = ad.droid.getPrivateDnsSpecifier()
116    ad.log.info("DNS mode is %s and DNS server is %s" % (mode, host))
117    asserts.assert_true(dns_mode == mode and host == hostname,
118                        "Failed to set DNS mode to %s and DNS to %s" % \
119                        (dns_mode, hostname))
120