|
| 1 | +import copy |
| 2 | + |
| 3 | + |
| 4 | + |
| 5 | +import pytest |
| 6 | + |
| 7 | + |
| 8 | +@pytest.fixture(autouse=True) |
| 9 | +def _reset_key_state(app_module): |
| 10 | + original_live = copy.deepcopy(app_module.API_KEYS['pp_test_suite_key']) |
| 11 | + original_test = copy.deepcopy(app_module.API_KEYS['pp_test_suite_key_test']) |
| 12 | + yield |
| 13 | + app_module.API_KEYS['pp_test_suite_key'].clear() |
| 14 | + app_module.API_KEYS['pp_test_suite_key'].update(original_live) |
| 15 | + app_module.API_KEYS['pp_test_suite_key_test'].clear() |
| 16 | + app_module.API_KEYS['pp_test_suite_key_test'].update(original_test) |
| 17 | + |
| 18 | +def _wallet_stub(chain, address): |
| 19 | + return { |
| 20 | + 'chain': chain, |
| 21 | + 'address_type': 'personal_wallet', |
| 22 | + 'rpc_used': False, |
| 23 | + 'details': 'stubbed for limits/policy tests', |
| 24 | + } |
| 25 | + |
| 26 | + |
| 27 | +def _preflight_payload(network='ethereum', asset='USDT', address='0x1111111111111111111111111111111111111111'): |
| 28 | + return { |
| 29 | + 'expected': { |
| 30 | + 'network': network, |
| 31 | + 'asset': asset, |
| 32 | + 'address': address, |
| 33 | + }, |
| 34 | + 'provided': { |
| 35 | + 'network': network, |
| 36 | + 'asset': asset, |
| 37 | + 'address': address, |
| 38 | + }, |
| 39 | + } |
| 40 | + |
| 41 | + |
| 42 | +def test_monthly_check_quota_blocks_second_request(client, api_headers, app_module, monkeypatch): |
| 43 | + monkeypatch.setattr(app_module, 'classify_address', _wallet_stub) |
| 44 | + app_module.API_KEYS['pp_test_suite_key']['limits'] = {'monthly_checks': 1} |
| 45 | + |
| 46 | + first = client.post('/api/preflight-check', json=_preflight_payload(), headers=api_headers) |
| 47 | + second = client.post('/api/preflight-check', json=_preflight_payload(), headers=api_headers) |
| 48 | + account = client.get('/api/account', headers=api_headers) |
| 49 | + |
| 50 | + assert first.status_code == 200 |
| 51 | + assert second.status_code == 429 |
| 52 | + second_body = second.get_json() |
| 53 | + account_body = account.get_json() |
| 54 | + |
| 55 | + assert second_body['error'] == 'QUOTA_EXCEEDED' |
| 56 | + assert second_body['details']['counter'] == 'monthly_checks' |
| 57 | + assert second_body['details']['remaining'] == 0 |
| 58 | + assert account.status_code == 200 |
| 59 | + assert account_body['limits']['items']['monthly_checks']['limit'] == 1 |
| 60 | + assert account_body['limits']['items']['monthly_checks']['used'] == 1 |
| 61 | + assert account_body['limits']['items']['monthly_checks']['remaining'] == 0 |
| 62 | + assert account_body['limits']['items']['monthly_checks']['exceeded'] is True |
| 63 | + |
| 64 | + |
| 65 | + |
| 66 | +def test_policy_origin_allowlist_blocks_unapproved_origin(client, api_headers, app_module, monkeypatch): |
| 67 | + monkeypatch.setattr(app_module, 'classify_address', _wallet_stub) |
| 68 | + app_module.API_KEYS['pp_test_suite_key']['policy'] = { |
| 69 | + 'allowed_origin_hosts': ['allowed.client.test'], |
| 70 | + } |
| 71 | + |
| 72 | + response = client.post( |
| 73 | + '/api/preflight-check', |
| 74 | + json=_preflight_payload(), |
| 75 | + headers={**api_headers, 'Origin': 'https://blocked.example'}, |
| 76 | + ) |
| 77 | + |
| 78 | + assert response.status_code == 403 |
| 79 | + body = response.get_json() |
| 80 | + assert body['error'] == 'POLICY_ORIGIN_NOT_ALLOWED' |
| 81 | + assert 'allowed_origin_hosts' in body['details'] |
| 82 | + |
| 83 | + |
| 84 | + |
| 85 | +def test_policy_network_allowlist_blocks_disallowed_network(client, api_headers, app_module, monkeypatch): |
| 86 | + monkeypatch.setattr(app_module, 'classify_address', _wallet_stub) |
| 87 | + app_module.API_KEYS['pp_test_suite_key']['policy'] = { |
| 88 | + 'allowed_networks': ['ethereum'], |
| 89 | + } |
| 90 | + |
| 91 | + payload = _preflight_payload(network='arbitrum') |
| 92 | + response = client.post('/api/preflight-check', json=payload, headers=api_headers) |
| 93 | + |
| 94 | + assert response.status_code == 403 |
| 95 | + body = response.get_json() |
| 96 | + assert body['error'] == 'POLICY_NETWORK_NOT_ALLOWED' |
| 97 | + assert 'arbitrum' in body['details']['disallowed_networks'] |
| 98 | + |
| 99 | + |
| 100 | + |
| 101 | +def test_viewer_role_is_read_only_for_write_endpoints(client, api_headers_test, app_module, monkeypatch): |
| 102 | + monkeypatch.setattr(app_module, 'classify_address', _wallet_stub) |
| 103 | + response = client.post('/api/preflight-check', json=_preflight_payload(), headers=api_headers_test) |
| 104 | + |
| 105 | + assert response.status_code == 403 |
| 106 | + body = response.get_json() |
| 107 | + assert body['error'] == 'ROLE_READ_ONLY' |
| 108 | + assert body['details']['role'] == 'viewer' |
0 commit comments