Source code
Revision control
Copy as Markdown
Other Tools
#!/usr/bin/env python3
import asyncio
import pathlib
import socket
import ssl
import aiohttp
from aiohttp import web
from aiohttp.abc import AbstractResolver
from aiohttp.resolver import DefaultResolver
from aiohttp.test_utils import unused_port
class FakeResolver(AbstractResolver):
_LOCAL_HOST = {0: "127.0.0.1", socket.AF_INET: "127.0.0.1", socket.AF_INET6: "::1"}
def __init__(self, fakes, *, loop):
"""fakes -- dns -> port dict"""
self._fakes = fakes
self._resolver = DefaultResolver(loop=loop)
async def resolve(self, host, port=0, family=socket.AF_INET):
fake_port = self._fakes.get(host)
if fake_port is not None:
return [
{
"hostname": host,
"host": self._LOCAL_HOST[family],
"port": fake_port,
"family": family,
"proto": 0,
"flags": socket.AI_NUMERICHOST,
}
]
else:
return await self._resolver.resolve(host, port, family)
async def close(self) -> None:
self._resolver.close()
class FakeFacebook:
def __init__(self, *, loop):
self.loop = loop
self.app = web.Application(loop=loop)
self.app.router.add_routes(
[
web.get("/v2.7/me", self.on_me),
web.get("/v2.7/me/friends", self.on_my_friends),
]
)
self.runner = web.AppRunner(self.app)
here = pathlib.Path(__file__)
ssl_cert = here.parent / "server.crt"
ssl_key = here.parent / "server.key"
self.ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
self.ssl_context.load_cert_chain(str(ssl_cert), str(ssl_key))
async def start(self):
port = unused_port()
self.runner = web.AppRunner(self.app)
await self.runner.setup()
site = web.TCPSite(self.runner, "127.0.0.1", port, ssl_context=self.ssl_context)
await site.start()
return {"graph.facebook.com": port}
async def stop(self):
await self.runner.cleanup()
async def on_me(self, request):
return web.json_response({"name": "John Doe", "id": "12345678901234567"})
async def on_my_friends(self, request):
return web.json_response(
{
"data": [
{"name": "Bill Doe", "id": "233242342342"},
{"name": "Mary Doe", "id": "2342342343222"},
{"name": "Alex Smith", "id": "234234234344"},
],
"paging": {
"cursors": {
"before": "QVFIUjRtc2c5NEl0ajN",
"after": "QVFIUlpFQWM0TmVuaDRad0dt",
},
"next": (
"friends?access_token=EAACEdEose0cB"
),
},
"summary": {"total_count": 3},
}
)
async def main(loop):
token = "ER34gsSGGS34XCBKd7u"
fake_facebook = FakeFacebook(loop=loop)
info = await fake_facebook.start()
resolver = FakeResolver(info, loop=loop)
connector = aiohttp.TCPConnector(loop=loop, resolver=resolver, verify_ssl=False)
async with aiohttp.ClientSession(connector=connector, loop=loop) as session:
async with session.get(
) as resp:
print(await resp.json())
async with session.get(
) as resp:
print(await resp.json())
await fake_facebook.stop()
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))