1#!/usr/bin/env python3
2#
3#  Copyright (c) 2020, The OpenThread Authors.
4#  All rights reserved.
5#
6#  Redistribution and use in source and binary forms, with or without
7#  modification, are permitted provided that the following conditions are met:
8#  1. Redistributions of source code must retain the above copyright
9#     notice, this list of conditions and the following disclaimer.
10#  2. Redistributions in binary form must reproduce the above copyright
11#     notice, this list of conditions and the following disclaimer in the
12#     documentation and/or other materials provided with the distribution.
13#  3. Neither the name of the copyright holder nor the
14#     names of its contributors may be used to endorse or promote products
15#     derived from this software without specific prior written permission.
16#
17#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
18#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
21#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
22#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
23#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
24#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
25#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
26#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27#  POSSIBILITY OF SUCH DAMAGE.
28#
29
30import unittest
31
32import pexpect
33import config
34import thread_cert
35
36LEADER = 1
37ROUTER = 2
38
39
40class TestCoapObserve(thread_cert.TestCase):
41    """
42    Test suite for CoAP Observations (RFC7641).
43    """
44
45    SUPPORT_NCP = False
46
47    TOPOLOGY = {
48        LEADER: {
49            'mode': 'rdn',
50            'allowlist': [ROUTER]
51        },
52        ROUTER: {
53            'mode': 'rdn',
54            'allowlist': [LEADER]
55        },
56    }
57
58    def _do_notification_test(self, con):
59        self.nodes[LEADER].start()
60        self.simulator.go(config.LEADER_STARTUP_DELAY)
61        self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
62
63        self.nodes[ROUTER].start()
64        self.simulator.go(config.ROUTER_STARTUP_DELAY)
65        self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
66
67        mleid = self.nodes[LEADER].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
68
69        self.nodes[LEADER].coap_start()
70        self.nodes[LEADER].coap_set_resource_path('test')
71        self.nodes[LEADER].coap_set_content('Test123')
72
73        self.nodes[ROUTER].coap_start()
74        response = self.nodes[ROUTER].coap_observe(mleid, 'test', con=con)
75
76        first_observe = response['observe']
77        self.assertIsNotNone(first_observe)
78        self.assertEqual(response['payload'], 'Test123')
79        self.assertEqual(response['source'], mleid)
80
81        # This should have been emitted already, so should return immediately
82        self.nodes[LEADER].coap_wait_subscribe()
83
84        # Now change the content on the leader and wait for it to show up
85        # on the router.  We will do this a few times with a short delay.
86        for n in range(0, 5):
87            content = 'msg%d' % n
88
89            self.nodes[LEADER].coap_set_content(content)
90
91            response = self.nodes[ROUTER].coap_wait_response()
92            self.assertGreater(response['observe'], first_observe)
93            self.assertEqual(response['payload'], content)
94            self.assertEqual(response['source'], mleid)
95
96        # Stop subscription
97        self.nodes[ROUTER].coap_cancel()
98
99        # We should see the response, but with no Observe option
100        response = self.nodes[ROUTER].coap_wait_response()
101        self.assertIsNone(response['observe'])
102        # Content won't have changed.
103        self.assertEqual(response['payload'], content)
104
105        # Make another change, no notification should be sent
106        self.nodes[LEADER].coap_set_content('LastNote')
107
108        # This should time out!
109        try:
110            self.nodes[ROUTER].coap_wait_response()
111            self.fail('Should not have received notification')
112        except pexpect.exceptions.TIMEOUT:
113            pass
114
115        self.nodes[ROUTER].coap_stop()
116        self.nodes[LEADER].coap_stop()
117
118    def test_con(self):
119        """
120        Test notification using CON messages.
121        """
122        for trial in range(0, 3):
123            try:
124                self._do_notification_test(con=True)
125                break
126            except (AssertionError, pexpect.exceptions.TIMEOUT):
127                continue
128
129    def test_non(self):
130        """
131        Test notification using NON messages.
132        """
133        for trial in range(0, 3):
134            try:
135                self._do_notification_test(con=False)
136                break
137            except (AssertionError, pexpect.exceptions.TIMEOUT):
138                continue
139
140
141if __name__ == '__main__':
142    unittest.main()
143