xref: /unit/test/test_configuration.py (revision 1913:3e9b04cd33bc)
1import socket
2
3import pytest
4
5from unit.control import TestControl
6
7
8class TestConfiguration(TestControl):
9    prerequisites = {'modules': {'python': 'any'}}
10
11    def try_addr(self, addr):
12        return self.conf(
13            {
14                "listeners": {addr: {"pass": "routes"}},
15                "routes": [{"action": {"return": 200}}],
16                "applications": {},
17            }
18        )
19
20    def test_json_empty(self):
21        assert 'error' in self.conf(''), 'empty'
22
23    def test_json_leading_zero(self):
24        assert 'error' in self.conf('00'), 'leading zero'
25
26    def test_json_unicode(self):
27        assert 'success' in self.conf(
28            u"""
29            {
30                "ap\u0070": {
31                    "type": "\u0070ython",
32                    "processes": { "spare": 0 },
33                    "path": "\u002Fapp",
34                    "module": "wsgi"
35                }
36            }
37            """,
38            'applications',
39        ), 'unicode'
40
41        assert self.conf_get('applications') == {
42            "app": {
43                "type": "python",
44                "processes": {"spare": 0},
45                "path": "/app",
46                "module": "wsgi",
47            }
48        }, 'unicode get'
49
50    def test_json_unicode_2(self):
51        assert 'success' in self.conf(
52            {
53                "приложение": {
54                    "type": "python",
55                    "processes": {"spare": 0},
56                    "path": "/app",
57                    "module": "wsgi",
58                }
59            },
60            'applications',
61        ), 'unicode 2'
62
63        assert 'приложение' in self.conf_get('applications'), 'unicode 2 get'
64
65    def test_json_unicode_number(self):
66        assert 'success' in self.conf(
67            u"""
68            {
69                "app": {
70                    "type": "python",
71                    "processes": { "spare": \u0030 },
72                    "path": "/app",
73                    "module": "wsgi"
74                }
75            }
76            """,
77            'applications',
78        ), 'unicode number'
79
80    def test_json_utf8_bom(self):
81        assert 'success' in self.conf(
82            b"""\xEF\xBB\xBF
83            {
84                "app": {
85                    "type": "python",
86                    "processes": {"spare": 0},
87                    "path": "/app",
88                    "module": "wsgi"
89                }
90            }
91            """,
92            'applications',
93        ), 'UTF-8 BOM'
94
95    def test_json_comment_single_line(self):
96        assert 'success' in self.conf(
97            b"""
98            // this is bridge
99            {
100                "//app": {
101                    "type": "python", // end line
102                    "processes": {"spare": 0},
103                    // inside of block
104                    "path": "/app",
105                    "module": "wsgi"
106                }
107                // double //
108            }
109            // end of json \xEF\t
110            """,
111            'applications',
112        ), 'single line comments'
113
114    def test_json_comment_multi_line(self):
115        assert 'success' in self.conf(
116            b"""
117            /* this is bridge */
118            {
119                "/*app": {
120                /**
121                 * multiple lines
122                 **/
123                    "type": "python",
124                    "processes": /* inline */ {"spare": 0},
125                    "path": "/app",
126                    "module": "wsgi"
127                    /*
128                    // end of block */
129                }
130                /* blah * / blah /* blah */
131            }
132            /* end of json \xEF\t\b */
133            """,
134            'applications',
135        ), 'multi line comments'
136
137    def test_json_comment_invalid(self):
138        assert 'error' in self.conf(b'/{}', 'applications'), 'slash'
139        assert 'error' in self.conf(b'//{}', 'applications'), 'comment'
140        assert 'error' in self.conf(b'{} /', 'applications'), 'slash end'
141        assert 'error' in self.conf(b'/*{}', 'applications'), 'slash star'
142        assert 'error' in self.conf(b'{} /*', 'applications'), 'slash star end'
143
144    def test_applications_open_brace(self):
145        assert 'error' in self.conf('{', 'applications'), 'open brace'
146
147    def test_applications_string(self):
148        assert 'error' in self.conf('"{}"', 'applications'), 'string'
149
150    @pytest.mark.skip('not yet, unsafe')
151    def test_applications_type_only(self):
152        assert 'error' in self.conf(
153            {"app": {"type": "python"}}, 'applications'
154        ), 'type only'
155
156    def test_applications_miss_quote(self):
157        assert 'error' in self.conf(
158            """
159            {
160                app": {
161                    "type": "python",
162                    "processes": { "spare": 0 },
163                    "path": "/app",
164                    "module": "wsgi"
165                }
166            }
167            """,
168            'applications',
169        ), 'miss quote'
170
171    def test_applications_miss_colon(self):
172        assert 'error' in self.conf(
173            """
174            {
175                "app" {
176                    "type": "python",
177                    "processes": { "spare": 0 },
178                    "path": "/app",
179                    "module": "wsgi"
180                }
181            }
182            """,
183            'applications',
184        ), 'miss colon'
185
186    def test_applications_miss_comma(self):
187        assert 'error' in self.conf(
188            """
189            {
190                "app": {
191                    "type": "python"
192                    "processes": { "spare": 0 },
193                    "path": "/app",
194                    "module": "wsgi"
195                }
196            }
197            """,
198            'applications',
199        ), 'miss comma'
200
201    def test_applications_skip_spaces(self):
202        assert 'success' in self.conf(
203            b'{ \n\r\t}', 'applications'
204        ), 'skip spaces'
205
206    def test_applications_relative_path(self):
207        assert 'success' in self.conf(
208            {
209                "app": {
210                    "type": "python",
211                    "processes": {"spare": 0},
212                    "path": "../app",
213                    "module": "wsgi",
214                }
215            },
216            'applications',
217        ), 'relative path'
218
219    @pytest.mark.skip('not yet, unsafe')
220    def test_listeners_empty(self):
221        assert 'error' in self.conf(
222            {"*:7080": {}}, 'listeners'
223        ), 'listener empty'
224
225    def test_listeners_no_app(self):
226        assert 'error' in self.conf(
227            {"*:7080": {"pass": "applications/app"}}, 'listeners'
228        ), 'listeners no app'
229
230    def test_listeners_addr(self):
231        assert 'success' in self.try_addr("*:7080"), 'wildcard'
232        assert 'success' in self.try_addr("127.0.0.1:7081"), 'explicit'
233        assert 'success' in self.try_addr("[::1]:7082"), 'explicit ipv6'
234
235    def test_listeners_addr_error(self):
236        assert 'error' in self.try_addr("127.0.0.1"), 'no port'
237
238    def test_listeners_addr_error_2(self, skip_alert):
239        skip_alert(r'bind.*failed', r'failed to apply new conf')
240
241        assert 'error' in self.try_addr(
242            "[f607:7403:1e4b:6c66:33b2:843f:2517:da27]:7080"
243        )
244
245    def test_listeners_port_release(self):
246        for i in range(10):
247            fail = False
248            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
249                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
250
251                self.conf(
252                    {
253                        "listeners": {"127.0.0.1:7080": {"pass": "routes"}},
254                        "routes": [],
255                    }
256                )
257
258                resp = self.conf({"listeners": {}, "applications": {}})
259
260                try:
261                    s.bind(('127.0.0.1', 7080))
262                    s.listen()
263
264                except OSError:
265                    fail = True
266
267                if fail:
268                    pytest.fail('cannot bind or listen to the address')
269
270                assert 'success' in resp, 'port release'
271
272    def test_json_application_name_large(self):
273        name = "X" * 1024 * 1024
274
275        assert 'success' in self.conf(
276            {
277                "listeners": {"*:7080": {"pass": "applications/" + name}},
278                "applications": {
279                    name: {
280                        "type": "python",
281                        "processes": {"spare": 0},
282                        "path": "/app",
283                        "module": "wsgi",
284                    }
285                },
286            }
287        )
288
289    @pytest.mark.skip('not yet')
290    def test_json_application_many(self):
291        apps = 999
292
293        conf = {
294            "applications": {
295                "app-"
296                + str(a): {
297                    "type": "python",
298                    "processes": {"spare": 0},
299                    "path": "/app",
300                    "module": "wsgi",
301                }
302                for a in range(apps)
303            },
304            "listeners": {
305                "*:" + str(7000 + a): {"pass": "applications/app-" + str(a)}
306                for a in range(apps)
307            },
308        }
309
310        assert 'success' in self.conf(conf)
311
312    def test_json_application_many2(self):
313        conf = {
314            "applications": {
315                "app-"
316                + str(a): {
317                    "type": "python",
318                    "processes": {"spare": 0},
319                    "path": "/app",
320                    "module": "wsgi",
321                }
322                # Larger number of applications can cause test fail with default
323                # open files limit due to the lack of file descriptors.
324                for a in range(100)
325            },
326            "listeners": {"*:7080": {"pass": "applications/app-1"}},
327        }
328
329        assert 'success' in self.conf(conf)
330
331    def test_unprivileged_user_error(self, is_su, skip_alert):
332        skip_alert(r'cannot set user "root"', r'failed to apply new conf')
333        if is_su:
334            pytest.skip('unprivileged tests')
335
336        assert 'error' in self.conf(
337            {
338                "app": {
339                    "type": "external",
340                    "processes": 1,
341                    "executable": "/app",
342                    "user": "root",
343                }
344            },
345            'applications',
346        ), 'setting user'
347