from datetime import datetime as dt
from datetime import timezone
import pytest
from conftest import DB, Response
from hircine.db.models import World
@pytest.fixture
def query_world(execute_id):
query = """
query world($id: Int!) {
world(id: $id) {
__typename
... on World {
id
name
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_id(query)
@pytest.fixture
def query_worlds(execute):
query = """
query worlds {
worlds {
__typename
count
edges {
id
name
}
}
}
"""
return execute(query)
@pytest.fixture
def add_world(execute_add):
mutation = """
mutation addWorld($input: AddWorldInput!) {
addWorld(input: $input) {
__typename
... on AddSuccess {
id
}
... on Error {
message
}
... on InvalidParameterError {
parameter
}
}
}
"""
return execute_add(mutation)
@pytest.fixture
def update_worlds(execute_update):
mutation = """
mutation updateWorlds($ids: [Int!]!, $input: UpdateWorldInput!) {
updateWorlds(ids: $ids, input: $input) {
__typename
... on Success {
message
}
... on Error {
message
}
... on IDNotFoundError {
id
}
... on InvalidParameterError {
parameter
}
}
}
""" # noqa: E501
return execute_update(mutation)
@pytest.fixture
def delete_worlds(execute_delete):
mutation = """
mutation deleteWorlds($ids: [Int!]!) {
deleteWorlds(ids: $ids) {
__typename
... on Success {
message
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_delete(mutation)
@pytest.mark.anyio
async def test_query_world(query_world, gen_world):
world = await DB.add(next(gen_world))
response = Response(await query_world(world.id))
response.assert_is("World")
assert response.id == world.id
assert response.name == world.name
@pytest.mark.anyio
async def test_query_world_fails_not_found(query_world):
response = Response(await query_world(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "World ID not found: '1'"
@pytest.mark.anyio
async def test_query_worlds(query_worlds, gen_world):
worlds = await DB.add_all(*gen_world)
response = Response(await query_worlds())
response.assert_is("WorldFilterResult")
assert response.count == len(worlds)
assert isinstance((response.edges), list)
assert len(response.edges) == len(worlds)
edges = iter(response.edges)
for world in sorted(worlds, key=lambda a: a.name):
edge = next(edges)
assert edge["id"] == world.id
assert edge["name"] == world.name
@pytest.mark.anyio
async def test_add_world(add_world):
response = Response(await add_world({"name": "added world"}))
response.assert_is("AddSuccess")
world = await DB.get(World, response.id)
assert world is not None
assert world.name == "added world"
@pytest.mark.anyio
async def test_add_world_fails_empty_parameter(add_world):
response = Response(await add_world({"name": ""}))
response.assert_is("InvalidParameterError")
assert response.parameter == "name"
assert response.message == "Invalid parameter 'name': cannot be empty"
@pytest.mark.anyio
async def test_add_world_fails_exists(add_world, gen_world):
world = await DB.add(next(gen_world))
response = Response(await add_world({"name": world.name}))
response.assert_is("NameExistsError")
assert response.message == "Another World with this name exists"
@pytest.mark.anyio
async def test_delete_world(delete_worlds, gen_world):
world = await DB.add(next(gen_world))
id = world.id
response = Response(await delete_worlds(id))
response.assert_is("DeleteSuccess")
world = await DB.get(World, id)
assert world is None
@pytest.mark.anyio
async def test_delete_world_not_found(delete_worlds):
response = Response(await delete_worlds(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "World ID not found: '1'"
@pytest.mark.anyio
async def test_update_world(update_worlds, gen_world):
world = await DB.add(next(gen_world))
input = {"name": "updated world"}
response = Response(await update_worlds(world.id, input))
response.assert_is("UpdateSuccess")
world = await DB.get(World, world.id)
assert world is not None
assert world.name == "updated world"
@pytest.mark.anyio
async def test_update_world_fails_exists(update_worlds, gen_world):
first = await DB.add(next(gen_world))
second = await DB.add(next(gen_world))
response = Response(await update_worlds(second.id, {"name": first.name}))
response.assert_is("NameExistsError")
assert response.message == "Another World with this name exists"
@pytest.mark.anyio
async def test_update_world_fails_not_found(update_worlds):
response = Response(await update_worlds(1, {"name": "updated world"}))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "World ID not found: '1'"
@pytest.mark.anyio
async def test_update_worlds_cannot_bulk_edit_name(update_worlds, gen_world):
first = await DB.add(next(gen_world))
second = await DB.add(next(gen_world))
response = Response(await update_worlds([first.id, second.id], {"name": "unique"}))
response.assert_is("InvalidParameterError")
@pytest.mark.parametrize(
"empty",
[
None,
"",
],
ids=[
"none",
"empty string",
],
)
@pytest.mark.anyio
async def test_update_world_fails_empty_parameter(update_worlds, gen_world, empty):
world = await DB.add(next(gen_world))
response = Response(await update_worlds(world.id, {"name": empty}))
response.assert_is("InvalidParameterError")
assert response.parameter == "name"
assert response.message == "Invalid parameter 'name': cannot be empty"
@pytest.mark.anyio
async def test_update_world_changes_updated_at(update_worlds):
original_world = World(name="world")
original_world.updated_at = dt(2023, 1, 1, tzinfo=timezone.utc)
original_world = await DB.add(original_world)
response = Response(await update_worlds(original_world.id, {"name": "updated"}))
response.assert_is("UpdateSuccess")
world = await DB.get(World, original_world.id)
assert world.updated_at > original_world.updated_at
