1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
#
# SPDX-License-Identifier: MIT
"""This test runs a resource directory and executes the examples from the RD
specification against it.
"""
import unittest
import aiocoap
from aiocoap.util import hostportjoin
from .test_server import WithAsyncLoop, Destructing, WithClient, asynctest
linkheader_modules = aiocoap.defaults.linkheader_missing_modules()
_skip_unless_linkheader = unittest.skipIf(
linkheader_modules,
"Modules missing for running RD tests: %s" % (linkheader_modules,),
)
if not linkheader_modules:
from aiocoap.util.linkformat import link_header
import aiocoap.cli.rd
class WithResourceDirectory(WithAsyncLoop, Destructing):
rd_address = "::1"
rd_port = 56830
rd_netloc = "[%s]:%d" % (rd_address, rd_port)
def setUp(self):
super().setUp()
self.rd = aiocoap.cli.rd.Main(
["--bind", hostportjoin("::1", self.rd_port)], loop=self.loop
)
self.loop.run_until_complete(self.rd.initializing)
def tearDown(self):
self.loop.run_until_complete(self.rd.shutdown())
super().tearDown()
self._del_to_be_sure("rd")
class TestDiscovery(WithResourceDirectory, WithClient):
@_skip_unless_linkheader
@asynctest
async def test_discovery(self):
request = aiocoap.Message(
code=aiocoap.GET,
uri="coap://%s/.well-known/core?rt=core.rd*" % self.rd_netloc,
)
response = await self.client.request(request).response
self.assertEqual(
response.code, aiocoap.CONTENT, "RD discovery did not give content"
)
links = link_header.parse(response.payload.decode("utf8"))
# Not checking for presence of group resources: not implemented here
for rt in ("core.rd", "core.rd-lookup-ep", "core.rd-lookup-res"):
self.assertEqual(
len([x for x in links.links if x.rt == [rt]]),
1,
"Not exactly one entry of rt=%s found" % rt,
)
async def _get_endpoint(self, rt):
"""Return the URI for a given rt in the configured RD"""
if not hasattr(self, "_endpoints"):
request = aiocoap.Message(
code=aiocoap.GET,
uri="coap://%s/.well-known/core?rt=core.rd*" % self.rd_netloc,
)
response = await self.client.request(request).response
self._endpoints = {
entry.rt[0]: entry.get_target(response.get_request_uri())
for entry in link_header.parse(response.payload.decode("utf8")).links
}
return self._endpoints[rt]
@_skip_unless_linkheader
@asynctest
async def test_registration(self):
request = aiocoap.Message(
code=aiocoap.POST,
uri=(await self._get_endpoint("core.rd")) + "?ep=node1",
content_format=40,
payload=b'</sensors/temp>;ct=41;rt="temperature-c";if="sensor",</sensors/light>;ct=41;rt="light-lux";if="sensor"',
)
response = await self.client.request(request).response
self.assertEqual(
response.code, aiocoap.CREATED, "Registration did not result in Created"
)
self.assertTrue(
len(response.opt.location_path) > 0,
"Registration did not result in non-empty registration resource",
)
# FIXME: there are many more things to be tested here
|
