| #!/usr/bin/python3.4 |
| # |
| # Copyright 2017 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import string |
| import time |
| |
| from acts import asserts |
| from acts.test_utils.wifi.aware import aware_const as aconsts |
| from acts.test_utils.wifi.aware import aware_test_utils as autils |
| from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest |
| |
| |
| class DiscoveryTest(AwareBaseTest): |
| """Set of tests for Wi-Fi Aware discovery.""" |
| |
| # configuration parameters used by tests |
| PAYLOAD_SIZE_MIN = 0 |
| PAYLOAD_SIZE_TYPICAL = 1 |
| PAYLOAD_SIZE_MAX = 2 |
| |
| # message strings |
| query_msg = "How are you doing? 你好嗎？" |
| response_msg = "Doing ok - thanks! 做的不錯 - 謝謝！" |
| |
| # message re-transmit counter (increases reliability in open-environment) |
| # Note: reliability of message transmission is tested elsewhere |
| msg_retx_count = 5 # hard-coded max value, internal API |
| |
| def __init__(self, controllers): |
| AwareBaseTest.__init__(self, controllers) |
| |
| def create_base_config(self, caps, is_publish, ptype, stype, payload_size, |
| ttl, term_ind_on, null_match): |
| """Create a base configuration based on input parameters. |
| |
| Args: |
| caps: device capability dictionary |
| is_publish: True if a publish config, else False |
| ptype: unsolicited or solicited (used if is_publish is True) |
| stype: passive or active (used if is_publish is False) |
| payload_size: min, typical, max (PAYLOAD_SIZE_xx) |
| ttl: time-to-live configuration (0 - forever) |
| term_ind_on: is termination indication enabled |
| null_match: null-out the middle match filter |
| Returns: |
| publish discovery configuration object. |
| """ |
| config = {} |
| if is_publish: |
| config[aconsts.DISCOVERY_KEY_DISCOVERY_TYPE] = ptype |
| else: |
| config[aconsts.DISCOVERY_KEY_DISCOVERY_TYPE] = stype |
| config[aconsts.DISCOVERY_KEY_TTL] = ttl |
| config[aconsts.DISCOVERY_KEY_TERM_CB_ENABLED] = term_ind_on |
| if payload_size == self.PAYLOAD_SIZE_MIN: |
| config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = "a" |
| config[aconsts.DISCOVERY_KEY_SSI] = None |
| config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = [] |
| elif payload_size == self.PAYLOAD_SIZE_TYPICAL: |
| config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = "GoogleTestServiceX" |
| if is_publish: |
| config[aconsts.DISCOVERY_KEY_SSI] = string.ascii_letters |
| else: |
| config[aconsts.DISCOVERY_KEY_SSI] = string.ascii_letters[:: |
| -1] # reverse |
| config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = autils.encode_list( |
| [(10).to_bytes(1, byteorder="big"), "hello there string" |
| if not null_match else None, |
| bytes(range(40))]) |
| else: # PAYLOAD_SIZE_MAX |
| config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = "VeryLong" + "X" * ( |
| caps[aconsts.CAP_MAX_SERVICE_NAME_LEN] - 8) |
| config[aconsts.DISCOVERY_KEY_SSI] = ("P" if is_publish else "S") * caps[ |
| aconsts.CAP_MAX_SERVICE_SPECIFIC_INFO_LEN] |
| mf = autils.construct_max_match_filter( |
| caps[aconsts.CAP_MAX_MATCH_FILTER_LEN]) |
| if null_match: |
| mf[2] = None |
| config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = autils.encode_list(mf) |
| |
| return config |
| |
| def create_publish_config(self, caps, ptype, payload_size, ttl, term_ind_on, |
| null_match): |
| """Create a publish configuration based on input parameters. |
| |
| Args: |
| caps: device capability dictionary |
| ptype: unsolicited or solicited |
| payload_size: min, typical, max (PAYLOAD_SIZE_xx) |
| ttl: time-to-live configuration (0 - forever) |
| term_ind_on: is termination indication enabled |
| null_match: null-out the middle match filter |
| Returns: |
| publish discovery configuration object. |
| """ |
| return self.create_base_config(caps, True, ptype, None, payload_size, ttl, |
| term_ind_on, null_match) |
| |
| def create_subscribe_config(self, caps, stype, payload_size, ttl, term_ind_on, |
| null_match): |
| """Create a subscribe configuration based on input parameters. |
| |
| Args: |
| caps: device capability dictionary |
| stype: passive or active |
| payload_size: min, typical, max (PAYLOAD_SIZE_xx) |
| ttl: time-to-live configuration (0 - forever) |
| term_ind_on: is termination indication enabled |
| null_match: null-out the middle match filter |
| Returns: |
| subscribe discovery configuration object. |
| """ |
| return self.create_base_config(caps, False, None, stype, payload_size, ttl, |
| term_ind_on, null_match) |
| |
| def positive_discovery_test_utility(self, ptype, stype, payload_size): |
| """Utility which runs a positive discovery test: |
| - Discovery (publish/subscribe) with TTL=0 (non-self-terminating) |
| - Exchange messages |
| - Update publish/subscribe |
| - Terminate |
| |
| Args: |
| ptype: Publish discovery type |
| stype: Subscribe discovery type |
| payload_size: One of PAYLOAD_SIZE_* constants - MIN, TYPICAL, MAX |
| """ |
| p_dut = self.android_devices[0] |
| p_dut.pretty_name = "Publisher" |
| s_dut = self.android_devices[1] |
| s_dut.pretty_name = "Subscriber" |
| |
| # Publisher+Subscriber: attach and wait for confirmation |
| p_id = p_dut.droid.wifiAwareAttach(False) |
| autils.wait_for_event(p_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| time.sleep(self.device_startup_offset) |
| s_id = s_dut.droid.wifiAwareAttach(False) |
| autils.wait_for_event(s_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| |
| # Publisher: start publish and wait for confirmation |
| p_config = self.create_publish_config( |
| p_dut.aware_capabilities, |
| ptype, |
| payload_size, |
| ttl=0, |
| term_ind_on=False, |
| null_match=False) |
| p_disc_id = p_dut.droid.wifiAwarePublish(p_id, p_config) |
| autils.wait_for_event(p_dut, aconsts.SESSION_CB_ON_PUBLISH_STARTED) |
| |
| # Subscriber: start subscribe and wait for confirmation |
| s_config = self.create_subscribe_config( |
| s_dut.aware_capabilities, |
| stype, |
| payload_size, |
| ttl=0, |
| term_ind_on=False, |
| null_match=True) |
| s_disc_id = s_dut.droid.wifiAwareSubscribe(s_id, s_config) |
| autils.wait_for_event(s_dut, aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED) |
| |
| # Subscriber: wait for service discovery |
| discovery_event = autils.wait_for_event( |
| s_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| peer_id_on_sub = discovery_event["data"][aconsts.SESSION_CB_KEY_PEER_ID] |
| |
| # Subscriber: validate contents of discovery: |
| # - SSI: publisher's |
| # - Match filter: UNSOLICITED - publisher, SOLICITED - subscriber |
| autils.assert_equal_strings( |
| bytes(discovery_event["data"][ |
| aconsts.SESSION_CB_KEY_SERVICE_SPECIFIC_INFO]).decode("utf-8"), |
| p_config[aconsts.DISCOVERY_KEY_SSI], |
| "Discovery mismatch: service specific info (SSI)") |
| asserts.assert_equal( |
| autils.decode_list( |
| discovery_event["data"][aconsts.SESSION_CB_KEY_MATCH_FILTER_LIST]), |
| autils.decode_list(p_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] |
| if ptype == aconsts.PUBLISH_TYPE_UNSOLICITED else |
| s_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST]), |
| "Discovery mismatch: match filter") |
| |
| # Subscriber: send message to peer (Publisher) |
| s_dut.droid.wifiAwareSendMessage(s_disc_id, peer_id_on_sub, |
| self.get_next_msg_id(), self.query_msg, |
| self.msg_retx_count) |
| autils.wait_for_event(s_dut, aconsts.SESSION_CB_ON_MESSAGE_SENT) |
| |
| # Publisher: wait for received message |
| pub_rx_msg_event = autils.wait_for_event( |
| p_dut, aconsts.SESSION_CB_ON_MESSAGE_RECEIVED) |
| peer_id_on_pub = pub_rx_msg_event["data"][aconsts.SESSION_CB_KEY_PEER_ID] |
| |
| # Publisher: validate contents of message |
| asserts.assert_equal( |
| pub_rx_msg_event["data"][aconsts.SESSION_CB_KEY_MESSAGE_AS_STRING], |
| self.query_msg, "Subscriber -> Publisher message corrupted") |
| |
| # Publisher: send message to peer (Subscriber) |
| p_dut.droid.wifiAwareSendMessage(p_disc_id, peer_id_on_pub, |
| self.get_next_msg_id(), self.response_msg, |
| self.msg_retx_count) |
| autils.wait_for_event(p_dut, aconsts.SESSION_CB_ON_MESSAGE_SENT) |
| |
| # Subscriber: wait for received message |
| sub_rx_msg_event = autils.wait_for_event( |
| s_dut, aconsts.SESSION_CB_ON_MESSAGE_RECEIVED) |
| |
| # Subscriber: validate contents of message |
| asserts.assert_equal( |
| sub_rx_msg_event["data"][aconsts.SESSION_CB_KEY_PEER_ID], |
| peer_id_on_sub, |
| "Subscriber received message from different peer ID then discovery!?") |
| autils.assert_equal_strings( |
| sub_rx_msg_event["data"][aconsts.SESSION_CB_KEY_MESSAGE_AS_STRING], |
| self.response_msg, "Publisher -> Subscriber message corrupted") |
| |
| # Subscriber: validate that we're not getting another Service Discovery |
| autils.fail_on_event(s_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| |
| # Publisher: update publish and wait for confirmation |
| p_config[aconsts.DISCOVERY_KEY_SSI] = "something else" |
| p_dut.droid.wifiAwareUpdatePublish(p_disc_id, p_config) |
| autils.wait_for_event(p_dut, aconsts.SESSION_CB_ON_SESSION_CONFIG_UPDATED) |
| |
| # Subscriber: expect a new service discovery |
| discovery_event = autils.wait_for_event( |
| s_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| |
| # Subscriber: validate contents of discovery |
| autils.assert_equal_strings( |
| bytes(discovery_event["data"][ |
| aconsts.SESSION_CB_KEY_SERVICE_SPECIFIC_INFO]).decode("utf-8"), |
| p_config[aconsts.DISCOVERY_KEY_SSI], |
| "Discovery mismatch (after pub update): service specific info (SSI)") |
| asserts.assert_equal( |
| autils.decode_list( |
| discovery_event["data"][aconsts.SESSION_CB_KEY_MATCH_FILTER_LIST]), |
| autils.decode_list(p_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] |
| if ptype == aconsts.PUBLISH_TYPE_UNSOLICITED else |
| s_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST]), |
| "Discovery mismatch: match filter") |
| |
| # Subscribe: update subscribe and wait for confirmation |
| s_config = self.create_subscribe_config( |
| s_dut.aware_capabilities, |
| stype, |
| payload_size, |
| ttl=0, |
| term_ind_on=False, |
| null_match=False) |
| s_dut.droid.wifiAwareUpdateSubscribe(s_disc_id, s_config) |
| autils.wait_for_event(s_dut, aconsts.SESSION_CB_ON_SESSION_CONFIG_UPDATED) |
| |
| # Publisher+Subscriber: Terminate sessions |
| p_dut.droid.wifiAwareDestroyDiscoverySession(p_disc_id) |
| s_dut.droid.wifiAwareDestroyDiscoverySession(s_disc_id) |
| |
| # sleep for timeout period and then verify all 'fail_on_event' together |
| time.sleep(autils.EVENT_TIMEOUT) |
| |
| # verify that there were no other events |
| autils.verify_no_more_events(p_dut, timeout=0) |
| autils.verify_no_more_events(s_dut, timeout=0) |
| |
| # verify that forbidden callbacks aren't called |
| autils.validate_forbidden_callbacks(p_dut, {aconsts.CB_EV_MATCH: 0}) |
| |
| def verify_discovery_session_term(self, dut, disc_id, config, is_publish, |
| term_ind_on): |
| """Utility to verify that the specified discovery session has terminated (by |
| waiting for the TTL and then attempting to reconfigure). |
| |
| Args: |
| dut: device under test |
| disc_id: discovery id for the existing session |
| config: configuration of the existing session |
| is_publish: True if the configuration was publish, False if subscribe |
| term_ind_on: True if a termination indication is expected, False otherwise |
| """ |
| # Wait for session termination |
| if term_ind_on: |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SESSION_TERMINATED, |
| disc_id)) |
| else: |
| # can't defer wait to end since in any case have to wait for session to |
| # expire |
| autils.fail_on_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SESSION_TERMINATED, |
| disc_id)) |
| |
| # Validate that session expired by trying to configure it (expect failure) |
| config[aconsts.DISCOVERY_KEY_SSI] = "something else" |
| if is_publish: |
| dut.droid.wifiAwareUpdatePublish(disc_id, config) |
| else: |
| dut.droid.wifiAwareUpdateSubscribe(disc_id, config) |
| |
| # The response to update discovery session is: |
| # term_ind_on=True: session was cleaned-up so won't get an explicit failure, but won't get a |
| # success either. Can check for no SESSION_CB_ON_SESSION_CONFIG_UPDATED but |
| # will defer to the end of the test (no events on queue). |
| # term_ind_on=False: session was not cleaned-up (yet). So expect |
| # SESSION_CB_ON_SESSION_CONFIG_FAILED. |
| if not term_ind_on: |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SESSION_CONFIG_FAILED, |
| disc_id)) |
| |
| def positive_ttl_test_utility(self, is_publish, ptype, stype, term_ind_on): |
| """Utility which runs a positive discovery session TTL configuration test |
| |
| Iteration 1: Verify session started with TTL |
| Iteration 2: Verify session started without TTL and reconfigured with TTL |
| Iteration 3: Verify session started with (long) TTL and reconfigured with |
| (short) TTL |
| |
| Args: |
| is_publish: True if testing publish, False if testing subscribe |
| ptype: Publish discovery type (used if is_publish is True) |
| stype: Subscribe discovery type (used if is_publish is False) |
| term_ind_on: Configuration of termination indication |
| """ |
| SHORT_TTL = 5 # 5 seconds |
| LONG_TTL = 100 # 100 seconds |
| dut = self.android_devices[0] |
| |
| # Attach and wait for confirmation |
| id = dut.droid.wifiAwareAttach(False) |
| autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED) |
| |
| # Iteration 1: Start discovery session with TTL |
| config = self.create_base_config(dut.aware_capabilities, is_publish, ptype, |
| stype, self.PAYLOAD_SIZE_TYPICAL, |
| SHORT_TTL, term_ind_on, False) |
| if is_publish: |
| disc_id = dut.droid.wifiAwarePublish(id, config, True) |
| autils.wait_for_event(dut, |
| autils.decorate_event( |
| aconsts.SESSION_CB_ON_PUBLISH_STARTED, disc_id)) |
| else: |
| disc_id = dut.droid.wifiAwareSubscribe(id, config, True) |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED, |
| disc_id)) |
| |
| # Wait for session termination & verify |
| self.verify_discovery_session_term(dut, disc_id, config, is_publish, |
| term_ind_on) |
| |
| # Iteration 2: Start a discovery session without TTL |
| config = self.create_base_config(dut.aware_capabilities, is_publish, ptype, |
| stype, self.PAYLOAD_SIZE_TYPICAL, 0, |
| term_ind_on, False) |
| if is_publish: |
| disc_id = dut.droid.wifiAwarePublish(id, config, True) |
| autils.wait_for_event(dut, |
| autils.decorate_event( |
| aconsts.SESSION_CB_ON_PUBLISH_STARTED, disc_id)) |
| else: |
| disc_id = dut.droid.wifiAwareSubscribe(id, config, True) |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED, |
| disc_id)) |
| |
| # Update with a TTL |
| config = self.create_base_config(dut.aware_capabilities, is_publish, ptype, |
| stype, self.PAYLOAD_SIZE_TYPICAL, |
| SHORT_TTL, term_ind_on, False) |
| if is_publish: |
| dut.droid.wifiAwareUpdatePublish(disc_id, config) |
| else: |
| dut.droid.wifiAwareUpdateSubscribe(disc_id, config) |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SESSION_CONFIG_UPDATED, |
| disc_id)) |
| |
| # Wait for session termination & verify |
| self.verify_discovery_session_term(dut, disc_id, config, is_publish, |
| term_ind_on) |
| |
| # Iteration 3: Start a discovery session with (long) TTL |
| config = self.create_base_config(dut.aware_capabilities, is_publish, ptype, |
| stype, self.PAYLOAD_SIZE_TYPICAL, LONG_TTL, |
| term_ind_on, False) |
| if is_publish: |
| disc_id = dut.droid.wifiAwarePublish(id, config, True) |
| autils.wait_for_event(dut, |
| autils.decorate_event( |
| aconsts.SESSION_CB_ON_PUBLISH_STARTED, disc_id)) |
| else: |
| disc_id = dut.droid.wifiAwareSubscribe(id, config, True) |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED, |
| disc_id)) |
| |
| # Update with a TTL |
| config = self.create_base_config(dut.aware_capabilities, is_publish, ptype, |
| stype, self.PAYLOAD_SIZE_TYPICAL, |
| SHORT_TTL, term_ind_on, False) |
| if is_publish: |
| dut.droid.wifiAwareUpdatePublish(disc_id, config) |
| else: |
| dut.droid.wifiAwareUpdateSubscribe(disc_id, config) |
| autils.wait_for_event( |
| dut, |
| autils.decorate_event(aconsts.SESSION_CB_ON_SESSION_CONFIG_UPDATED, |
| disc_id)) |
| |
| # Wait for session termination & verify |
| self.verify_discovery_session_term(dut, disc_id, config, is_publish, |
| term_ind_on) |
| |
| # verify that there were no other events |
| autils.verify_no_more_events(dut) |
| |
| # verify that forbidden callbacks aren't called |
| if not term_ind_on: |
| autils.validate_forbidden_callbacks(dut, { |
| aconsts.CB_EV_PUBLISH_TERMINATED: 0, |
| aconsts.CB_EV_SUBSCRIBE_TERMINATED: 0 |
| }) |
| |
| def discovery_mismatch_test_utility(self, |
| is_expected_to_pass, |
| p_type, |
| s_type, |
| p_service_name=None, |
| s_service_name=None, |
| p_mf_1=None, |
| s_mf_1=None): |
| """Utility which runs the negative discovery test for mismatched service |
| configs. |
| |
| Args: |
| is_expected_to_pass: True if positive test, False if negative |
| p_type: Publish discovery type |
| s_type: Subscribe discovery type |
| p_service_name: Publish service name (or None to leave unchanged) |
| s_service_name: Subscribe service name (or None to leave unchanged) |
| p_mf_1: Publish match filter element [1] (or None to leave unchanged) |
| s_mf_1: Subscribe match filter element [1] (or None to leave unchanged) |
| """ |
| p_dut = self.android_devices[0] |
| p_dut.pretty_name = "Publisher" |
| s_dut = self.android_devices[1] |
| s_dut.pretty_name = "Subscriber" |
| |
| # create configurations |
| p_config = self.create_publish_config( |
| p_dut.aware_capabilities, |
| p_type, |
| self.PAYLOAD_SIZE_TYPICAL, |
| ttl=0, |
| term_ind_on=False, |
| null_match=False) |
| if p_service_name is not None: |
| p_config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = p_service_name |
| if p_mf_1 is not None: |
| p_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = autils.encode_list( |
| [(10).to_bytes(1, byteorder="big"), |
| p_mf_1, |
| bytes(range(40))]) |
| s_config = self.create_publish_config( |
| s_dut.aware_capabilities, |
| s_type, |
| self.PAYLOAD_SIZE_TYPICAL, |
| ttl=0, |
| term_ind_on=False, |
| null_match=False) |
| if s_service_name is not None: |
| s_config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = s_service_name |
| if s_mf_1 is not None: |
| s_config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = autils.encode_list( |
| [(10).to_bytes(1, byteorder="big"), |
| s_mf_1, |
| bytes(range(40))]) |
| |
| p_id = p_dut.droid.wifiAwareAttach(False) |
| autils.wait_for_event(p_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| time.sleep(self.device_startup_offset) |
| s_id = s_dut.droid.wifiAwareAttach(False) |
| autils.wait_for_event(s_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| |
| # Publisher: start publish and wait for confirmation |
| p_disc_id = p_dut.droid.wifiAwarePublish(p_id, p_config) |
| autils.wait_for_event(p_dut, aconsts.SESSION_CB_ON_PUBLISH_STARTED) |
| |
| # Subscriber: start subscribe and wait for confirmation |
| s_disc_id = s_dut.droid.wifiAwareSubscribe(s_id, s_config) |
| autils.wait_for_event(s_dut, aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED) |
| |
| # Subscriber: fail on service discovery |
| if is_expected_to_pass: |
| autils.wait_for_event(s_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| else: |
| autils.fail_on_event(s_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| |
| # Publisher+Subscriber: Terminate sessions |
| p_dut.droid.wifiAwareDestroyDiscoverySession(p_disc_id) |
| s_dut.droid.wifiAwareDestroyDiscoverySession(s_disc_id) |
| |
| # verify that there were no other events (including terminations) |
| time.sleep(autils.EVENT_TIMEOUT) |
| autils.verify_no_more_events(p_dut, timeout=0) |
| autils.verify_no_more_events(s_dut, timeout=0) |
| |
| |
| ####################################### |
| # Positive tests key: |
| # |
| # names is: test_<pub_type>_<sub_type>_<size> |
| # where: |
| # |
| # pub_type: Type of publish discovery session: unsolicited or solicited. |
| # sub_type: Type of subscribe discovery session: passive or active. |
| # size: Size of payload fields (service name, service specific info, and match |
| # filter: typical, max, or min. |
| ####################################### |
| |
| def test_positive_unsolicited_passive_typical(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Solicited publish + passive subscribe |
| - Typical payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| payload_size=self.PAYLOAD_SIZE_TYPICAL) |
| |
| def test_positive_unsolicited_passive_min(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Solicited publish + passive subscribe |
| - Minimal payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| payload_size=self.PAYLOAD_SIZE_MIN) |
| |
| def test_positive_unsolicited_passive_max(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Solicited publish + passive subscribe |
| - Maximal payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| payload_size=self.PAYLOAD_SIZE_MAX) |
| |
| |
| def test_positive_solicited_active_typical(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Unsolicited publish + active subscribe |
| - Typical payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_SOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| payload_size=self.PAYLOAD_SIZE_TYPICAL) |
| |
| def test_positive_solicited_active_min(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Unsolicited publish + active subscribe |
| - Minimal payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_SOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| payload_size=self.PAYLOAD_SIZE_MIN) |
| |
| def test_positive_solicited_active_max(self): |
| """Functional test case / Discovery test cases / positive test case: |
| - Unsolicited publish + active subscribe |
| - Maximal payload fields size |
| |
| Verifies that discovery and message exchange succeeds. |
| """ |
| self.positive_discovery_test_utility( |
| ptype=aconsts.PUBLISH_TYPE_SOLICITED, |
| stype=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| payload_size=self.PAYLOAD_SIZE_MAX) |
| |
| ####################################### |
| # TTL tests key: |
| # |
| # names is: test_ttl_<pub_type|sub_type>_<term_ind> |
| # where: |
| # |
| # pub_type: Type of publish discovery session: unsolicited or solicited. |
| # sub_type: Type of subscribe discovery session: passive or active. |
| # term_ind: ind_on or ind_off |
| ####################################### |
| |
| def test_ttl_unsolicited_ind_on(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Unsolicited publish |
| - Termination indication enabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=True, |
| ptype=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| stype=None, |
| term_ind_on=True) |
| |
| def test_ttl_unsolicited_ind_off(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Unsolicited publish |
| - Termination indication disabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=True, |
| ptype=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| stype=None, |
| term_ind_on=False) |
| |
| def test_ttl_solicited_ind_on(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Solicited publish |
| - Termination indication enabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=True, |
| ptype=aconsts.PUBLISH_TYPE_SOLICITED, |
| stype=None, |
| term_ind_on=True) |
| |
| def test_ttl_solicited_ind_off(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Solicited publish |
| - Termination indication disabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=True, |
| ptype=aconsts.PUBLISH_TYPE_SOLICITED, |
| stype=None, |
| term_ind_on=False) |
| |
| def test_ttl_passive_ind_on(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Passive subscribe |
| - Termination indication enabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=False, |
| ptype=None, |
| stype=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| term_ind_on=True) |
| |
| def test_ttl_passive_ind_off(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Passive subscribe |
| - Termination indication disabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=False, |
| ptype=None, |
| stype=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| term_ind_on=False) |
| |
| def test_ttl_active_ind_on(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Active subscribe |
| - Termination indication enabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=False, |
| ptype=None, |
| stype=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| term_ind_on=True) |
| |
| def test_ttl_active_ind_off(self): |
| """Functional test case / Discovery test cases / TTL test case: |
| - Active subscribe |
| - Termination indication disabled |
| """ |
| self.positive_ttl_test_utility( |
| is_publish=False, |
| ptype=None, |
| stype=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| term_ind_on=False) |
| |
| ####################################### |
| # Mismatched service name tests key: |
| # |
| # names is: test_mismatch_service_name_<pub_type>_<sub_type> |
| # where: |
| # |
| # pub_type: Type of publish discovery session: unsolicited or solicited. |
| # sub_type: Type of subscribe discovery session: passive or active. |
| ####################################### |
| |
| def test_mismatch_service_name_unsolicited_passive(self): |
| """Functional test case / Discovery test cases / Mismatch service name |
| - Unsolicited publish |
| - Passive subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=False, |
| p_type=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| p_service_name="GoogleTestServiceXXX", |
| s_service_name="GoogleTestServiceYYY") |
| |
| def test_mismatch_service_name_solicited_active(self): |
| """Functional test case / Discovery test cases / Mismatch service name |
| - Solicited publish |
| - Active subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=False, |
| p_type=aconsts.PUBLISH_TYPE_SOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| p_service_name="GoogleTestServiceXXX", |
| s_service_name="GoogleTestServiceYYY") |
| |
| ####################################### |
| # Mismatched discovery session type tests key: |
| # |
| # names is: test_mismatch_service_type_<pub_type>_<sub_type> |
| # where: |
| # |
| # pub_type: Type of publish discovery session: unsolicited or solicited. |
| # sub_type: Type of subscribe discovery session: passive or active. |
| ####################################### |
| |
| def test_mismatch_service_type_unsolicited_active(self): |
| """Functional test case / Discovery test cases / Mismatch service name |
| - Unsolicited publish |
| - Active subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=True, |
| p_type=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_ACTIVE) |
| |
| def test_mismatch_service_type_solicited_passive(self): |
| """Functional test case / Discovery test cases / Mismatch service name |
| - Unsolicited publish |
| - Active subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=False, |
| p_type=aconsts.PUBLISH_TYPE_SOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_PASSIVE) |
| |
| ####################################### |
| # Mismatched discovery match filter tests key: |
| # |
| # names is: test_mismatch_match_filter_<pub_type>_<sub_type> |
| # where: |
| # |
| # pub_type: Type of publish discovery session: unsolicited or solicited. |
| # sub_type: Type of subscribe discovery session: passive or active. |
| ####################################### |
| |
| def test_mismatch_match_filter_unsolicited_passive(self): |
| """Functional test case / Discovery test cases / Mismatch match filter |
| - Unsolicited publish |
| - Passive subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=False, |
| p_type=aconsts.PUBLISH_TYPE_UNSOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_PASSIVE, |
| p_mf_1="hello there string", |
| s_mf_1="goodbye there string") |
| |
| def test_mismatch_match_filter_solicited_active(self): |
| """Functional test case / Discovery test cases / Mismatch match filter |
| - Solicited publish |
| - Active subscribe |
| """ |
| self.discovery_mismatch_test_utility( |
| is_expected_to_pass=False, |
| p_type=aconsts.PUBLISH_TYPE_SOLICITED, |
| s_type=aconsts.SUBSCRIBE_TYPE_ACTIVE, |
| p_mf_1="hello there string", |
| s_mf_1="goodbye there string") |