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