xref: /unit/test/test_static_fallback.py (revision 1971:3410f9d2a662)
1import os
2from pathlib import Path
3
4import pytest
5from unit.applications.proto import TestApplicationProto
6
7
8class TestStaticFallback(TestApplicationProto):
9    prerequisites = {}
10
11    @pytest.fixture(autouse=True)
12    def setup_method_fixture(self, temp_dir):
13        os.makedirs(temp_dir + '/assets/dir')
14        Path(temp_dir + '/assets/index.html').write_text('0123456789')
15
16        os.makedirs(temp_dir + '/assets/403')
17        os.chmod(temp_dir + '/assets/403', 0o000)
18
19        self._load_conf(
20            {
21                "listeners": {
22                    "*:7080": {"pass": "routes"},
23                    "*:7081": {"pass": "routes"},
24                },
25                "routes": [{"action": {"share": temp_dir + "/assets$uri"}}],
26                "applications": {},
27            }
28        )
29
30        yield
31
32        try:
33            os.chmod(temp_dir + '/assets/403', 0o777)
34        except FileNotFoundError:
35            pass
36
37    def action_update(self, conf):
38        assert 'success' in self.conf(conf, 'routes/0/action')
39
40    def test_static_fallback(self):
41        self.action_update({"share": "/blah"})
42        assert self.get()['status'] == 404, 'bad path no fallback'
43
44        self.action_update({"share": "/blah", "fallback": {"return": 200}})
45
46        resp = self.get()
47        assert resp['status'] == 200, 'bad path fallback status'
48        assert resp['body'] == '', 'bad path fallback'
49
50    def test_static_fallback_valid_path(self, temp_dir):
51        self.action_update(
52            {"share": temp_dir + "/assets$uri", "fallback": {"return": 200}}
53        )
54        resp = self.get()
55        assert resp['status'] == 200, 'fallback status'
56        assert resp['body'] == '0123456789', 'fallback'
57
58        resp = self.get(url='/403/')
59        assert resp['status'] == 200, 'fallback status 403'
60        assert resp['body'] == '', 'fallback 403'
61
62        resp = self.post()
63        assert resp['status'] == 200, 'fallback status 405'
64        assert resp['body'] == '', 'fallback 405'
65
66        assert self.get(url='/dir')['status'] == 301, 'fallback status 301'
67
68    def test_static_fallback_nested(self):
69        self.action_update(
70            {
71                "share": "/blah",
72                "fallback": {
73                    "share": "/blah/blah",
74                    "fallback": {"return": 200},
75                },
76            }
77        )
78
79        resp = self.get()
80        assert resp['status'] == 200, 'fallback nested status'
81        assert resp['body'] == '', 'fallback nested'
82
83    def test_static_fallback_share(self, temp_dir):
84        self.action_update(
85            {"share": "/blah", "fallback": {"share": temp_dir + "/assets$uri"},}
86        )
87
88        resp = self.get()
89        assert resp['status'] == 200, 'fallback share status'
90        assert resp['body'] == '0123456789', 'fallback share'
91
92        resp = self.head()
93        assert resp['status'] == 200, 'fallback share status HEAD'
94        assert resp['body'] == '', 'fallback share HEAD'
95
96        assert (
97            self.get(url='/dir')['status'] == 301
98        ), 'fallback share status 301'
99
100    def test_static_fallback_proxy(self):
101        assert 'success' in self.conf(
102            [
103                {
104                    "match": {"destination": "*:7081"},
105                    "action": {"return": 200},
106                },
107                {
108                    "action": {
109                        "share": "/blah",
110                        "fallback": {"proxy": "http://127.0.0.1:7081"},
111                    }
112                },
113            ],
114            'routes',
115        ), 'configure fallback proxy route'
116
117        resp = self.get()
118        assert resp['status'] == 200, 'fallback proxy status'
119        assert resp['body'] == '', 'fallback proxy'
120
121    @pytest.mark.skip('not yet')
122    def test_static_fallback_proxy_loop(self, skip_alert):
123        skip_alert(
124            r'open.*/blah/index.html.*failed',
125            r'accept.*failed',
126            r'socket.*failed',
127            r'new connections are not accepted',
128        )
129
130        self.action_update(
131            {"share": "/blah", "fallback": {"proxy": "http://127.0.0.1:7080"}}
132        )
133        self.get(no_recv=True)
134
135        assert 'success' in self.conf_delete('listeners/*:7081')
136        self.get(read_timeout=1)
137
138    def test_static_fallback_invalid(self):
139        def check_error(conf):
140            assert 'error' in self.conf(conf, 'routes/0/action')
141
142        check_error({"share": "/blah", "fallback": {}})
143        check_error({"share": "/blah", "fallback": ""})
144        check_error({"return": 200, "fallback": {"share": "/blah"}})
145        check_error(
146            {"proxy": "http://127.0.0.1:7081", "fallback": {"share": "/blah"}}
147        )
148        check_error({"fallback": {"share": "/blah"}})
149