KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use proxy_server method in localstack
Best Python code snippet using localstack_python
backend_services.py
Source:backend_services.py 
1from django.contrib.auth import SESSION_KEY2from django.core.cache import cache3from django.conf import settings4from django.http import HttpResponse, HttpResponseServerError5from proxy_server.response import AJAX_REQUEST6import httplib, json, proxy_server7def invoke_backend_service(method, function_path, json_data=dict(), request=None, response_token=True, public=False, secure=False):8 error_message = None9 try:10 if public is False and request is None:11 error_message = 'A private web service must receive Django\'s request'12 raise Exception13 if response_token is True and request is None:14 error_message = 'A web service cannot expect a response token and not receive Django\'s request'15 raise Exception16 if not hasattr(settings, 'BACKEND_HOST'):17 error_message = 'No backend host and/or port specified'18 raise Exception19 if secure:20 if hasattr(settings, 'BACKEND_PORT'):21 conn = httplib.HTTPSConnection(settings.BACKEND_HOST, settings.BACKEND_PORT)22 else:23 conn = httplib.HTTPSConnection(settings.BACKEND_HOST)24 else:25 if hasattr(settings, 'BACKEND_PORT'):26 conn = httplib.HTTPConnection(settings.BACKEND_HOST, settings.BACKEND_PORT)27 else:28 conn = httplib.HTTPConnection(settings.BACKEND_HOST)29 headers = proxy_server.RESTFUL_HEADER30 headers[proxy_server.API_KEY] = settings.SECRET_KEY31 if request is not None:32 pk = cache.get(AJAX_REQUEST, None)33 if pk:34 request.user.pk = pk35 cache.delete(AJAX_REQUEST)36 headers[proxy_server.USER_TOKEN] = request.user.pk37 headers[proxy_server.CLIENT_IP] = request.META.get(proxy_server.HTTP_FROM)38 try:39 conn.request(method, function_path, json.dumps(json_data), headers)40 except:41 error_message = 'Could not connect to service'42 raise Exception43 response = conn.getresponse()44 response_data = response.read()45 conn.close()46 if response.status == 403:47 return 403, None48 if response.status == 204:49 if response_token is True:50 error_message = 'Backend server didn\'t respond with a token'51 raise Exception52 return 204, None53 else:54 try:55 response_json = json.loads(response_data)56 except:57 error_message = 'Unknown response format'58 raise Exception59 if response_token is True:60 user_dict = None61 if SESSION_KEY in request.session:62 user_dict = cache.get(request.session[SESSION_KEY])63 cache.delete(request.session[SESSION_KEY])64 request.session[SESSION_KEY] = response_json[proxy_server.USER_TOKEN]65 request.user.pk = response_json[proxy_server.USER_TOKEN]66 request.session[proxy_server.EXPIRATION_DATE] = response_json[proxy_server.EXPIRATION_DATE]67 if user_dict:68 user_dict['pk'] = request.user.pk69 cache.set(request.session[SESSION_KEY], user_dict)70 if response.status == 200:71 if response_token is True and proxy_server.USER_TOKEN not in response_json:72 error_message = 'Server expected user token in response'73 raise Exception74 result = None75 if proxy_server.RESPONSE in response_json:76 result = response_json[proxy_server.RESPONSE]77 return 200, result78 else:79 code = response.status80 if proxy_server.ERROR in response_json:81 error_message = response_json[proxy_server.ERROR][proxy_server.MESSAGE]82 raise Exception(code)83 else:84 error_message = response.reason85 raise Exception(code)86 except Exception as e:87 if error_message is None:88 error_message = 'Unknown error in service invocation'89 code = int(str(e)) if e is not None and isinstance(str(e), int) else 50090 error = {91 'error': {92 'code': code,93 'type': 'ProxyServerError',94 'message': error_message95 }96 }97 return code, error98def invoke_backend_service_as_proxy(request, method, function_path, json_data=dict(), response_token=True, secure=False):99 error_message = None100 try:101 if not hasattr(settings, 'BACKEND_HOST'):102 error_message = 'No backend host and/or port specified'103 raise Exception104 if secure:105 if hasattr(settings, 'BACKEND_PORT'):106 conn = httplib.HTTPSConnection(settings.BACKEND_HOST, settings.BACKEND_PORT)107 else:108 conn = httplib.HTTPSConnection(settings.BACKEND_HOST)109 else:110 if hasattr(settings, 'BACKEND_PORT'):111 conn = httplib.HTTPConnection(settings.BACKEND_HOST, settings.BACKEND_PORT)112 else:113 conn = httplib.HTTPConnection(settings.BACKEND_HOST)114 headers = proxy_server.RESTFUL_HEADER115 headers[proxy_server.USER_TOKEN] = request.META.get(proxy_server.HTTP_USER_TOKEN)116 headers[proxy_server.CLIENT_IP] = request.META.get(proxy_server.HTTP_FROM)117 headers[proxy_server.API_KEY] = request.META.get(proxy_server.HTTP_API_KEY)118 try:119 conn.request(method, function_path, json.dumps(json_data), headers)120 except:121 error_message = 'Could not connect to service'122 raise Exception123 response = conn.getresponse()124 response_data = response.read()125 conn.close()126 if response.status == 403:127 resp = HttpResponse(status=response.status, reason=response.reason)128 for header, value in response.getheaders():129 resp[header] = value130 for header in proxy_server.HOP_BY_HOP:131 del resp[header]132 resp[proxy_server.HEADER_SERVER] = proxy_server.VALUE_SERVER133 return resp134 if response.status == 204:135 if response_token is True:136 error_message = 'Backend server didn\'t respond with a token'137 raise Exception138 resp = HttpResponse(status=response.status, content_type='application/json', reason=response.reason)139 for header, value in response.getheaders():140 resp[header] = value141 for header in proxy_server.HOP_BY_HOP:142 del resp[header]143 resp[proxy_server.HEADER_SERVER] = proxy_server.VALUE_SERVER144 return resp145 else:146 try:147 response_json = json.loads(response_data)148 except:149 error_message = 'Unknown response format'150 raise Exception151 if response.status == 200:152 if response_token is True and proxy_server.USER_TOKEN not in response_json:153 error_message = 'Server expected user token in response'154 raise Exception155 resp = HttpResponse(response_data, status=response.status, content_type='application/json', reason=response.reason)156 for header, value in response.getheaders():157 resp[header] = value158 for header in proxy_server.HOP_BY_HOP:159 del resp[header]160 resp[proxy_server.HEADER_SERVER] = proxy_server.VALUE_SERVER161 return resp162 except Exception as e:163 if error_message is None:164 error_message = 'Unknown error in service invocation'165 code = int(str(e)) if e is not None and isinstance(str(e), int) else 500166 error = {167 'error': {168 'code': code,169 'type': 'ProxyServerError',170 'message': error_message171 }172 }...
test_proxy.py
Source:test_proxy.py
1"""2Test ability to proxy requests.3"""4import os5import ssl6import unittest7from geopy.compat import urlopen8from geopy.exc import GeocoderServiceError9from geopy.geocoders.base import Geocoder10from test.proxy_server import ProxyServerThread11try:12 from urllib.request import getproxies13except ImportError: # py214 from urllib2 import getproxies15CERT_SELFSIGNED_CA = os.path.join(os.path.dirname(__file__), 'selfsigned_ca.pem')16# Are system proxies set? System proxies are set in:17# - Environment variables (HTTP_PROXY/HTTPS_PROXY) on Unix;18# - System Configuration Framework on macOS;19# - Registry's Internet Settings section on Windows.20WITH_SYSTEM_PROXIES = bool(getproxies())21class DummyGeocoder(Geocoder):22 def geocode(self, location):23 geo_request = self._call_geocoder(location, raw=True)24 geo_html = geo_request.read()25 return geo_html if geo_html else None26class ProxyTestCase(unittest.TestCase):27 remote_website_http = "http://example.org/"28 remote_website_https = "https://example.org/"29 timeout = 530 def setUp(self):31 self.proxy_server = ProxyServerThread(timeout=self.timeout)32 self.proxy_server.start()33 self.proxy_url = self.proxy_server.get_proxy_url()34 def tearDown(self):35 self.proxy_server.stop()36 self.proxy_server.join()37 def test_geocoder_constructor_uses_http_proxy(self):38 base_http = urlopen(self.remote_website_http, timeout=self.timeout)39 base_html = base_http.read()40 geocoder_dummy = DummyGeocoder(proxies={"http": self.proxy_url},41 timeout=self.timeout)42 self.assertEqual(0, len(self.proxy_server.requests))43 self.assertEqual(44 base_html,45 geocoder_dummy.geocode(self.remote_website_http)46 )47 self.assertEqual(1, len(self.proxy_server.requests))48 def test_geocoder_constructor_uses_https_proxy(self):49 base_http = urlopen(self.remote_website_https, timeout=self.timeout)50 base_html = base_http.read()51 geocoder_dummy = DummyGeocoder(proxies={"https": self.proxy_url},52 timeout=self.timeout)53 self.assertEqual(0, len(self.proxy_server.requests))54 self.assertEqual(55 base_html,56 geocoder_dummy.geocode(self.remote_website_https)57 )58 self.assertEqual(1, len(self.proxy_server.requests))59 @unittest.skipUnless(60 hasattr(ssl, 'create_default_context'),61 "The current Python version doesn't support `ssl.create_default_context`."62 )63 def test_ssl_context_with_proxy_is_respected(self):64 # Create an ssl context which should not allow the negotiation with65 # the `self.remote_website_https`.66 bad_ctx = ssl.create_default_context(cafile=CERT_SELFSIGNED_CA)67 geocoder_dummy = DummyGeocoder(proxies={"https": self.proxy_url},68 ssl_context=bad_ctx,69 timeout=self.timeout)70 self.assertEqual(0, len(self.proxy_server.requests))71 with self.assertRaises(GeocoderServiceError) as cm:72 geocoder_dummy.geocode(self.remote_website_https)73 self.assertIn('SSL', str(cm.exception))74 self.assertEqual(1, len(self.proxy_server.requests))75 def test_geocoder_constructor_uses_str_proxy(self):76 base_http = urlopen(self.remote_website_http, timeout=self.timeout)77 base_html = base_http.read()78 geocoder_dummy = DummyGeocoder(proxies=self.proxy_url,79 timeout=self.timeout)80 self.assertEqual(0, len(self.proxy_server.requests))81 self.assertEqual(82 base_html,83 geocoder_dummy.geocode(self.remote_website_http)84 )85 self.assertEqual(1, len(self.proxy_server.requests))86 def test_geocoder_constructor_has_both_schemes_proxy(self):87 g = DummyGeocoder(proxies=self.proxy_url, scheme='http')88 self.assertDictEqual(g.proxies, {'http': self.proxy_url,89 'https': self.proxy_url})90@unittest.skipUnless(not WITH_SYSTEM_PROXIES,91 "There're active system proxies")92class SystemProxiesTestCase(unittest.TestCase):93 remote_website_http = "http://example.org/"94 timeout = 595 def setUp(self):96 self.proxy_server = ProxyServerThread(timeout=self.timeout)97 self.proxy_server.start()98 self.proxy_url = self.proxy_server.get_proxy_url()99 self.assertIsNone(os.environ.get('http_proxy'))100 self.assertIsNone(os.environ.get('https_proxy'))101 os.environ['http_proxy'] = self.proxy_url102 os.environ['https_proxy'] = self.proxy_url103 def tearDown(self):104 self.proxy_server.stop()105 self.proxy_server.join()106 os.environ.pop('http_proxy', None)107 os.environ.pop('https_proxy', None)108 def test_system_proxies_are_respected_by_default(self):109 geocoder_dummy = DummyGeocoder(timeout=self.timeout)110 self.assertEqual(0, len(self.proxy_server.requests))111 geocoder_dummy.geocode(self.remote_website_http)112 self.assertEqual(1, len(self.proxy_server.requests))113 def test_system_proxies_are_respected_with_none(self):114 # proxies=None means "use system proxies", e.g. from the ENV.115 geocoder_dummy = DummyGeocoder(proxies=None, timeout=self.timeout)116 self.assertEqual(0, len(self.proxy_server.requests))117 geocoder_dummy.geocode(self.remote_website_http)118 self.assertEqual(1, len(self.proxy_server.requests))119 def test_system_proxies_are_reset_with_empty_dict(self):120 geocoder_dummy = DummyGeocoder(proxies={}, timeout=self.timeout)121 self.assertEqual(0, len(self.proxy_server.requests))122 geocoder_dummy.geocode(self.remote_website_http)123 self.assertEqual(0, len(self.proxy_server.requests))124 def test_string_value_overrides_system_proxies(self):125 os.environ['http_proxy'] = '127.0.0.1:1'126 os.environ['https_proxy'] = '127.0.0.1:1'127 geocoder_dummy = DummyGeocoder(proxies=self.proxy_url,128 timeout=self.timeout)129 self.assertEqual(0, len(self.proxy_server.requests))130 geocoder_dummy.geocode(self.remote_website_http)...
route.py
Source:route.py
1import typing as t2from dimensigon.network.low_level import check_host3from dimensigon.utils.typos import UUID4from dimensigon.web import db, errors5if t.TYPE_CHECKING:6 from dimensigon.domain.entities import Server, Gate7class RouteContainer:8 def __init__(self, proxy_server: t.Optional['Server'], gate: t.Optional['Gate'], cost: t.Optional[int]):9 self.proxy_server = proxy_server10 self.gate = gate11 self.cost = cost12 def __str__(self):13 return f"proxy_server={getattr(self.proxy_server, 'name', None)}, gate={self.gate}, cost={self.cost}"14 def __repr__(self):15 return f"RouteContainer(proxy_server={getattr(self.proxy_server, 'id', None)}, " \16 f"gate={getattr(self.gate, 'id', None)}, cost={self.cost}"17 def __iter__(self):18 yield self.proxy_server19 yield self.gate20 yield self.cost21 def __getitem__(self, item):22 if item == 0:23 return self.proxy_server24 elif item == 1:25 return self.gate26 elif item == 2:27 return self.cost28 else:29 IndexError('list index out of range')30 def __eq__(self, other):31 return isinstance(other, self.__class__) and self.cost == other.cost and self.gate == other.gate \32 and self.proxy_server == other.proxy_server33class Route(db.Model):34 __tablename__ = 'L_route'35 destination_id = db.Column(UUID, db.ForeignKey('D_server.id'), primary_key=True, nullable=False)36 proxy_server_id = db.Column(UUID, db.ForeignKey('D_server.id'))37 gate_id = db.Column(UUID, db.ForeignKey('D_gate.id'))38 cost = db.Column(db.Integer)39 destination = db.relationship("Server", foreign_keys=[destination_id], back_populates="route", lazy='joined')40 proxy_server = db.relationship("Server", foreign_keys=[proxy_server_id], lazy='joined')41 gate = db.relationship("Gate", foreign_keys=[gate_id], lazy='joined')42 def __init__(self, destination: 'Server', proxy_server_or_gate: t.Union['Server', 'Gate'] = None, cost: int = None):43 # avoid cycle import44 from dimensigon.domain.entities import Server45 self.destination = destination46 if isinstance(proxy_server_or_gate, Server):47 proxy_server = proxy_server_or_gate48 gate = None49 else:50 proxy_server = None51 gate = proxy_server_or_gate52 if proxy_server:53 if proxy_server == destination:54 raise ValueError('You must specify a gate when proxy_server equals destination')55 else:56 if cost is None or cost == 0:57 raise ValueError("Cost must be specified and greater than 0 when proxy_server")58 self.proxy_server = proxy_server59 self.cost = cost60 elif gate:61 # check if gate is from neighbour or from a proxy server62 if destination == gate.server:63 if cost is not None and cost > 0:64 raise ValueError("Cost must be set to 0 when defining route for a neighbour")65 self.gate = gate66 self.cost = 067 else:68 if cost is None or cost <= 0:69 raise ValueError("Cost must be specified and greater than 0 when gate is from a proxy_server")70 else:71 self.proxy_server = gate.server72 self.cost = cost73 elif cost == 0:74 # find a gateway and set that gateway as default75 if len(destination.external_gates) == 1:76 self.gate = destination.external_gates[0]77 self.cost = 078 else:79 for gate in destination.external_gates:80 if check_host(gate.dns or str(gate.ip), gate.port, timeout=1, retry=3, delay=0.5):81 self.gate = gate82 self.cost = 083 break84 # if not (self.gate or self.proxy_server):85 # raise ValueError('Not a valid route')86 def validate_route(self, rc: RouteContainer):87 if rc.proxy_server:88 if not (rc.gate is None and rc.cost > 0):89 raise errors.InvalidRoute(self.destination, rc)90 if rc.proxy_server._me:91 raise errors.InvalidRoute(self.destination, rc)92 elif rc.gate:93 if not rc.cost == 0:94 raise errors.InvalidRoute(self.destination, rc)95 else:96 if rc.cost is not None:97 raise errors.InvalidRoute(self.destination, rc)98 def set_route(self, rc: RouteContainer):99 self.validate_route(rc)100 self.proxy_server, self.gate, self.cost = rc101 def __str__(self):102 return f"{self.destination} -> " \103 f"{self.proxy_server or self.gate}, {self.cost}"104 def __repr__(self):105 return f"Route({self.to_json()})"106 def to_json(self, human=False):107 if not self.destination.id or (self.proxy_server and not self.proxy_server.id) or (108 self.gate and not self.gate.id):109 raise RuntimeError("commit object before dump to json")110 if human:111 return {'destination': str(self.destination) if self.destination else None,112 'proxy_server': str(self.proxy_server) if self.proxy_server else None,113 'gate': str(self.gate) if self.gate else None,114 'cost': self.cost}115 else:116 return {'destination_id': self.destination_id,117 'proxy_server_id': self.proxy_server_id,118 'gate_id': self.gate_id,...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
