File: test_resource_extend.py

package info (click to toggle)
python-neutron-lib 3.21.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 7,660 kB
  • sloc: python: 22,829; sh: 137; makefile: 24
file content (65 lines) | stat: -rw-r--r-- 1,922 bytes parent folder | download | duplicates (2)
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
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from oslotest import base

from neutron_lib.db import resource_extend
from neutron_lib import fixture


@resource_extend.has_resource_extenders
class _DBExtender:

    @resource_extend.extends('ExtendedA')
    def _extend_a(self, resp, db_obj):
        pass

    @resource_extend.extends('ExtendedB')
    def _extend_b(self, resp, db_obj):
        pass


class TestResourceExtendClass(base.BaseTestCase):

    def test_extends(self):
        self.assertIsNotNone(resource_extend.get_funcs('ExtendedA'))
        self.assertIsNotNone(resource_extend.get_funcs('ExtendedB'))


class TestResourceExtend(base.BaseTestCase):

    def setUp(self):
        super().setUp()
        self.useFixture(fixture.DBResourceExtendFixture())

    def test_register_funcs(self):
        resources = ['A', 'B', 'C']
        for r in resources:
            resource_extend.register_funcs(r, (lambda x: x,))

        for r in resources:
            self.assertIsNotNone(resource_extend.get_funcs(r))

    def test_apply_funcs(self):
        resources = ['A', 'B', 'C']
        callbacks = []

        def _cb(resp, db_obj):
            callbacks.append(resp)

        for r in resources:
            resource_extend.register_funcs(r, (_cb,))

        for r in resources:
            resource_extend.apply_funcs(r, None, None)

        self.assertEqual(3, len(callbacks))