1"""
2Common test fixtures
3####################
4
5Copyright (c) 2023 Nordic Semiconductor ASA
6
7SPDX-License-Identifier: Apache-2.0
8
9"""
10
11import time
12import logging
13import os
14import binascii
15import random
16import string
17import pytest
18from leshan import Leshan
19
20from twister_harness import Shell
21from twister_harness import DeviceAdapter
22
23LESHAN_IP: str = '192.0.2.2'
24COAP_PORT: int = 5683
25COAPS_PORT: int = 5684
26BOOTSTRAP_COAPS_PORT: int = 5784
27
28logger = logging.getLogger(__name__)
29
30class Endpoint:
31    def __init__(self, name: str, shell: Shell, registered: bool = False, bootstrap: bool = False):
32        self.name = name
33        self.registered = registered
34        self.bootstrap = bootstrap
35        self.shell = shell
36        self.last_update = 0.0
37
38    def check_update(self):
39        if not self.registered:
40            return
41        if self.last_update < time.time() - 5:
42            self.shell.exec_command('lwm2m update')
43            self.last_update = time.time()
44
45    def __str__(self):
46        return self.name
47
48
49@pytest.fixture(scope='session')
50def leshan() -> Leshan:
51    """
52    Fixture that returns a Leshan object for interacting with the Leshan server.
53
54    :return: The Leshan object.
55    :rtype: Leshan
56    """
57    try:
58        return Leshan("http://localhost:8080/api")
59    except RuntimeError:
60        pytest.skip('Leshan server not available')
61
62@pytest.fixture(scope='session')
63def leshan_bootstrap() -> Leshan:
64    """
65    Fixture that returns a Leshan object for interacting with the Bootstrap Leshan server.
66
67    :return: The Leshan object.
68    :rtype: Leshan
69    """
70    try:
71        return Leshan("http://localhost:8081/api")
72    except RuntimeError:
73        pytest.skip('Leshan Bootstrap server not available')
74
75@pytest.fixture(scope='module')
76def helperclient() -> object:
77    """
78    Fixture that returns a helper client object for testing.
79
80    :return: The helper client object.
81    :rtype: object
82    """
83    try:
84        from coapthon.client.helperclient import HelperClient
85    except ModuleNotFoundError:
86        pytest.skip('CoAPthon3 package not installed')
87    return HelperClient(server=('127.0.0.1', COAP_PORT))
88
89
90@pytest.fixture(scope='module')
91def endpoint_nosec(shell: Shell, dut: DeviceAdapter, leshan: Leshan) -> str:
92    """Fixture that returns an endpoint that starts on no-secure mode"""
93    # Allow engine to start & stop once.
94    time.sleep(2)
95
96    # Generate randon device id and password (PSK key)
97    ep = 'client_' + binascii.b2a_hex(os.urandom(1)).decode()
98
99    #
100    # Registration Interface test cases (using Non-secure mode)
101    #
102    shell.exec_command(f'lwm2m write 0/0/0 -s coap://{LESHAN_IP}:{COAP_PORT}')
103    shell.exec_command('lwm2m write 0/0/1 -b 0')
104    shell.exec_command('lwm2m write 0/0/2 -u8 3')
105    shell.exec_command(f'lwm2m write 0/0/3 -s {ep}')
106    shell.exec_command('lwm2m create 1/0')
107    shell.exec_command('lwm2m write 0/0/10 -u16 1')
108    shell.exec_command('lwm2m write 1/0/0 -u16 1')
109    shell.exec_command('lwm2m write 1/0/1 -u32 86400')
110    shell.exec_command(f'lwm2m start {ep} -b 0')
111
112    yield Endpoint(ep, shell)
113
114    # All done
115    shell.exec_command('lwm2m stop')
116    dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0)
117
118@pytest.fixture(scope='module')
119def endpoint_bootstrap(shell: Shell, dut: DeviceAdapter, leshan: Leshan, leshan_bootstrap: Leshan) -> str:
120    """Fixture that returns an endpoint that starts the bootstrap."""
121    try:
122        # Generate randon device id and password (PSK key)
123        ep = 'client_' + binascii.b2a_hex(os.urandom(1)).decode()
124        bs_passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16))
125        passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16))
126
127        logger.debug('Endpoint: %s', ep)
128        logger.debug('Boostrap PSK: %s', binascii.b2a_hex(bs_passwd.encode()).decode())
129        logger.debug('PSK: %s', binascii.b2a_hex(passwd.encode()).decode())
130
131        # Create device entries in Leshan and Bootstrap server
132        leshan_bootstrap.create_bs_device(ep, f'coaps://{LESHAN_IP}:{COAPS_PORT}', bs_passwd, passwd)
133        leshan.create_psk_device(ep, passwd)
134
135        # Allow engine to start & stop once.
136        time.sleep(2)
137
138        # Write bootsrap server information and PSK keys
139        shell.exec_command(f'lwm2m write 0/0/0 -s coaps://{LESHAN_IP}:{BOOTSTRAP_COAPS_PORT}')
140        shell.exec_command('lwm2m write 0/0/1 -b 1')
141        shell.exec_command('lwm2m write 0/0/2 -u8 0')
142        shell.exec_command(f'lwm2m write 0/0/3 -s {ep}')
143        shell.exec_command(f'lwm2m write 0/0/5 -s {bs_passwd}')
144        shell.exec_command(f'lwm2m start {ep} -b 1')
145
146        yield Endpoint(ep, shell)
147
148        shell.exec_command('lwm2m stop')
149        dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0)
150
151    finally:
152        # Remove device and bootstrap information
153        # Leshan does not accept non-secure connection if device information is provided with PSK
154        leshan.delete_device(ep)
155        leshan_bootstrap.delete_bs_device(ep)
156
157@pytest.fixture(scope='module')
158def endpoint_registered(endpoint_bootstrap, dut: DeviceAdapter) -> str:
159    """Fixture that returns an endpoint that is registered."""
160    if not endpoint_bootstrap.registered:
161        dut.readlines_until(regex='.*Registration Done', timeout=5.0)
162        endpoint_bootstrap.bootstrap = True
163        endpoint_bootstrap.registered = True
164    return endpoint_bootstrap
165
166@pytest.fixture(scope='function')
167def endpoint(endpoint_registered) -> str:
168    """Fixture that returns an endpoint that is registered."""
169    endpoint_registered.check_update()
170    return endpoint_registered
171