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
15from mobly import asserts
16from net_tests_utils.host.python import adb_utils, apf_utils, assert_utils, multi_devices_test_base, tether_utils
17from net_tests_utils.host.python.tether_utils import UpstreamType
18import time
19
20class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase):
21
22  def setup_class(self):
23    super().setup_class()
24
25    # Check test preconditions.
26    asserts.abort_class_if(
27        not self.client.isAtLeastV(),
28        "Do not enforce the test until V+ since chipset potential bugs are"
29        " expected to be fixed on V+ releases.",
30    )
31    tether_utils.assume_hotspot_test_preconditions(
32        self.serverDevice, self.clientDevice, UpstreamType.NONE
33    )
34    asserts.abort_class_if(
35        not apf_utils.is_send_raw_packet_downstream_supported(
36            self.serverDevice
37        ),
38        "NetworkStack is too old to support send raw packet, skip test.",
39    )
40
41    # Fetch device properties and storing them locally for later use.
42    # TODO: refactor to separate instances to store client and server device
43    self.server_iface_name, client_network = (
44        tether_utils.setup_hotspot_and_client_for_upstream_type(
45            self.serverDevice, self.clientDevice, UpstreamType.NONE
46        )
47    )
48    self.client_iface_name = self.client.getInterfaceNameFromNetworkHandle(
49        client_network
50    )
51    self.server_mac_address = apf_utils.get_hardware_address(
52        self.serverDevice, self.server_iface_name
53    )
54    self.client_mac_address = apf_utils.get_hardware_address(
55        self.clientDevice, self.client_iface_name
56    )
57    self.server_ipv4_addresses = apf_utils.get_ipv4_addresses(
58        self.serverDevice, self.server_iface_name
59    )
60    self.client_ipv4_addresses = apf_utils.get_ipv4_addresses(
61        self.clientDevice, self.client_iface_name
62    )
63    self.server_ipv6_addresses = apf_utils.get_ipv6_addresses(
64        self.serverDevice, self.server_iface_name
65    )
66    self.client_ipv6_addresses = apf_utils.get_ipv6_addresses(
67        self.clientDevice, self.client_iface_name
68    )
69
70    # Enable doze mode to activate APF.
71    adb_utils.set_doze_mode(self.clientDevice, True)
72
73  def teardown_class(self):
74    adb_utils.set_doze_mode(self.clientDevice, False)
75    tether_utils.cleanup_tethering_for_upstream_type(
76        self.serverDevice, UpstreamType.NONE
77    )
78
79  def send_packet_and_expect_counter_increased(
80      self, packet: str, counter_name: str
81  ) -> None:
82    count_before_test = apf_utils.get_apf_counter(
83        self.clientDevice,
84        self.client_iface_name,
85        counter_name,
86    )
87    apf_utils.send_raw_packet_downstream(
88        self.serverDevice, self.server_iface_name, packet
89    )
90
91    assert_utils.expect_with_retry(
92        lambda: apf_utils.get_apf_counter(
93            self.clientDevice,
94            self.client_iface_name,
95            counter_name,
96        )
97        > count_before_test
98    )
99
100  def send_packet_and_expect_reply_received(
101      self, send_packet: str, counter_name: str, receive_packet: str
102  ) -> None:
103    try:
104        apf_utils.start_capture_packets(self.serverDevice, self.server_iface_name)
105
106        self.send_packet_and_expect_counter_increased(send_packet, counter_name)
107
108        assert_utils.expect_with_retry(
109            lambda: apf_utils.get_matched_packet_counts(
110                self.serverDevice, self.server_iface_name, receive_packet
111            )
112            == 1
113        )
114    finally:
115        apf_utils.stop_capture_packets(self.serverDevice, self.server_iface_name)
116