Update tests

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/tests/api/test_cvat_webhook_api.py

@@ -1,33 +1,35 @@
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
+import pytest
 from fastapi.testclient import TestClient
 
 from src.core.types import AssignmentStatuses, JobStatuses
+from src.utils.time import utcnow
 
+from tests.utils.constants import WALLET_ADDRESS1, WALLET_ADDRESS2
 from tests.utils.setup_cvat import (
-    add_asignment_to_db,
+    add_assignment_to_db,
     add_cvat_job_to_db,
     add_cvat_project_to_db,
     add_cvat_task_to_db,
     generate_cvat_signature,
     get_cvat_job_from_db,
 )
 
-api_url = "http://localhost:8080/api/"
+API_URL = "http://localhost:8080/api/"
 
+PING_EVENT_DATA = {
+    "event": "ping",
+}
 
-def test_ping_incoming_webhook(client: TestClient) -> None:
-    data = {
-        "event": "ping",
-    }
-    signature = generate_cvat_signature(data)
 
+def test_ping_incoming_webhook(client: TestClient) -> None:
     # Should respond with 200 status to a "ping" event
     response = client.post(
         "/cvat-webhook",
-        headers={"X-Signature-256": signature},
-        json=data,
+        headers={"X-Signature-256": generate_cvat_signature(PING_EVENT_DATA)},
+        json=PING_EVENT_DATA,
     )
 
     assert response.status_code == 200
@@ -36,13 +38,13 @@ def test_ping_incoming_webhook(client: TestClient) -> None:
 def test_incoming_webhook_200(client: TestClient) -> None:
     # Create some entities in test DB
     add_cvat_project_to_db(cvat_id=1)
-    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1, status="annotation")
+    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1)
 
     # Payload for "create:job" event
     data = {
         "event": "create:job",
         "job": {
-            "url": api_url + "jobs/1",
+            "url": API_URL + "jobs/1",
             "id": 1,
             "task_id": 1,
             "project_id": 1,
@@ -71,107 +73,130 @@ def test_incoming_webhook_200(client: TestClient) -> None:
     assert job.cvat_project_id == 1
 
 
-def test_incoming_webhook_200_update_expired_assignmets(client: TestClient) -> None:
+@pytest.mark.parametrize("is_last_assignment", [True, False])
+def test_incoming_webhook_can_update_expired_assignment(
+    client: TestClient, is_last_assignment: bool
+):
+    # Check if an "update:job" event can update an expired assignment,
+    # if the assignment is the last one for the job. Updates to other assignments should be ignored.
+
     add_cvat_project_to_db(cvat_id=1)
-    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1, status="annotation")
-    add_cvat_job_to_db(cvat_id=1, cvat_task_id=1, cvat_project_id=1, status="new")
-    (job, _) = get_cvat_job_from_db(1)
-    # Check if "update:job" event works with expired assignments
-    wallet_address = "0x86e83d346041E8806e352681f3F14549C0d2BC68"
-    add_asignment_to_db(wallet_address, 1, job.cvat_id, datetime.now(tz=timezone.utc))
+    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1)
+    job = add_cvat_job_to_db(
+        cvat_id=1, cvat_task_id=1, cvat_project_id=1, status=JobStatuses.in_progress
+    )
+
+    user_cvat_id = 1
+    add_assignment_to_db(WALLET_ADDRESS1, user_cvat_id, job.cvat_id, expires_at=utcnow())
+
+    if not is_last_assignment:
+        user_cvat_id += 1
+        add_assignment_to_db(WALLET_ADDRESS2, user_cvat_id, job.cvat_id, expires_at=utcnow())
 
     data = {
         "event": "update:job",
         "job": {
-            "url": api_url + "jobs/1",
+            "url": API_URL + "jobs/1",
             "id": 1,
             "task_id": 1,
             "project_id": 1,
             "state": "completed",
             "start_frame": 0,
             "stop_frame": 1,
             "assignee": {
-                "url": api_url + "users/1",
-                "id": 1,
+                "url": API_URL + f"users/{user_cvat_id}",
+                "id": user_cvat_id,
             },
-            "updated_date": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z",
+            "updated_date": (utcnow() + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z",
         },
         "before_update": {"state": "new", "assignee": None},
         "webhook_id": 1,
     }
 
-    signature = generate_cvat_signature(data)
-
-    with patch("src.handlers.cvat_events.cvat_api"):
+    with patch("src.handlers.cvat_events.cvat_api.update_job_assignee") as mock_update_job_assignee:
         response = client.post(
             "/cvat-webhook",
-            headers={"X-Signature-256": signature},
+            headers={"X-Signature-256": generate_cvat_signature(data)},
             json=data,
         )
 
     assert response.status_code == 200
 
-    (job, asignees) = get_cvat_job_from_db(1)
-    assert job.status == JobStatuses.new.value
-    assert asignees[0].status == AssignmentStatuses.expired.value
+    (job, assignments) = get_cvat_job_from_db(1)
+    assert job.status == JobStatuses.new
+    assert assignments[-1].status == AssignmentStatuses.expired
+    mock_update_job_assignee.assert_called_once_with(job.cvat_id, assignee_id=None)
+
+    if not is_last_assignment:
+        for assignment in assignments[:-1]:
+            assert assignment.status == AssignmentStatuses.created
 
 
-def test_incoming_webhook_200_update(client: TestClient) -> None:
+@pytest.mark.parametrize("assignment_status", AssignmentStatuses)
+def test_incoming_webhook_can_update_active_assignment(
+    client: TestClient, assignment_status: AssignmentStatuses
+):
     add_cvat_project_to_db(cvat_id=1)
-    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1, status="annotation")
-    add_cvat_job_to_db(cvat_id=1, cvat_task_id=1, cvat_project_id=1, status="new")
-    (job, _) = get_cvat_job_from_db(1)
-    # Check if "update:job" event works correctly
-    wallet_address = "0x86e83d346041E8806e352681f3F14549C0d2BC69"
-    add_asignment_to_db(wallet_address, 2, job.cvat_id, datetime.now() + timedelta(hours=1))
+    add_cvat_task_to_db(cvat_id=1, cvat_project_id=1)
+    job = add_cvat_job_to_db(
+        cvat_id=1, cvat_task_id=1, cvat_project_id=1, status=JobStatuses.in_progress
+    )
+    add_assignment_to_db(
+        WALLET_ADDRESS1,
+        1,
+        job.cvat_id,
+        status=assignment_status,
+        expires_at=datetime.now()
+        if assignment_status == AssignmentStatuses.expired
+        else datetime.now() + timedelta(hours=1),
+    )
 
     data = {
         "event": "update:job",
         "job": {
-            "url": api_url + "jobs/1",
+            "url": API_URL + "jobs/1",
             "id": 1,
             "task_id": 1,
             "project_id": 1,
             "state": "completed",
             "start_frame": 0,
             "stop_frame": 1,
             "assignee": {
-                "url": api_url + "users/1",
-                "id": 2,
+                "url": API_URL + "users/1",
+                "id": 1,
             },
-            "updated_date": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z",
+            "updated_date": utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z",
         },
-        "before_update": {"state": "new", "assignee": None},
+        "before_update": {"state": "in_progress", "assignee": None},
         "webhook_id": 1,
     }
 
-    signature = generate_cvat_signature(data)
-
-    with patch("src.handlers.cvat_events.cvat_api"):
+    with patch("src.handlers.cvat_events.cvat_api.update_job_assignee") as mock_update_job_assignee:
         response = client.post(
             "/cvat-webhook",
-            headers={"X-Signature-256": signature},
+            headers={"X-Signature-256": generate_cvat_signature(data)},
             json=data,
         )
 
     assert response.status_code == 200
 
-    (job, asignees) = get_cvat_job_from_db(1)
-    assert job.status == JobStatuses.completed.value
-    assert asignees[0].status == AssignmentStatuses.completed.value
-
-
-data = {
-    "event": "ping",
-}
+    (job, assignments) = get_cvat_job_from_db(1)
+    if assignment_status == AssignmentStatuses.created:
+        assert job.status == JobStatuses.completed
+        assert assignments[0].status == AssignmentStatuses.completed
+        mock_update_job_assignee.assert_called_once_with(job.cvat_id, assignee_id=None)
+    else:
+        assert job.status == JobStatuses.in_progress
+        assert assignments[0].status == assignment_status
+        mock_update_job_assignee.assert_not_called()
 
 
 def test_incoming_webhook_401_bad_signature(client: TestClient) -> None:
     # Send a request with bad signature
     response = client.post(
         "/cvat-webhook",
         headers={"X-Signature-256": "dummy_signature"},
-        json=data,
+        json=PING_EVENT_DATA,
     )
     assert response.status_code == 401
     assert response.json() == {"message": "Unauthorized"}
@@ -180,7 +205,7 @@ def test_incoming_webhook_401_bad_signature(client: TestClient) -> None:
 def test_incoming_webhook_401_without_signature(client: TestClient) -> None:
     response = client.post(
         "/cvat-webhook",
-        json=data,
+        json=PING_EVENT_DATA,
     )
 
     # Send a request without a signature