Spaces:
Running
Running
| """A local stand-in for api.telegra.ph (createAccount and createPage), used by the Telegraph and /mediainfo tests.""" | |
| import asyncio | |
| from aiohttp import web | |
| from aiohttp.test_utils import TestServer | |
| class FakeTelegraph: | |
| """api.telegra.ph for tests: createAccount and createPage, with switches for the ways it can go wrong.""" | |
| def __init__(self): | |
| self.accounts = 0 | |
| self.valid = set() | |
| self.attempts = [] # every createPage form, accepted or not | |
| self.pages = [] # the accepted ones | |
| self.error = None # answer every createPage with this error | |
| self.reject_all = False # a token is never good enough | |
| self.delay = 0.0 | |
| self.raw = None # (status, body) to answer createPage with instead of JSON | |
| async def create_account(self, request): | |
| form = await request.post() | |
| self.accounts += 1 | |
| token = f"token{self.accounts}" | |
| self.valid.add(token) | |
| return web.json_response({"ok": True, "result": {"short_name": form["short_name"], "access_token": token}}) | |
| async def create_page(self, request): | |
| form = dict(await request.post()) | |
| self.attempts.append(form) | |
| if self.delay: | |
| await asyncio.sleep(self.delay) | |
| if self.raw is not None: | |
| return web.Response(status=self.raw[0], text=self.raw[1], content_type="text/html") | |
| if self.error: | |
| return web.json_response({"ok": False, "error": self.error}) | |
| if self.reject_all or form.get("access_token") not in self.valid: | |
| return web.json_response({"ok": False, "error": "ACCESS_TOKEN_INVALID"}) | |
| self.pages.append(form) | |
| n = len(self.pages) | |
| return web.json_response({"ok": True, "result": {"path": f"Page-{n}", "url": f"https://telegra.ph/Page-{n}", "title": form["title"]}}) | |
| async def serve_fake_telegraph() -> "FakeTelegraph": | |
| """Start the fake server; ``fake.url`` is the API address and ``await fake.server.close()`` stops it.""" | |
| fake = FakeTelegraph() | |
| app = web.Application() | |
| app.router.add_post("/createAccount", fake.create_account) | |
| app.router.add_post("/createPage", fake.create_page) | |
| fake.server = TestServer(app) | |
| await fake.server.start_server() | |
| fake.url = str(fake.server.make_url("")).rstrip("/") | |
| return fake | |