mirror of https://gitlab.com/litecord/litecord.git
tests: add support for test guild resources
This commit is contained in:
parent
d01fde15da
commit
494a6b5c63
109
tests/common.py
109
tests/common.py
|
|
@ -22,7 +22,9 @@ from typing import Optional
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
from litecord.common.users import create_user, delete_user
|
from litecord.common.users import create_user, delete_user
|
||||||
|
from litecord.common.guilds import delete_guild
|
||||||
from litecord.blueprints.auth import make_token
|
from litecord.blueprints.auth import make_token
|
||||||
|
from litecord.storage import int_
|
||||||
|
|
||||||
|
|
||||||
def email() -> str:
|
def email() -> str:
|
||||||
|
|
@ -38,15 +40,94 @@ class WrappedUser:
|
||||||
password: str
|
password: str
|
||||||
token: str
|
token: str
|
||||||
|
|
||||||
async def refetch(self):
|
async def refetch(self) -> dict:
|
||||||
async with self.test_cli.app.app_context():
|
async with self.test_cli.app.app_context():
|
||||||
return await self.test_cli.app.storage.get_user(self.id)
|
return await self.test_cli.app.storage.get_user(self.id)
|
||||||
|
|
||||||
async def delete(self):
|
async def delete(self):
|
||||||
async with self.test_cli.app.app_context():
|
|
||||||
return await delete_user(self.id)
|
return await delete_user(self.id)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class WrappedGuild:
|
||||||
|
test_cli: "TestClient"
|
||||||
|
id: int
|
||||||
|
owner: bool # value depends on the user that fetched guild
|
||||||
|
owner_id: int
|
||||||
|
name: str
|
||||||
|
unavailable: bool
|
||||||
|
icon: Optional[str]
|
||||||
|
splash: Optional[str]
|
||||||
|
region: Optional[str]
|
||||||
|
afk_timeout: int
|
||||||
|
afk_channel_id: Optional[str]
|
||||||
|
afk_timeout: int
|
||||||
|
verification_level: int
|
||||||
|
default_message_notifications: int
|
||||||
|
explicit_content_filter: int
|
||||||
|
mfa_level: int
|
||||||
|
embed_enabled: bool
|
||||||
|
embed_channel_id: int
|
||||||
|
widget_enabled: bool
|
||||||
|
widget_channel_id: int
|
||||||
|
system_channel_id: int
|
||||||
|
rules_channel_id: int
|
||||||
|
public_updates_channel_id: int
|
||||||
|
features: str
|
||||||
|
features: str
|
||||||
|
banner: Optional[str]
|
||||||
|
description: Optional[str]
|
||||||
|
preferred_locale: Optional[str]
|
||||||
|
discovery_splash: Optional[str]
|
||||||
|
|
||||||
|
vanity_url_code: Optional[str]
|
||||||
|
max_presences: int
|
||||||
|
max_members: int
|
||||||
|
guild_scheduled_events: list
|
||||||
|
|
||||||
|
joined_at: str # value depends on the user that fetched the guild
|
||||||
|
|
||||||
|
member_count: int
|
||||||
|
members: list
|
||||||
|
channels: list
|
||||||
|
roles: list
|
||||||
|
presences: list
|
||||||
|
emojis: list
|
||||||
|
voice_states: list
|
||||||
|
|
||||||
|
large: Optional[bool] = None
|
||||||
|
|
||||||
|
async def delete(self):
|
||||||
|
await delete_guild(self.id)
|
||||||
|
|
||||||
|
async def refetch(self) -> "WrappedGuild":
|
||||||
|
async with self.test_cli.app.app_context():
|
||||||
|
guild = await self.test_cli.app.storage.get_guild_full(self.id)
|
||||||
|
return WrappedGuild.from_json(self.test_cli, guild)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_json(cls, test_cli, rjson):
|
||||||
|
return cls(
|
||||||
|
test_cli,
|
||||||
|
**{
|
||||||
|
**rjson,
|
||||||
|
**{
|
||||||
|
"id": int(rjson["id"]),
|
||||||
|
"owner_id": int(rjson["owner_id"]),
|
||||||
|
"afk_channel_id": int_(rjson["afk_channel_id"]),
|
||||||
|
"embed_channel_id": int_(rjson["embed_channel_id"]),
|
||||||
|
"widget_enabled": int_(rjson["widget_enabled"]),
|
||||||
|
"widget_channel_id": int_(rjson["widget_channel_id"]),
|
||||||
|
"system_channel_id": int_(rjson["system_channel_id"]),
|
||||||
|
"rules_channel_id": int_(rjson["rules_channel_id"]),
|
||||||
|
"public_updates_channel_id": int_(
|
||||||
|
rjson["public_updates_channel_id"]
|
||||||
|
),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class TestClient:
|
class TestClient:
|
||||||
"""Test client wrapper class. Adds Authorization headers to all requests
|
"""Test client wrapper class. Adds Authorization headers to all requests
|
||||||
and manages test resource setup and destruction."""
|
and manages test resource setup and destruction."""
|
||||||
|
|
@ -66,6 +147,7 @@ class TestClient:
|
||||||
|
|
||||||
async def cleanup(self):
|
async def cleanup(self):
|
||||||
for resource in self.resources:
|
for resource in self.resources:
|
||||||
|
async with self.app.app_context():
|
||||||
await resource.delete()
|
await resource.delete()
|
||||||
|
|
||||||
async def create_user(
|
async def create_user(
|
||||||
|
|
@ -85,10 +167,33 @@ class TestClient:
|
||||||
WrappedUser(self, user_id, username, email, password, user_token)
|
WrappedUser(self, user_id, username, email, password, user_token)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def create_guild(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
name: Optional[str] = None,
|
||||||
|
region: Optional[str] = None,
|
||||||
|
owner: Optional["WrappedUser"] = None,
|
||||||
|
) -> WrappedGuild:
|
||||||
|
name = name or secrets.token_hex(6)
|
||||||
|
owner_token = owner.token if owner else self.user["token"]
|
||||||
|
|
||||||
|
async with self.app.app_context():
|
||||||
|
# TODO move guild creation logic to litecord.common.guild
|
||||||
|
# TODO make tests use aiosqlite on memory for db
|
||||||
|
resp = await self.post(
|
||||||
|
"/api/v6/guilds",
|
||||||
|
json={"name": name, "region": region},
|
||||||
|
headers={"authorization": owner_token},
|
||||||
|
)
|
||||||
|
rjson = await resp.json
|
||||||
|
|
||||||
|
return self.add_resource(WrappedGuild.from_json(self, rjson))
|
||||||
|
|
||||||
def _inject_auth(self, kwargs: dict) -> list:
|
def _inject_auth(self, kwargs: dict) -> list:
|
||||||
"""Inject the test user's API key into the test request before
|
"""Inject the test user's API key into the test request before
|
||||||
passing the request on to the underlying TestClient."""
|
passing the request on to the underlying TestClient."""
|
||||||
headers = kwargs.get("headers", {})
|
headers = kwargs.get("headers", {})
|
||||||
|
if "authorization" not in headers:
|
||||||
headers["authorization"] = self.user["token"]
|
headers["authorization"] = self.user["token"]
|
||||||
return headers
|
return headers
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue