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