Skip to content

Instantly share code, notes, and snippets.

@julienr
Created February 16, 2022 16:51
Show Gist options
  • Save julienr/fe7825cbfbd5a6797aedc79383733a37 to your computer and use it in GitHub Desktop.
Save julienr/fe7825cbfbd5a6797aedc79383733a37 to your computer and use it in GitHub Desktop.
"""
This is a local proxy to intercept GDAL vsigs request (when storing data
on google cloud storage) and inject error codes to exercise the logging
codepath.
Usage:
$ python vsi_proxy.py
$ CPL_GS_ENDPOINT=http://localhost:9999/ gdal_translate -srcwin 5000 5000 2000 2000 /vsigs/mybucketfile.tif out.tit
"""
from http.server import BaseHTTPRequestHandler
from http.server import HTTPServer
import requests
from urllib.error import HTTPError
import urllib.request
import shutil
import logging
import sys
import random
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
ENDPOINT = 'https://storage.googleapis.com'
class ProxyHandler(BaseHTTPRequestHandler):
def do_GET(self): # noqa
headers = dict(self.headers)
headers['Host'] = 'storage.googleapis.com'
# Fail some range requests
if 'Range' in headers and random.random() < 0.5:
print('\nFAILING range request\n')
self.send_response(403)
self.end_headers()
self.wfile.write('Unauthorized'.encode('utf-8'))
else:
print('path=%s, headers=%s' % (self.path, headers))
resp = requests.request('GET', ENDPOINT + self.path, headers=headers)
self.send_response(resp.status_code)
#print('response code=%s , headers: ' % resp.status_code)
for k, v in resp.headers.items():
#print(k, v)
self.send_header(k, v)
self.end_headers()
self.wfile.write(resp.content)
mock_server = HTTPServer(('localhost', 9999), ProxyHandler)
mock_server.serve_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment