File: test_compiled_router.py

package info (click to toggle)
python-falcon 4.0.2-2
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 5,172 kB
  • sloc: python: 33,608; javascript: 92; sh: 50; makefile: 50
file content (147 lines) | stat: -rw-r--r-- 3,548 bytes parent folder | download
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
from threading import Barrier
from threading import Thread
from time import sleep
from unittest.mock import MagicMock

import pytest

from falcon.routing import compiled
from falcon.routing import CompiledRouter
from falcon.routing import CompiledRouterOptions


def test_find_src(monkeypatch):
    called = False
    find = CompiledRouter.find

    def mock(*args):
        nonlocal called
        called = True
        find(*args)

    monkeypatch.setattr(CompiledRouter, 'find', mock)
    router = CompiledRouter()

    assert router.finder_src is not None
    assert called


@pytest.fixture
def patch_add_route(monkeypatch):
    add_route = CompiledRouter.add_route
    mock = MagicMock(side_effect=lambda *a, **k: add_route(router, *a, **k))
    monkeypatch.setattr(CompiledRouter, 'add_route', mock)
    router = CompiledRouter()
    return mock, router


@pytest.mark.parametrize(
    'kwargs',
    (
        {},
        {'compile': False},
        {'compile': False, 'suffix': 'other'},
        {'suffix': 'other'},
    ),
)
def test_no_compile_kw(patch_add_route, kwargs):
    mock, router = patch_add_route

    res = MockResource()
    router.add_route('/foo', res, **kwargs)

    assert mock.call_count == 1
    mock.assert_has_calls(((('/foo', res), kwargs),))
    assert router._find == router._compile_and_find


def test_compile(patch_add_route):
    mock, router = patch_add_route

    res = MockResource()
    router.add_route('/foo', res, compile=True)
    assert mock.call_count == 1
    assert router._find != router._compile_and_find


def test_add_route_after_first_request():
    router = CompiledRouter()

    router.add_route('/foo', MockResource())
    assert router.find('/foo') is not None
    assert router._find != router._compile_and_find

    router.add_route('/bar', MockResource(), suffix='other')
    assert router._find == router._compile_and_find
    assert router.find('/bar') is not None
    assert router._find != router._compile_and_find


def test_multithread_compile(monkeypatch):
    def side_effect():
        sleep(0.05)
        return lambda *args: None

    mock = MagicMock(side_effect=side_effect)
    monkeypatch.setattr(CompiledRouter, '_compile', mock)

    router = CompiledRouter()
    mr = MockResource()

    router.add_route('/foo', mr)

    calls = 0
    num_threads = 3
    barrier = Barrier(num_threads)

    def find():
        nonlocal calls
        barrier.wait()
        assert router.find('/foo') is None
        calls += 1

    threads = [Thread(target=find) for i in range(num_threads)]
    for t in threads:
        t.start()

    for t in threads:
        t.join()

    assert calls == 3
    mock.assert_called_once_with()


class MockResource:
    def on_get(self, req, res):
        pass

    def on_get_other(self, req, res):
        pass


def test_cannot_replace_compiled():
    opt = CompiledRouterOptions()
    with pytest.raises(AttributeError, match='Cannot set'):
        opt.converters = {}
    with pytest.raises(AttributeError, match='object has no attribute'):
        opt.other = 123


def test_converter_not_subclass():
    class X:
        def convert(self, v):
            return v

    router = CompiledRouter()
    router.options.converters['x'] = X

    router.add_route('/foo/{bar:x}', MockResource())
    res = router.find('/foo/bar')
    assert res is not None
    assert res[2] == {'bar': 'bar'}
    assert router.find('/foo/bar/bar') is None


def test_base_classes():
    with pytest.raises(NotImplementedError):
        compiled._CxChild().src(42)