Skip to content
Snippets Groups Projects
Verified Commit c8b5d24e authored by Daniel Göbel's avatar Daniel Göbel
Browse files

Add tests for workflow statistics endpoint

#24
parent 68b6beda
No related branches found
No related tags found
1 merge request!30Resolve "Workflow usage statistics"
Pipeline #28081 passed
from datetime import date
from io import BytesIO from io import BytesIO
from uuid import UUID, uuid4 from uuid import UUID, uuid4
import pytest import pytest
from clowmdb.models import Workflow, WorkflowVersion from clowmdb.models import Workflow, WorkflowExecution, WorkflowVersion
from fastapi import status from fastapi import status
from httpx import AsyncClient from httpx import AsyncClient
from sqlalchemy import delete, select from sqlalchemy import delete, select
...@@ -380,6 +381,34 @@ class TestWorkflowRoutesGet(_TestWorkflowRoutes): ...@@ -380,6 +381,34 @@ class TestWorkflowRoutesGet(_TestWorkflowRoutes):
) )
assert response.status_code == status.HTTP_404_NOT_FOUND assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_get_workflow_statistics(
self,
client: AsyncClient,
random_user: UserWithAuthHeader,
random_workflow: WorkflowOut,
random_workflow_execution: WorkflowExecution,
) -> None:
"""
Test for getting a non-existing workflow.
Parameters
----------
client : httpx.AsyncClient
HTTP Client to perform the request on. pytest fixture.
random_user : app.tests.utils.user.UserWithAuthHeader
Random user for testing. pytest fixture.
"""
response = await client.get(
"/".join([self.base_path, str(random_workflow.workflow_id), "statistics"]),
headers=random_user.auth_headers,
)
assert response.status_code == status.HTTP_200_OK
statistics = response.json()
assert len(statistics) == 1
assert statistics[0]["day"] == str(date.today())
assert statistics[0]["count"] == 1
class TestWorkflowRoutesDelete(_TestWorkflowRoutes): class TestWorkflowRoutesDelete(_TestWorkflowRoutes):
@pytest.mark.asyncio @pytest.mark.asyncio
......
import random import random
from datetime import date
import pytest import pytest
from clowmdb.models import Workflow, WorkflowVersion from clowmdb.models import Workflow, WorkflowExecution, WorkflowVersion
from sqlalchemy import delete, select from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
...@@ -134,6 +135,25 @@ class TestWorkflowCRUDGet: ...@@ -134,6 +135,25 @@ class TestWorkflowCRUDGet:
assert workflow is not None assert workflow is not None
assert workflow.workflow_id == random_workflow.workflow_id assert workflow.workflow_id == random_workflow.workflow_id
@pytest.mark.asyncio
async def test_get_workflow_statistics(
self, db: AsyncSession, random_workflow: WorkflowOut, random_workflow_execution: WorkflowExecution
) -> None:
"""
Test for getting a workflow by its name from CRUD Repository.
Parameters
----------
db : sqlalchemy.ext.asyncio.AsyncSession.
Async database session to perform query on. pytest fixture.
random_workflow : app.schemas.workflow.WorkflowOut
Random bucket for testing. pytest fixture.
"""
statistics = await CRUDWorkflow.statistics(db, random_workflow.workflow_id)
assert len(statistics) == 1
assert statistics[0].day == date.today()
assert statistics[0].count == 1
class TestWorkflowCRUDCreate: class TestWorkflowCRUDCreate:
@pytest.mark.asyncio @pytest.mark.asyncio
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment