xref: /unit/test/test_static_fallback.py (revision 2066:242192963d93) 
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            {
86                "share": "/blah",
87                "fallback": {"share": temp_dir + "/assets$uri"},
88            }
89        )
90
91        resp = self.get()
92        assert resp['status'] == 200, 'fallback share status'
93        assert resp['body'] == '0123456789', 'fallback share'
94
95        resp = self.head()
96        assert resp['status'] == 200, 'fallback share status HEAD'
97        assert resp['body'] == '', 'fallback share HEAD'
98
99        assert (
100            self.get(url='/dir')['status'] == 301
101        ), 'fallback share status 301'
102
103    def test_static_fallback_proxy(self):
104        assert 'success' in self.conf(
105            [
106                {
107                    "match": {"destination": "*:7081"},
108                    "action": {"return": 200},
109                },
110                {
111                    "action": {
112                        "share": "/blah",
113                        "fallback": {"proxy": "http://127.0.0.1:7081"},
114                    }
115                },
116            ],
117            'routes',
118        ), 'configure fallback proxy route'
119
120        resp = self.get()
121        assert resp['status'] == 200, 'fallback proxy status'
122        assert resp['body'] == '', 'fallback proxy'
123
124    @pytest.mark.skip('not yet')
125    def test_static_fallback_proxy_loop(self, skip_alert):
126        skip_alert(
127            r'open.*/blah/index.html.*failed',
128            r'accept.*failed',
129            r'socket.*failed',
130            r'new connections are not accepted',
131        )
132
133        self.action_update(
134            {"share": "/blah", "fallback": {"proxy": "http://127.0.0.1:7080"}}
135        )
136        self.get(no_recv=True)
137
138        assert 'success' in self.conf_delete('listeners/*:7081')
139        self.get(read_timeout=1)
140
141    def test_static_fallback_invalid(self):
142        def check_error(conf):
143            assert 'error' in self.conf(conf, 'routes/0/action')
144
145        check_error({"share": "/blah", "fallback": {}})
146        check_error({"share": "/blah", "fallback": ""})
147        check_error({"return": 200, "fallback": {"share": "/blah"}})
148        check_error(
149            {"proxy": "http://127.0.0.1:7081", "fallback": {"share": "/blah"}}
150        )
151        check_error({"fallback": {"share": "/blah"}})
152