Test Failed
Pull Request — master (#375)
by Vinicius
05:01
created

tests.unit.test_core.test_api_server   C

Complexity

Total Complexity 53

Size/Duplication

Total Lines 519
Duplicated Lines 0 %

Test Coverage

Coverage 100%

Importance

Changes 0
Metric Value
eloc 370
dl 0
loc 519
ccs 314
cts 314
cp 1
rs 6.96
c 0
b 0
f 0
wmc 53

1 Function

Rating   Name   Duplication   Size   Complexity  
A test_custom_encoder() 0 5 1

40 Methods

Rating   Name   Duplication   Size   Complexity  
A TestAPIServer.setup_method() 0 14 1
A TestAPIServer.test_enable_napp__success() 0 7 1
A TestAPIServer.test_update_web_ui_error() 0 33 2
A TestAPIServer.test_get_napp_metadata() 0 8 1
A TestAPIServer.test_fetch_latest_ui_tag() 0 11 1
A TestAPIServer.test_fetch_latest_ui_tag_fallback() 0 9 1
A TestAPIServer.test_install_napp__http_error() 0 10 1
A TestAPIServer.test_uninstall_napp__success_not_installed() 0 8 1
A TestAPIServer.test_status_api() 0 5 1
A TestAPIServer.test_disable_napp__error_not_enabling() 0 7 1
A TestAPIServer.test_web_ui() 0 14 4
A TestAPIServer.test_get_ui_components() 0 12 1
A TestAPIDecorator.test_route_ordering() 0 29 1
A TestAPIServer.test_enable_napp__error_not_installed() 0 6 1
A TestAPIDecorator.test_route_from_classmethod() 0 15 1
A TestAPIDecorator.test_sync_route() 0 22 2
A TestAPIServer.test_install_napp__error_not_installing() 0 7 1
A TestAPIServer.test_install_napp__success() 0 7 1
A TestAPIDecorator.test_remove_napp_endpoints() 0 18 1
A TestAPIServer.test_install_napp__success_is_installed() 0 6 1
A TestAPIServer.test_static_web_ui() 0 12 3
A TestAPIServer.test_list_enabled_napps() 0 9 1
A TestAPIDecorator.test_route_from_staticmethod() 0 15 1
A RESTNApp.__init__() 0 4 1
A TestAPIServer.test_get_napp_metadata__not_installed() 0 7 1
A TestAPIServer.test_get_napp_metadata__invalid_key() 0 6 1
A TestAPIServer.test_list_installed_napps() 0 9 1
A TestAPIServer.test_unzip_backup_web_ui_error() 0 11 2
A TestAPIDecorator.test_async_route() 0 16 1
A TestAPIServer.test_deprecation_warning() 0 10 3
A TestAPIServer.test_disable_napp__success() 0 7 1
A TestAPIServer.test_disable_napp__error_not_installed() 0 6 1
A TestAPIServer.test_enable_napp__error_not_enabling() 0 7 1
A TestAPIServer.test_unzip_backup_web_ui() 0 21 2
A TestAPIServer.test_uninstall_napp__error_not_uninstalling() 0 9 1
A TestAPIServer.test_run() 0 4 1
A TestAPIServer.test_uninstall_napp__success() 0 7 1
A TestAPIServer.test_update_web_ui() 0 24 2
A TestAPIServer.test_stop_api_server() 0 4 1
A TestAPIServer.test_run_error() 0 7 1

How to fix   Complexity   

Complexity

Complex classes like tests.unit.test_core.test_api_server often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
"""APIServer tests."""
2 1
import asyncio
3 1
import json
4 1
import warnings
5 1
from datetime import datetime
6
# Disable not-grouped imports that conflicts with isort
7 1
from unittest.mock import MagicMock
8
from urllib.error import HTTPError, URLError
9 1
10
import pytest
11 1
from httpx import AsyncClient, RequestError
12 1
13
from kytos.core.api_server import APIServer
14 1
from kytos.core.napps import rest
15 1
from kytos.core.rest_api import JSONResponse, Request
16
17
18 1
def test_custom_encoder() -> None:
19
    """Test json custom encoder."""
20 1
    some_datetime = datetime(year=2022, month=1, day=30)
21 1
    resp = JSONResponse(some_datetime)
22 1
    assert json.loads(resp.body.decode()) == "2022-01-30T00:00:00"
23
24
25
# pylint: disable=protected-access,attribute-defined-outside-init
26 1
class TestAPIServer:
27
    """Test the class APIServer."""
28
29 1
    def setup_method(self):
30
        """Instantiate a APIServer."""
31 1
        self.api_server = APIServer()
32 1
        self.app = self.api_server.app
33 1
        self.napps_manager = MagicMock()
34 1
        self.api_server.server = MagicMock()
35 1
        self.api_server.napps_manager = self.napps_manager
36 1
        self.api_server.napps_dir = 'napps_dir'
37
        self.api_server.flask_dir = 'flask_dir'
38 1
        self.api_server.start_api()
39
        self.api_server.app.router.mount = MagicMock()
40 1
        self.api_server.start_web_ui()
41 1
        base_url = "http://127.0.0.1/api/kytos/core/"
42 1
        self.client = AsyncClient(app=self.app, base_url=base_url)
43
44 1
    def test_deprecation_warning(self):
45 1
        """Deprecated method should suggest @rest decorator."""
46 1
        with warnings.catch_warnings(record=True) as wrngs:
47 1
            warnings.simplefilter("always")  # trigger all warnings
48
            self.api_server.register_rest_endpoint(
49 1
                'rule', lambda x: x, ['POST'])
50
            assert 1 == len(wrngs)
51 1
            warning = wrngs[0]
52
            assert warning.category == DeprecationWarning
53 1
            assert '@rest' in str(warning.message)
54
55
    def test_run(self):
56
        """Test run method."""
57 1
        self.api_server.run()
58 1
        self.api_server.server.run.assert_called_with()
59
60 1
    def test_run_error(self, monkeypatch):
61 1
        """Test run method to error case."""
62
        mock_exit = MagicMock()
63 1
        monkeypatch.setattr("sys.exit", mock_exit)
64
        self.api_server.server.run.side_effect = OSError
65 1
        self.api_server.run()
66
        mock_exit.assert_called()
67 1
68 1
    async def test_status_api(self):
69 1
        """Test status_api method."""
70 1
        response = await self.client.get("status/")
71
        assert response.status_code == 200
72 1
        assert response.json() == {"response": "running"}
73
74 1
    def test_stop_api_server(self):
75 1
        """Test stop_api_server method."""
76 1
        self.api_server.stop_api_server()
77 1
        assert self.api_server.server.should_exit
78
79 1
    @pytest.mark.parametrize("file_exist", [True, False])
80
    async def test_static_web_ui(self, monkeypatch, file_exist):
81 1
        """Test static_web_ui method to success case."""
82 1
        monkeypatch.setattr("os.path.exists", lambda x: file_exist)
83
        monkeypatch.setattr("kytos.core.api_server.FileResponse",
84 1
                            lambda x: JSONResponse({}))
85 1
86
        endpoint = f"/ui/kytos/napp/{self.api_server.napps_dir}"
87 1
        client = AsyncClient(app=self.app, base_url="http://127.0.0.1")
88
        response = await client.get(endpoint)
89 1
        expected_status_code = {True: 200, False: 404}
90 1
        assert expected_status_code[file_exist] == response.status_code
91
92 1
    async def test_get_ui_components(self, monkeypatch):
93 1
        """Test get_ui_components method."""
94 1
        mock_glob = MagicMock()
95
        mock_glob.return_value = ['napps_dir/*/*/ui/*/*.kytos']
96 1
        monkeypatch.setattr("kytos.core.api_server.glob", mock_glob)
97 1
        expected_json = [{'name': '*-*-*-*', 'url': 'ui/*/*/*/*.kytos'}]
98
99 1
        endpoint = "/ui/k-toolbar/"
100
        client = AsyncClient(app=self.app, base_url="http://127.0.0.1")
101 1
        resp = await client.get(endpoint)
102 1
        assert resp.status_code == 200
103
        assert resp.json() == expected_json
104 1
105
    @pytest.mark.parametrize("file_exist", [True, False])
106 1
    async def test_web_ui(self, monkeypatch, file_exist):
107 1
        """Test web_ui method."""
108
        monkeypatch.setattr("os.path.exists", lambda x: file_exist)
109 1
        monkeypatch.setattr("kytos.core.api_server.FileResponse",
110 1
                            lambda x: JSONResponse({}))
111
112 1
        client = AsyncClient(app=self.app, base_url="http://127.0.0.1")
113 1
        coros = [client.get("/index.html"), client.get("/")]
114 1
        coros = asyncio.gather(*coros)
115
        responses = await coros
116 1
        expected_status_code = {True: 200, False: 404}
117 1
        for response in responses:
118 1
            assert expected_status_code[file_exist] == response.status_code
119
120 1
    def test_unzip_backup_web_ui(self, monkeypatch) -> None:
121 1
        """Test _unzip_backup_web_ui."""
122 1
        mock_log, mock_shutil = MagicMock(), MagicMock()
123
        mock_zipfile, zipfile = MagicMock(), MagicMock()
124 1
        zip_data, mock_mkdir = MagicMock(), MagicMock()
125 1
        zipfile.__enter__.return_value = zip_data
126
        zip_data.testzip.return_value = None
127 1
        mock_zipfile.return_value = zipfile
128
129 1
        monkeypatch.setattr("zipfile.ZipFile", mock_zipfile)
130 1
        monkeypatch.setattr("os.path.exists", lambda x: True)
131
        monkeypatch.setattr("os.mkdir", mock_mkdir)
132 1
        monkeypatch.setattr("shutil.move", mock_shutil)
133 1
        monkeypatch.setattr("kytos.core.api_server.LOG", mock_log)
134
        package, uri = "/tmp/file", "http://localhost/some_file.zip"
135 1
        self.api_server._unzip_backup_web_ui(package, uri)
136
        assert mock_mkdir.call_count == 1
137 1
        assert mock_shutil.call_count == 1
138 1
        assert zip_data.extractall.call_count == 1
139 1
        assert zip_data.close.call_count == 1
140 1
        assert mock_log.info.call_count == 1
141 1
142 1
    def test_unzip_backup_web_ui_error(self, monkeypatch) -> None:
143 1
        """Test _unzip_backup_web_ui error."""
144
        mock_log, mock_zipfile, zipdata = MagicMock(), MagicMock(), MagicMock()
145 1
        mock_zipfile.__enter__.return_value = zipdata
146
        monkeypatch.setattr("zipfile.ZipFile", mock_zipfile)
147 1
        monkeypatch.setattr("kytos.core.api_server.LOG", mock_log)
148 1
        package, uri = "/tmp/file", "http://localhost/some_file.zip"
149 1
        with pytest.raises(ValueError) as exc:
150 1
            self.api_server._unzip_backup_web_ui(package, uri)
151 1
        assert "is corrupted" in str(exc)
152
        assert mock_log.error.call_count == 1
153 1
154 1
    def test_fetch_latest_ui_tag(self, monkeypatch) -> None:
155 1
        """Test fetch lastest ui."""
156
        mock_get, mock_res, ver = MagicMock(), MagicMock(), "X.Y.Z"
157 1
        mock_get.return_value = mock_res
158
        mock_res.json.return_value = {"tag_name": ver}
159 1
        monkeypatch.setattr("kytos.core.api_server.httpx.get", mock_get)
160
        version = self.api_server._fetch_latest_ui_tag()
161 1
        assert version == ver
162
        url = 'https://api.github.com/repos/kytos-ng/' \
163 1
              'ui/releases/latest'
164
        mock_get.assert_called_with(url, timeout=10)
165 1
166
    def test_fetch_latest_ui_tag_fallback(self, monkeypatch) -> None:
167 1
        """Test fetch lastest ui fallback tag."""
168
        mock_get, mock_log = MagicMock(), MagicMock()
169 1
        mock_get.side_effect = RequestError("some error")
170 1
        monkeypatch.setattr("kytos.core.api_server.httpx.get", mock_get)
171 1
        monkeypatch.setattr("kytos.core.api_server.LOG", mock_log)
172
        version = self.api_server._fetch_latest_ui_tag()
173 1
        assert version == "2022.3.0"
174 1
        assert mock_log.warning.call_count == 1
175 1
176 1
    async def test_update_web_ui(self, monkeypatch):
177
        """Test update_web_ui method."""
178 1
        mock_log, mock_urlretrieve = MagicMock(), MagicMock()
179
        mock_urlretrieve.return_value = ["/tmp/file"]
180 1
        monkeypatch.setattr("os.path.exists", lambda x: False)
181 1
        monkeypatch.setattr("kytos.core.api_server.LOG", mock_log)
182 1
        monkeypatch.setattr("kytos.core.api_server.urlretrieve",
183
                            mock_urlretrieve)
184 1
185
        version = "2022.3.0"
186 1
        mock_fetch = MagicMock(return_value=version)
187
        self.api_server._fetch_latest_ui_tag = mock_fetch
188 1
        self.api_server._unzip_backup_web_ui = MagicMock()
189 1
        response = await self.client.post("web/update")
190
        assert response.status_code == 200
191
        assert response.json() == "Web UI was updated"
192 1
193
        assert self.api_server._fetch_latest_ui_tag.call_count == 1
194 1
        assert self.api_server._unzip_backup_web_ui.call_count == 1
195
        assert mock_log.info.call_count == 2
196 1
197
        url = 'https://github.com/kytos-ng/ui/releases/' + \
198 1
              f'download/{version}/latest.zip'
199 1
        mock_urlretrieve.assert_called_with(url)
200
201 1
    @pytest.mark.parametrize(
202 1
        "err_name,exp_msg",
203 1
        [
204 1
            ("HTTPError", "Uri not found"),
205 1
            ("URLError", "Error accessing"),
206
        ],
207
    )
208 1
    async def test_update_web_ui_error(self, monkeypatch, err_name, exp_msg):
209 1
        """Test update_web_ui method error.
210 1
211 1
        HTTPError had issues when being initialized in the decorator args,
212
        so it it's being instantiated dynamically in the test body, see:
213 1
        https://bugs.python.org/issue45955
214
        """
215 1
        mock_log, mock_urlretrieve = MagicMock(), MagicMock()
216 1
        http_error = HTTPError("errorx", 500, "errorx", {}, None)
217
        errs = {"HTTPError": http_error, "URLError": URLError("some reason")}
218 1
        mock_urlretrieve.side_effect = errs[err_name]
219
        monkeypatch.setattr("os.path.exists", lambda x: False)
220 1
        monkeypatch.setattr("kytos.core.api_server.LOG", mock_log)
221
        monkeypatch.setattr("kytos.core.api_server.urlretrieve",
222 1
                            mock_urlretrieve)
223
224 1
        version = "2022.3.0"
225
        self.api_server._unzip_backup_web_ui = MagicMock()
226 1
        response = await self.client.post(f"web/update/{version}")
227
        assert response.status_code == 500
228
        assert exp_msg in response.json()
229 1
        assert mock_log.error.call_count == 1
230 1
231
        url = 'https://github.com/kytos-ng/ui/releases/' + \
232 1
              f'download/{version}/latest.zip'
233
        mock_urlretrieve.assert_called_with(url)
234 1
235
    async def test_enable_napp__error_not_installed(self):
236 1
        """Test _enable_napp method error case when napp is not installed."""
237
        self.napps_manager.is_installed.return_value = False
238 1
        resp = await self.client.get("/napps/kytos/napp/enable")
239 1
        assert resp.status_code == 400
240
        assert resp.json() == {"response": "not installed"}
241 1
242
    async def test_enable_napp__error_not_enabling(self):
243 1
        """Test _enable_napp method error case when napp is not enabling."""
244 1
        self.napps_manager.is_installed.return_value = True
245
        self.napps_manager.is_enabled.side_effect = [False, False]
246 1
        resp = await self.client.get("napps/kytos/napp/enable")
247
        assert resp.status_code == 500
248 1
        assert resp.json() == {"response": "error"}
249 1
250
    async def test_enable_napp__success(self):
251 1
        """Test _enable_napp method success case."""
252
        self.napps_manager.is_installed.return_value = True
253 1
        self.napps_manager.is_enabled.side_effect = [False, True]
254 1
        resp = await self.client.get("napps/kytos/napp/enable")
255
        assert resp.status_code == 200
256 1
        assert resp.json() == {"response": "enabled"}
257
258 1
    async def test_disable_napp__error_not_installed(self):
259 1
        """Test _disable_napp method error case when napp is not installed."""
260
        self.napps_manager.is_installed.return_value = False
261 1
        resp = await self.client.get("napps/kytos/napp/disable")
262
        assert resp.status_code == 400
263 1
        assert resp.json() == {"response": "not installed"}
264
265 1
    async def test_disable_napp__error_not_enabling(self):
266
        """Test _disable_napp method error case when napp is not enabling."""
267 1
        self.napps_manager.is_installed.return_value = True
268 1
        self.napps_manager.is_enabled.side_effect = [True, True]
269
        resp = await self.client.get("napps/kytos/napp/disable")
270 1
        assert resp.status_code == 500
271
        assert resp.json() == {"response": "error"}
272 1
273 1
    async def test_disable_napp__success(self):
274
        """Test _disable_napp method success case."""
275 1
        self.napps_manager.is_installed.return_value = True
276
        self.napps_manager.is_enabled.side_effect = [True, False]
277 1
        resp = await self.client.get("napps/kytos/napp/disable")
278 1
        assert resp.status_code == 200
279
        assert resp.json() == {"response": "disabled"}
280 1
281
    async def test_install_napp__error_not_installing(self):
282 1
        """Test _install_napp method error case when napp is not installing."""
283 1
        self.napps_manager.is_installed.return_value = False
284
        self.napps_manager.install.return_value = False
285 1
        resp = await self.client.get("napps/kytos/napp/install")
286
        assert resp.status_code == 500
287 1
        assert resp.json() == {"response": "error"}
288 1
289
    async def test_install_napp__http_error(self):
290 1
        """Test _install_napp method to http error case."""
291
        self.napps_manager.is_installed.return_value = False
292 1
        self.napps_manager.install.side_effect = HTTPError('url', 123, 'msg',
293 1
                                                           'hdrs', MagicMock())
294
295 1
        resp = await self.client.get("napps/kytos/napp/install")
296
        assert resp.status_code == 123
297 1
        assert resp.json() == {"response": "error"}
298 1
        self.napps_manager.install.side_effect = None
299
300 1
    async def test_install_napp__success_is_installed(self):
301
        """Test _install_napp method success case when napp is installed."""
302 1
        self.napps_manager.is_installed.return_value = True
303 1
        resp = await self.client.get("napps/kytos/napp/install")
304
        assert resp.status_code == 200
305
        assert resp.json() == {"response": "installed"}
306 1
307
    async def test_install_napp__success(self):
308 1
        """Test _install_napp method success case."""
309 1
        self.napps_manager.is_installed.return_value = False
310
        self.napps_manager.install.return_value = True
311 1
        resp = await self.client.get("napps/kytos/napp/install")
312
        assert resp.status_code == 200
313 1
        assert resp.json() == {"response": "installed"}
314
315 1
    async def test_uninstall_napp__error_not_uninstalling(self):
316
        """Test _uninstall_napp method error case when napp is not
317 1
           uninstalling.
318 1
        """
319
        self.napps_manager.is_installed.return_value = True
320 1
        self.napps_manager.uninstall.return_value = False
321
        resp = await self.client.get("napps/kytos/napp/uninstall")
322 1
        assert resp.status_code == 500
323 1
        assert resp.json() == {"response": "error"}
324
325 1
    async def test_uninstall_napp__success_not_installed(self):
326
        """Test _uninstall_napp method success case when napp is not
327 1
           installed.
328 1
        """
329
        self.napps_manager.is_installed.return_value = False
330 1
        resp = await self.client.get("napps/kytos/napp/uninstall")
331
        assert resp.status_code == 200
332
        assert resp.json() == {"response": "uninstalled"}
333
334 1
    async def test_uninstall_napp__success(self):
335 1
        """Test _uninstall_napp method success case."""
336
        self.napps_manager.is_installed.return_value = True
337 1
        self.napps_manager.uninstall.return_value = True
338
        resp = await self.client.get("napps/kytos/napp/uninstall")
339 1
        assert resp.status_code == 200
340 1
        assert resp.json() == {"response": "uninstalled"}
341
342 1
    async def test_list_enabled_napps(self):
343
        """Test _list_enabled_napps method."""
344
        napp = MagicMock()
345
        napp.username = "kytos"
346 1
        napp.name = "name"
347
        self.napps_manager.get_enabled_napps.return_value = [napp]
348 1
        resp = await self.client.get("napps_enabled")
349
        assert resp.status_code == 200
350 1
        assert resp.json() == {"napps": [["kytos", "name"]]}
351 1
352
    async def test_list_installed_napps(self):
353 1
        """Test _list_installed_napps method."""
354
        napp = MagicMock()
355 1
        napp.username = "kytos"
356 1
        napp.name = "name"
357
        self.napps_manager.get_installed_napps.return_value = [napp]
358 1
        resp = await self.client.get("napps_installed")
359
        assert resp.status_code == 200
360 1
        assert resp.json() == {"napps": [["kytos", "name"]]}
361 1
362
    async def test_get_napp_metadata__not_installed(self):
363 1
        """Test _get_napp_metadata method to error case when napp is not
364
           installed."""
365 1
        self.napps_manager.is_installed.return_value = False
366 1
        resp = await self.client.get("napps/kytos/napp/metadata/version")
367 1
        assert resp.status_code == 400
368 1
        assert resp.json() == "NApp is not installed."
369
370 1
    async def test_get_napp_metadata__invalid_key(self):
371
        """Test _get_napp_metadata method to error case when key is invalid."""
372 1
        self.napps_manager.is_installed.return_value = True
373 1
        resp = await self.client.get("napps/kytos/napp/metadata/any")
374
        assert resp.status_code == 400
375 1
        assert resp.json() == "Invalid key."
376
377 1
    async def test_get_napp_metadata(self):
378 1
        """Test _get_napp_metadata method."""
379 1
        value = "1.0"
380 1
        self.napps_manager.is_installed.return_value = True
381
        self.napps_manager.get_napp_metadata.return_value = value
382 1
        resp = await self.client.get("napps/kytos/napp/metadata/version")
383
        assert resp.status_code == 200
384 1
        assert resp.json() == {"version": value}
385 1
386
387 1
class RESTNApp:  # pylint: disable=too-few-public-methods
388
    """Bare minimum for the decorator to work. Not a functional NApp."""
389
390 1
    def __init__(self):
391 1
        self.username = 'test'
392
        self.name = 'MyNApp'
393
        self.napp_id = 'test/MyNApp'
394 1
395 1
396
class TestAPIDecorator:
397 1
    """Test suite for @rest decorator."""
398
399 1
    async def test_sync_route(self, controller, api_client):
400 1
        """Test rest decorator sync route."""
401
        class MyNApp(RESTNApp):
402
            """API decorator example usage."""
403 1
404 1
            @rest("some_route/", methods=["POST"])
405
            def some_endpoint(self, _request: Request) -> JSONResponse:
406 1
                """Some endpoint."""
407
                return JSONResponse({"some_response": "some_value"})
408 1
409
        napp = MyNApp()
410
        controller.api_server.register_napp_endpoints(napp)
411 1
412 1
        routes = ["test/MyNApp/some_route/", "test/MyNApp/some_route"]
413 1
        coros = [api_client.post(route) for route in routes]
414
        responses = await asyncio.gather(*coros)
415
        for resp in responses:
416 1
            assert resp.status_code == 200
417 1
            assert resp.json() == {"some_response": "some_value"}
418 1
419
        resp = await api_client.get("test/MyNApp/some_route/")
420
        assert resp.status_code == 405
421 1
422
    async def test_async_route(self, controller, api_client):
423
        """Test rest decorator async route."""
424 1
        class MyNApp(RESTNApp):
425 1
            """API decorator example usage."""
426 1
427 1
            @rest("some_route/", methods=["POST"])
428
            async def some_endpoint(self, _request: Request) -> JSONResponse:
429
                """Some endpoint."""
430 1
                await asyncio.sleep(0)
431
                return JSONResponse({})
432
433 1
        napp = MyNApp()
434 1
        controller.api_server.register_napp_endpoints(napp)
435 1
        resp = await api_client.post("test/MyNApp/some_route/")
436
        assert resp.status_code == 200
437 1
        assert resp.json() == {}
438
439 1
    async def test_route_ordering(self, controller, api_client):
440
        """Test rest decorator route ordering."""
441 1
        class MyNApp(RESTNApp):
442
            """API decorator example usage."""
443
444 1
            @rest("some_route/hello", methods=["POST"])
445 1
            async def hello(self, _request: Request) -> JSONResponse:
446
                """Hello endpoint."""
447
                await asyncio.sleep(0)
448 1
                return JSONResponse({"hello": "world"})
449 1
450
            @rest("some_route/{some_arg}", methods=["POST"])
451 1
            async def ahello_arg(self, request: Request) -> JSONResponse:
452 1
                """Hello endpoint."""
453 1
                arg = request.path_params["some_arg"]
454
                await asyncio.sleep(0)
455
                return JSONResponse({"hello": arg})
456 1
457 1
        napp = MyNApp()
458
        controller.api_server.register_napp_endpoints(napp)
459 1
        coros = [
460
            api_client.post("test/MyNApp/some_route/abc"),
461
            api_client.post("test/MyNApp/some_route/hello")
462 1
        ]
463 1
        responses = await asyncio.gather(*coros)
464 1
        assert responses[0].status_code == 200
465 1
        assert responses[1].status_code == 200
466
        assert responses[0].json() == {"hello": "abc"}
467 1
        assert responses[1].json() == {"hello": "world"}
468 1
469
    async def test_remove_napp_endpoints(self, controller, api_client):
470 1
        """Test remove napp endpoints."""
471 1
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
472 1
            """API decorator example usage."""
473 1
474 1
            @rest("some_route/", methods=["POST"])
475
            def some_endpoint(self, _request: Request) -> JSONResponse:
476 1
                """Some endpoint."""
477 1
                return JSONResponse({})
478
479 1
        napp = MyNApp()
480
        controller.api_server.register_napp_endpoints(napp)
481
        resp = await api_client.post("test/MyNApp/some_route/")
482 1
        assert resp.status_code == 200
483 1
484
        controller.api_server.remove_napp_endpoints(napp)
485 1
        resp = await api_client.post("test/MyNApp/some_route/")
486
        assert resp.status_code == 404
487 1
488
    async def test_route_from_classmethod(self, controller, api_client):
489 1
        """Test route from classmethod."""
490 1
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
491 1
            """API decorator example usage."""
492
493 1
            @rest("some_route/", methods=["POST"])
494
            @classmethod
495
            def some_endpoint(cls, _request: Request) -> JSONResponse:
496 1
                """Some endpoint."""
497 1
                return JSONResponse({})
498
499
        napp = MyNApp()
500 1
        controller.api_server.register_napp_endpoints(napp)
501 1
        resp = await api_client.post("test/MyNApp/some_route/")
502 1
        assert resp.status_code == 200
503
504 1
    async def test_route_from_staticmethod(self, controller, api_client):
505 1
        """Test route from staticmethod."""
506 1
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
507
            """API decorator example usage."""
508 1
509
            @rest("some_route/", methods=["POST"])
510
            @staticmethod
511 1
            def some_endpoint(_request: Request) -> JSONResponse:
512 1
                """Some endpoint."""
513 1
                return JSONResponse({})
514
515
        napp = MyNApp()
516 1
        controller.api_server.register_napp_endpoints(napp)
517 1
        resp = await api_client.post("test/MyNApp/some_route/")
518
        assert resp.status_code == 200
519