from datetime import datetime as dt
from datetime import timezone
import pytest
from conftest import DB, Response
from hircine.db.models import Circle
@pytest.fixture
def query_circle(execute_id):
query = """
query circle($id: Int!) {
circle(id: $id) {
__typename
... on Circle {
id
name
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_id(query)
@pytest.fixture
def query_circles(execute):
query = """
query circles {
circles {
__typename
count
edges {
id
name
}
}
}
"""
return execute(query)
@pytest.fixture
def add_circle(execute_add):
mutation = """
mutation addCircle($input: AddCircleInput!) {
addCircle(input: $input) {
__typename
... on AddSuccess {
id
}
... on Error {
message
}
... on InvalidParameterError {
parameter
}
}
}
"""
return execute_add(mutation)
@pytest.fixture
def update_circles(execute_update):
mutation = """
mutation updateCircles($ids: [Int!]!, $input: UpdateCircleInput!) {
updateCircles(ids: $ids, input: $input) {
__typename
... on Success {
message
}
... on IDNotFoundError {
id
}
... on Error {
message
}
... on InvalidParameterError {
parameter
}
}
}
""" # noqa: E501
return execute_update(mutation)
@pytest.fixture
def delete_circles(execute_delete):
mutation = """
mutation deleteCircles($ids: [Int!]!) {
deleteCircles(ids: $ids) {
__typename
... on Success {
message
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_delete(mutation)
@pytest.mark.anyio
async def test_query_circle(query_circle, gen_circle):
circle = await DB.add(next(gen_circle))
response = Response(await query_circle(circle.id))
response.assert_is("Circle")
assert response.id == circle.id
assert response.name == circle.name
@pytest.mark.anyio
async def test_query_circle_fails_not_found(query_circle):
response = Response(await query_circle(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Circle ID not found: '1'"
@pytest.mark.anyio
async def test_query_circles(query_circles, gen_circle):
circles = await DB.add_all(*gen_circle)
response = Response(await query_circles())
response.assert_is("CircleFilterResult")
assert response.count == len(circles)
assert isinstance((response.edges), list)
assert len(response.edges) == len(circles)
edges = iter(response.edges)
for circle in sorted(circles, key=lambda a: a.name):
edge = next(edges)
assert edge["id"] == circle.id
assert edge["name"] == circle.name
@pytest.mark.anyio
async def test_add_circle(add_circle):
response = Response(await add_circle({"name": "added circle"}))
response.assert_is("AddSuccess")
circle = await DB.get(Circle, response.id)
assert circle is not None
assert circle.name == "added circle"
@pytest.mark.anyio
async def test_add_circle_fails_empty_parameter(add_circle):
response = Response(await add_circle({"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_circle_fails_exists(add_circle, gen_circle):
circle = await DB.add(next(gen_circle))
response = Response(await add_circle({"name": circle.name}))
response.assert_is("NameExistsError")
assert response.message == "Another Circle with this name exists"
@pytest.mark.anyio
async def test_delete_circle(delete_circles, gen_circle):
circle = await DB.add(next(gen_circle))
id = circle.id
response = Response(await delete_circles(id))
response.assert_is("DeleteSuccess")
circle = await DB.get(Circle, id)
assert circle is None
@pytest.mark.anyio
async def test_delete_circle_not_found(delete_circles):
response = Response(await delete_circles(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Circle ID not found: '1'"
@pytest.mark.anyio
async def test_update_circle(update_circles, gen_circle):
circle = await DB.add(next(gen_circle))
input = {"name": "updated circle"}
response = Response(await update_circles(circle.id, input))
response.assert_is("UpdateSuccess")
circle = await DB.get(Circle, circle.id)
assert circle is not None
assert circle.name == "updated circle"
@pytest.mark.anyio
async def test_update_circle_fails_exists(update_circles, gen_circle):
first = await DB.add(next(gen_circle))
second = await DB.add(next(gen_circle))
response = Response(await update_circles(second.id, {"name": first.name}))
response.assert_is("NameExistsError")
assert response.message == "Another Circle with this name exists"
@pytest.mark.anyio
async def test_update_circle_fails_not_found(update_circles):
response = Response(await update_circles(1, {"name": "updated circle"}))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Circle ID not found: '1'"
@pytest.mark.anyio
async def test_update_circles_cannot_bulk_edit_name(update_circles, gen_circle):
first = await DB.add(next(gen_circle))
second = await DB.add(next(gen_circle))
response = Response(await update_circles([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_circle_fails_empty_parameter(update_circles, gen_circle, empty):
circle = await DB.add(next(gen_circle))
response = Response(await update_circles(circle.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_circle_changes_updated_at(update_circles):
original_circle = Circle(name="circle")
original_circle.updated_at = dt(2023, 1, 1, tzinfo=timezone.utc)
original_circle = await DB.add(original_circle)
response = Response(await update_circles(original_circle.id, {"name": "updated"}))
response.assert_is("UpdateSuccess")
circle = await DB.get(Circle, original_circle.id)
assert circle.updated_at > original_circle.updated_at
