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