How to use load_module method in Radish

Best Python code snippet using radish

Run Radish automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

test_plugin_registry.py

Source: test_plugin_registry.py Github

copy
1import pytest
2from unittest.mock import call
3
4from asynctest import TestCase as AsyncTestCase, mock as async_mock, call
5
6from ...config.injection_context import InjectionContext
7from ...utils.classloader import ClassLoader, ModuleLoadError
8
9from ..plugin_registry import PluginRegistry
10from ..protocol_registry import ProtocolRegistry
11
12from ..error import ProtocolDefinitionValidationError
13
14
15class TestPluginRegistry(AsyncTestCase):
16    def setUp(self):
17        self.registry = PluginRegistry()
18
19        self.context = InjectionContext(enforce_typing=False)
20        self.proto_registry = async_mock.MagicMock(
21            register_message_types=async_mock.MagicMock(),
22            register_controllers=async_mock.MagicMock(),
23        )
24        self.context.injector.bind_instance(ProtocolRegistry, self.proto_registry)
25
26    async def test_setup(self):
27        mod_name = "test_mod"
28        mod = async_mock.MagicMock()
29        mod.__name__ = mod_name
30        ctx = async_mock.MagicMock()
31        self.registry._plugins[mod_name] = mod
32        assert list(self.registry.plugin_names) == [mod_name]
33        assert list(self.registry.plugins) == [mod]
34        mod.setup = async_mock.CoroutineMock()
35        await self.registry.init_context(ctx)
36        mod.setup.assert_awaited_once_with(ctx)
37
38    async def test_register_routes(self):
39        mod_name = "test_mod"
40        mod = async_mock.MagicMock()
41        mod.__name__ = mod_name
42        app = async_mock.MagicMock()
43        self.registry._plugins[mod_name] = mod
44        mod.routes.register = async_mock.CoroutineMock()
45        definition = async_mock.MagicMock()
46        definition.versions = [
47            {
48                "major_version": 1,
49                "minimum_minor_version": 0,
50                "current_minor_version": 0,
51                "path": "v1_0",
52            }
53        ]
54
55        with async_mock.patch.object(
56            ClassLoader,
57            "load_module",
58            async_mock.MagicMock(side_effect=[definition, mod.routes]),
59        ) as load_module:
60            await self.registry.register_admin_routes(app)
61
62            calls = [
63                call("definition", mod_name),
64                call(f"{mod_name}.{definition.versions[0]['path']}.routes"),
65            ]
66            load_module.assert_has_calls(calls)
67        assert mod.routes.register.call_count == 1
68
69        with async_mock.patch.object(
70            ClassLoader,
71            "load_module",
72            async_mock.MagicMock(side_effect=[definition, ModuleLoadError()]),
73        ) as load_module:
74            await self.registry.register_admin_routes(app)
75
76            calls = [
77                call("definition", mod_name),
78                call(f"{mod_name}.{definition.versions[0]['path']}.routes"),
79            ]
80            load_module.assert_has_calls(calls)
81        assert mod.routes.register.call_count == 1
82
83    async def test_register_routes_mod_no_version(self):
84        mod_name = "test_mod"
85        mod = async_mock.MagicMock()
86        mod.__name__ = mod_name
87        app = async_mock.MagicMock()
88        self.registry._plugins[mod_name] = mod
89        mod.routes.register = async_mock.CoroutineMock()
90
91        with async_mock.patch.object(
92            ClassLoader,
93            "load_module",
94            async_mock.MagicMock(side_effect=[None, mod.routes]),
95        ) as load_module:
96            await self.registry.register_admin_routes(app)
97
98            calls = [call("definition", mod_name), call(f"{mod_name}.routes")]
99            load_module.assert_has_calls(calls)
100        assert mod.routes.register.call_count == 1
101
102        with async_mock.patch.object(
103            ClassLoader,
104            "load_module",
105            async_mock.MagicMock(side_effect=[None, ModuleLoadError()]),
106        ) as load_module:
107            await self.registry.register_admin_routes(app)
108
109            calls = [
110                call("definition", mod_name),
111                call(f"{mod_name}.routes"),
112            ]
113            load_module.assert_has_calls(calls)
114        assert mod.routes.register.call_count == 1
115
116    async def test_post_process_routes(self):
117        mod_name = "test_mod"
118        mod = async_mock.MagicMock()
119        mod.__name__ = mod_name
120        app = async_mock.MagicMock()
121        self.registry._plugins[mod_name] = mod
122        mod.routes.post_process_routes = async_mock.MagicMock()
123        definition = async_mock.MagicMock()
124        definition.versions = [
125            {
126                "major_version": 1,
127                "minimum_minor_version": 0,
128                "current_minor_version": 0,
129                "path": "v1_0",
130            }
131        ]
132
133        with async_mock.patch.object(
134            ClassLoader,
135            "load_module",
136            async_mock.MagicMock(side_effect=[definition, mod.routes]),
137        ) as load_module:
138            self.registry.post_process_routes(app)
139
140            calls = [
141                call("definition", mod_name),
142                call(f"{mod_name}.{definition.versions[0]['path']}.routes"),
143            ]
144            load_module.assert_has_calls(calls)
145        assert mod.routes.post_process_routes.call_count == 1
146
147        with async_mock.patch.object(
148            ClassLoader,
149            "load_module",
150            async_mock.MagicMock(side_effect=[definition, ModuleLoadError()]),
151        ) as load_module:
152            self.registry.post_process_routes(app)
153
154            calls = [
155                call("definition", mod_name),
156                call(f"{mod_name}.{definition.versions[0]['path']}.routes"),
157            ]
158            load_module.assert_has_calls(calls)
159        assert mod.routes.post_process_routes.call_count == 1
160
161    async def test_post_process_routes_mod_no_version(self):
162        mod_name = "test_mod"
163        mod = async_mock.MagicMock()
164        mod.__name__ = mod_name
165        app = async_mock.MagicMock()
166        self.registry._plugins[mod_name] = mod
167        mod.routes.register = async_mock.CoroutineMock()
168
169        with async_mock.patch.object(
170            ClassLoader,
171            "load_module",
172            async_mock.MagicMock(side_effect=[None, mod.routes]),
173        ) as load_module:
174            self.registry.post_process_routes(app)
175
176            calls = [call("definition", mod_name), call(f"{mod_name}.routes")]
177            load_module.assert_has_calls(calls)
178        assert mod.routes.post_process_routes.call_count == 1
179
180        with async_mock.patch.object(
181            ClassLoader,
182            "load_module",
183            async_mock.MagicMock(side_effect=[None, ModuleLoadError()]),
184        ) as load_module:
185            self.registry.post_process_routes(app)
186
187            calls = [call("definition", mod_name), call(f"{mod_name}.routes")]
188            load_module.assert_has_calls(calls)
189        assert mod.routes.post_process_routes.call_count == 1
190
191    async def test_validate_version_not_a_list(self):
192        mod_name = "test_mod"
193        mod = async_mock.MagicMock()
194        mod.__name__ = mod_name
195
196        versions_not_a_list = {}
197
198        with async_mock.patch.object(
199            ClassLoader, "load_module", async_mock.MagicMock()
200        ) as load_module:
201            with pytest.raises(ProtocolDefinitionValidationError):
202                self.registry.validate_version(versions_not_a_list, mod_name)
203
204    async def test_validate_version_list_element_not_an_object(self):
205        mod_name = "test_mod"
206        mod = async_mock.MagicMock()
207        mod.__name__ = mod_name
208
209        versions = [{}, []]
210
211        with async_mock.patch.object(
212            ClassLoader, "load_module", async_mock.MagicMock()
213        ) as load_module:
214            with pytest.raises(ProtocolDefinitionValidationError):
215                self.registry.validate_version(versions, mod_name)
216
217    async def test_validate_version_list_element_empty(self):
218        mod_name = "test_mod"
219        mod = async_mock.MagicMock()
220        mod.__name__ = mod_name
221
222        versions = []
223
224        with async_mock.patch.object(
225            ClassLoader, "load_module", async_mock.MagicMock()
226        ) as load_module:
227            with pytest.raises(ProtocolDefinitionValidationError):
228                self.registry.validate_version(versions, mod_name)
229
230    async def test_validate_version_list_missing_attribute(self):
231        mod_name = "test_mod"
232        mod = async_mock.MagicMock()
233        mod.__name__ = mod_name
234
235        versions = [
236            {
237                "major_version": 1,
238                "minimum_minor_version": 0,
239                "current_minor_version": 0,
240                # "path": "v1_0", # missing
241            }
242        ]
243
244        with async_mock.patch.object(
245            ClassLoader, "load_module", async_mock.MagicMock()
246        ) as load_module:
247            with pytest.raises(ProtocolDefinitionValidationError):
248                self.registry.validate_version(versions, mod_name)
249
250    async def test_validate_version_negative_version(self):
251        mod_name = "test_mod"
252        mod = async_mock.MagicMock()
253        mod.__name__ = mod_name
254
255        versions = [
256            {
257                "major_version": -1,
258                "minimum_minor_version": 0,
259                "current_minor_version": 0,
260                "path": "v1_0",
261            }
262        ]
263
264        with async_mock.patch.object(
265            ClassLoader, "load_module", async_mock.MagicMock()
266        ) as load_module:
267            with pytest.raises(ProtocolDefinitionValidationError):
268                self.registry.validate_version(versions, mod_name)
269
270    async def test_validate_version_min_greater_current(self):
271        mod_name = "test_mod"
272        mod = async_mock.MagicMock()
273        mod.__name__ = mod_name
274
275        versions = [
276            {
277                "major_version": 1,
278                "minimum_minor_version": 1,
279                "current_minor_version": 0,
280                "path": "v1_0",
281            }
282        ]
283
284        with async_mock.patch.object(
285            ClassLoader, "load_module", async_mock.MagicMock()
286        ) as load_module:
287            with pytest.raises(ProtocolDefinitionValidationError):
288                self.registry.validate_version(versions, mod_name)
289
290    async def test_validate_version_multiple_major(self):
291        mod_name = "test_mod"
292        mod = async_mock.MagicMock()
293        mod.__name__ = mod_name
294
295        versions = [
296            {
297                "major_version": 1,
298                "minimum_minor_version": 0,
299                "current_minor_version": 0,
300                "path": "v1_0",
301            },
302            {
303                "major_version": 1,
304                "minimum_minor_version": 0,
305                "current_minor_version": 1,
306                "path": "v1_1",
307            },
308        ]
309
310        with async_mock.patch.object(
311            ClassLoader, "load_module", async_mock.MagicMock()
312        ) as load_module:
313            with pytest.raises(ProtocolDefinitionValidationError):
314                self.registry.validate_version(versions, mod_name)
315
316    async def test_validate_version_bad_path(self):
317        mod_name = "test_mod"
318        mod = async_mock.MagicMock()
319        mod.__name__ = mod_name
320
321        versions = [
322            {
323                "major_version": 1,
324                "minimum_minor_version": 1,
325                "current_minor_version": 0,
326                "path": "v1_0",
327            }
328        ]
329
330        with async_mock.patch.object(
331            ClassLoader, "load_module", async_mock.MagicMock(return_value=None)
332        ) as load_module:
333            with pytest.raises(ProtocolDefinitionValidationError):
334                self.registry.validate_version(versions, mod_name)
335
336    async def test_validate_version_list_correct(self):
337        mod_name = "test_mod"
338        mod = async_mock.MagicMock()
339        mod.__name__ = mod_name
340
341        versions = [
342            {
343                "major_version": 1,
344                "minimum_minor_version": 0,
345                "current_minor_version": 0,
346                "path": "v1_0",
347            },
348            {
349                "major_version": 2,
350                "minimum_minor_version": 0,
351                "current_minor_version": 0,
352                "path": "v2_0",
353            },
354        ]
355
356        with async_mock.patch.object(
357            ClassLoader, "load_module", async_mock.MagicMock()
358        ) as load_module:
359            assert self.registry.validate_version(versions, mod_name) is True
360
361            load_module.has_calls(
362                call(versions[0]["path"], mod_name),
363                call(versions[1]["path"], mod_name),
364            )
365
366    async def test_validate_version_list_extra_attributes_ok(self):
367        mod_name = "test_mod"
368        mod = async_mock.MagicMock()
369        mod.__name__ = mod_name
370
371        versions = [
372            {
373                "major_version": 1,
374                "minimum_minor_version": 0,
375                "current_minor_version": 0,
376                "path": "v1_0",
377                "not": "an attribute",
378            }
379        ]
380
381        with async_mock.patch.object(
382            ClassLoader, "load_module", async_mock.MagicMock()
383        ) as load_module:
384            assert self.registry.validate_version(versions, mod_name) is True
385
386    async def test_validate_version_no_such_mod(self):
387        mod_name = "no_mod"
388        mod = async_mock.MagicMock()
389        mod.__name__ = mod_name
390
391        versions = [
392            {
393                "major_version": 1,
394                "minimum_minor_version": 0,
395                "current_minor_version": 0,
396                "path": "v1_0",
397            }
398        ]
399
400        with async_mock.patch.object(
401            ClassLoader, "load_module", async_mock.MagicMock()
402        ) as load_module:
403            load_module.return_value = None
404
405            with self.assertRaises(ProtocolDefinitionValidationError):
406                self.registry.validate_version(versions, mod_name)
407
408    async def test_register_plugin_already_present(self):
409        mod_name = "test_mod"
410        mod = async_mock.MagicMock()
411        self.registry._plugins[mod_name] = mod
412        assert mod == self.registry.register_plugin(mod_name)
413
414    async def test_register_plugin_load_x(self):
415        with async_mock.patch.object(
416            ClassLoader, "load_module", async_mock.MagicMock()
417        ) as load_module:
418            load_module.side_effect = ModuleLoadError("failure to load")
419            assert self.registry.register_plugin("dummy") is None
420
421    async def test_register_plugin_no_mod(self):
422        with async_mock.patch.object(
423            ClassLoader, "load_module", async_mock.MagicMock()
424        ) as load_module:
425            load_module.return_value = None
426            assert self.registry.register_plugin("dummy") is None
427
428    async def test_register_plugin_no_definition(self):
429        class MODULE:
430            no_setup = "no setup attr"
431
432        obj = MODULE()
433        with async_mock.patch.object(
434            ClassLoader, "load_module", async_mock.MagicMock()
435        ) as load_module:
436            load_module.side_effect = [
437                obj,  # module
438                None,  # routes
439                None,  # message types
440                None,  # definition
441            ]
442            assert self.registry.register_plugin("dummy") is None
443
444    async def test_register_plugin_no_versions(self):
445        class MODULE:
446            no_setup = "no setup attr"
447
448        obj = MODULE()
449        with async_mock.patch.object(
450            ClassLoader, "load_module", async_mock.MagicMock()
451        ) as load_module:
452            load_module.side_effect = [
453                obj,  # module
454                None,  # routes
455                None,  # message types
456                "str-has-no-versions-attr",  # definition without versions attr
457            ]
458            assert self.registry.register_plugin("dummy") is None
459
460    async def test_register_plugin_has_setup(self):
461        class MODULE:
462            setup = "present"
463
464        obj = MODULE()
465        with async_mock.patch.object(
466            ClassLoader, "load_module", async_mock.MagicMock()
467        ) as load_module:
468            load_module.side_effect = [
469                obj,  # module
470                None,  # routes
471                None,  # message types
472                None,  # definition without versions attr
473            ]
474            assert self.registry.register_plugin("dummy") == obj
475
476    async def test_register_definitions_malformed(self):
477        class MODULE:
478            no_setup = "no setup attr"
479
480        obj = MODULE()
481        with async_mock.patch.object(
482            ClassLoader, "load_module", async_mock.MagicMock()
483        ) as load_module:
484            load_module.side_effect = [
485                obj,  # module
486                None,  # routes
487                None,  # message types
488                async_mock.MagicMock(versions="not-a-list"),
489            ]
490            assert self.registry.register_plugin("dummy") is None
491
492    async def test_register_package_x(self):
493        with async_mock.patch.object(
494            ClassLoader, "scan_subpackages", async_mock.MagicMock()
495        ) as load_module:
496            load_module.side_effect = ModuleLoadError()
497            assert not self.registry.register_package("dummy")
498
499    async def test_load_protocols_load_x(self):
500        mock_plugin = async_mock.MagicMock(__name__="dummy")
501        with async_mock.patch.object(
502            ClassLoader, "load_module", async_mock.MagicMock()
503        ) as load_module:
504            load_module.side_effect = ModuleLoadError()
505            await self.registry.load_protocols(None, mock_plugin)
506            assert load_module.call_count == 1
507
508    async def test_load_protocols_load_mod(self):
509        mock_plugin = async_mock.MagicMock(__name__="dummy")
510        mock_mod = async_mock.MagicMock()
511        mock_mod.MESSAGE_TYPES = async_mock.MagicMock()
512        mock_mod.CONTROLLERS = async_mock.MagicMock()
513
514        with async_mock.patch.object(
515            ClassLoader, "load_module", async_mock.MagicMock()
516        ) as load_module:
517            load_module.return_value = mock_mod
518            await self.registry.load_protocols(self.context, mock_plugin)
519
520    async def test_load_protocols_no_mod_load_x(self):
521        mock_plugin = async_mock.MagicMock(__name__="dummy")
522
523        with async_mock.patch.object(
524            ClassLoader, "load_module", async_mock.MagicMock()
525        ) as load_module:
526            load_module.side_effect = [None, ModuleLoadError()]
527            await self.registry.load_protocols(self.context, mock_plugin)
528            assert load_module.call_count == 2
529
530    async def test_load_protocols_no_mod_def_no_message_types(self):
531        mock_plugin = async_mock.MagicMock(__name__="dummy")
532        mock_def = async_mock.MagicMock(
533            versions=[
534                {
535                    "major_version": 1,
536                    "minimum_minor_version": 0,
537                    "current_minor_version": 0,
538                    "path": "v1_0",
539                }
540            ]
541        )
542
543        with async_mock.patch.object(
544            ClassLoader, "load_module", async_mock.MagicMock()
545        ) as load_module:
546            load_module.side_effect = [None, mock_def, ModuleLoadError()]
547            await self.registry.load_protocols(self.context, mock_plugin)
548            assert load_module.call_count == 3
549
550    async def test_load_protocols_no_mod_def_message_types(self):
551        mock_plugin = async_mock.MagicMock(__name__="dummy")
552        mock_def = async_mock.MagicMock(
553            versions=[
554                {
555                    "major_version": 1,
556                    "minimum_minor_version": 0,
557                    "current_minor_version": 0,
558                    "path": "v1_0",
559                },
560                {
561                    "major_version": 2,
562                    "minimum_minor_version": 0,
563                    "current_minor_version": 0,
564                    "path": "v2_0",
565                },
566            ]
567        )
568        mock_mod = async_mock.MagicMock()
569        mock_mod.MESSAGE_TYPES = async_mock.MagicMock()
570        mock_mod.CONTROLLERS = async_mock.MagicMock()
571
572        with async_mock.patch.object(
573            ClassLoader, "load_module", async_mock.MagicMock()
574        ) as load_module:
575            load_module.side_effect = [None, mock_def, mock_mod, mock_mod]
576            await self.registry.load_protocols(self.context, mock_plugin)
577            assert load_module.call_count == 4
578
579    def test_repr(self):
580        assert type(repr(self.registry)) is str
581
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Radish on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)