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