from datetime import datetime as dt
from datetime import timezone
import pytest
from conftest import DB, Response
from hircine.db.models import Namespace
@pytest.fixture
def query_namespace(execute_id):
query = """
query namespace($id: Int!) {
namespace(id: $id) {
__typename
... on Namespace {
id
name
sortName
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_id(query)
@pytest.fixture
def query_namespaces(execute):
query = """
query namespaces {
namespaces {
__typename
count
edges {
id
name
}
}
}
"""
return execute(query)
@pytest.fixture
def add_namespace(execute_add):
mutation = """
mutation addNamespace($input: AddNamespaceInput!) {
addNamespace(input: $input) {
__typename
... on AddSuccess {
id
}
... on Error {
message
}
... on InvalidParameterError {
parameter
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_add(mutation)
@pytest.fixture
def update_namespaces(execute_update):
mutation = """
mutation updateNamespaces($ids: [Int!]!, $input: UpdateNamespaceInput!) {
updateNamespaces(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_namespaces(execute_delete):
mutation = """
mutation deleteNamespaces($ids: [Int!]!) {
deleteNamespaces(ids: $ids) {
__typename
... on Success {
message
}
... on Error {
message
}
... on IDNotFoundError {
id
}
}
}
"""
return execute_delete(mutation)
@pytest.mark.anyio
async def test_query_namespace(query_namespace, gen_namespace):
namespace = await DB.add(next(gen_namespace))
response = Response(await query_namespace(namespace.id))
response.assert_is("Namespace")
assert response.id == namespace.id
assert response.name == namespace.name
assert response.sortName == namespace.sort_name
@pytest.mark.anyio
async def test_query_namespace_fails_not_found(query_namespace):
response = Response(await query_namespace(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Namespace ID not found: '1'"
@pytest.mark.anyio
async def test_query_namespaces(query_namespaces, gen_namespace):
namespaces = await DB.add_all(*gen_namespace)
response = Response(await query_namespaces())
response.assert_is("NamespaceFilterResult")
assert response.count == len(namespaces)
assert isinstance((response.edges), list)
assert len(response.edges) == len(namespaces)
edges = iter(response.edges)
for namespace in sorted(namespaces, key=lambda a: a.name):
edge = next(edges)
assert edge["id"] == namespace.id
assert edge["name"] == namespace.name
@pytest.mark.anyio
async def test_add_namespace(add_namespace):
response = Response(await add_namespace({"name": "added", "sortName": "foo"}))
response.assert_is("AddSuccess")
namespace = await DB.get(Namespace, response.id)
assert namespace is not None
assert namespace.name == "added"
assert namespace.sort_name == "foo"
@pytest.mark.anyio
async def test_add_namespace_fails_empty_parameter(add_namespace):
response = Response(await add_namespace({"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_namespace_fails_exists(add_namespace, gen_namespace):
namespace = await DB.add(next(gen_namespace))
response = Response(await add_namespace({"name": namespace.name}))
response.assert_is("NameExistsError")
assert response.message == "Another Namespace with this name exists"
@pytest.mark.anyio
async def test_delete_namespace(delete_namespaces, gen_namespace):
namespace = await DB.add(next(gen_namespace))
id = namespace.id
response = Response(await delete_namespaces(id))
response.assert_is("DeleteSuccess")
namespace = await DB.get(Namespace, id)
assert namespace is None
@pytest.mark.anyio
async def test_delete_namespace_not_found(delete_namespaces):
response = Response(await delete_namespaces(1))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Namespace ID not found: '1'"
@pytest.mark.anyio
async def test_update_namespace(update_namespaces, gen_namespace):
namespace = await DB.add(next(gen_namespace))
input = {"name": "updated", "sortName": "foo"}
response = Response(await update_namespaces(namespace.id, input))
response.assert_is("UpdateSuccess")
namespace = await DB.get(Namespace, namespace.id)
assert namespace is not None
assert namespace.name == "updated"
assert namespace.sort_name == "foo"
@pytest.mark.anyio
async def test_update_namespace_fails_exists(update_namespaces, gen_namespace):
first = await DB.add(next(gen_namespace))
second = await DB.add(next(gen_namespace))
response = Response(await update_namespaces(second.id, {"name": first.name}))
response.assert_is("NameExistsError")
assert response.message == "Another Namespace with this name exists"
@pytest.mark.anyio
async def test_update_namespace_fails_not_found(update_namespaces):
response = Response(await update_namespaces(1, {"name": "updated"}))
response.assert_is("IDNotFoundError")
assert response.id == 1
assert response.message == "Namespace ID not found: '1'"
@pytest.mark.anyio
async def test_update_namespaces_cannot_bulk_edit_name(
update_namespaces, gen_namespace
):
first = await DB.add(next(gen_namespace))
second = await DB.add(next(gen_namespace))
response = Response(
await update_namespaces([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_namespace_fails_empty_parameter(
update_namespaces, gen_namespace, empty
):
namespace = await DB.add(next(gen_namespace))
response = Response(await update_namespaces(namespace.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_namespace_changes_updated_at(update_namespaces):
original_namespace = Namespace(name="namespace")
original_namespace.updated_at = dt(2023, 1, 1, tzinfo=timezone.utc)
original_namespace = await DB.add(original_namespace)
response = Response(
await update_namespaces(original_namespace.id, {"name": "updated"})
)
response.assert_is("UpdateSuccess")
namespace = await DB.get(Namespace, original_namespace.id)
assert namespace.updated_at > original_namespace.updated_at
