Daniil
156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from hmac import new
|
|
from typing import assert_type
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
|
|
from cpv3.modules.captions import service as captions_service
|
|
from cpv3.modules.tasks import service as task_service
|
|
|
|
|
|
def _make_document_json() -> dict:
|
|
return {
|
|
"segments": [
|
|
{
|
|
"text": "Привет",
|
|
"semantic_tags": [],
|
|
"structure_tags": [],
|
|
"time": {"start": 0.0, "end": 1.0},
|
|
"lines": [
|
|
{
|
|
"text": "Привет",
|
|
"semantic_tags": [],
|
|
"structure_tags": [],
|
|
"time": {"start": 0.0, "end": 1.0},
|
|
"words": [
|
|
{
|
|
"text": "Привет",
|
|
"semantic_tags": [],
|
|
"structure_tags": [],
|
|
"time": {"start": 0.0, "end": 1.0},
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload: dict) -> None:
|
|
self._payload = payload
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict:
|
|
return self._payload
|
|
|
|
|
|
def test_captions_generate_actor_sends_fallback_done_when_callback_not_confirmed(
|
|
monkeypatch,
|
|
) -> None:
|
|
sent_events: list[task_service.TaskWebhookEvent] = []
|
|
|
|
async def fake_generate_captions(**_: object) -> str:
|
|
return "render-123"
|
|
|
|
monkeypatch.setattr(captions_service, "generate_captions", fake_generate_captions)
|
|
monkeypatch.setattr(task_service, "_run_async", asyncio.run)
|
|
monkeypatch.setattr(task_service, "_raise_if_job_cancelled", lambda _job_id: None)
|
|
monkeypatch.setattr(
|
|
task_service,
|
|
"_send_webhook_event",
|
|
lambda _url, event: sent_events.append(event),
|
|
)
|
|
monkeypatch.setattr(task_service.time, "sleep", lambda _seconds: None)
|
|
monkeypatch.setattr(
|
|
task_service,
|
|
"get_settings",
|
|
lambda: SimpleNamespace(remotion_service_url="http://remotion.test"),
|
|
)
|
|
monkeypatch.setattr(
|
|
task_service.httpx,
|
|
"get",
|
|
lambda *_args, **_kwargs: _FakeResponse(
|
|
{
|
|
"status": "done",
|
|
"output_path": "projects/1/captioned/video.mp4",
|
|
"callback_delivered": False,
|
|
}
|
|
),
|
|
)
|
|
|
|
task_service.captions_generate_actor.fn(
|
|
job_id=str(uuid.uuid4()),
|
|
webhook_url="http://backend.test/api/tasks/webhook/job-1/",
|
|
video_s3_path="uploads/source.mp4",
|
|
folder="projects/1",
|
|
transcription_json=_make_document_json(),
|
|
style_config=None,
|
|
)
|
|
|
|
done_events = [
|
|
event for event in sent_events if event.status == task_service.JOB_STATUS_DONE
|
|
]
|
|
|
|
assert len(done_events) == 1
|
|
assert done_events[0].progress_pct == task_service.PROGRESS_COMPLETE
|
|
assert done_events[0].current_message == "Готово"
|
|
assert done_events[0].output_data == {
|
|
"output_path": "projects/1/captioned/video.mp4"
|
|
}
|
|
|
|
|
|
def test_captions_generate_actor_skips_fallback_when_done_webhook_confirmed(
|
|
monkeypatch,
|
|
) -> None:
|
|
sent_events: list[task_service.TaskWebhookEvent] = []
|
|
|
|
async def fake_generate_captions(**_: object) -> str:
|
|
return "render-123"
|
|
|
|
monkeypatch.setattr(captions_service, "generate_captions", fake_generate_captions)
|
|
monkeypatch.setattr(task_service, "_run_async", asyncio.run)
|
|
monkeypatch.setattr(task_service, "_raise_if_job_cancelled", lambda _job_id: None)
|
|
monkeypatch.setattr(
|
|
task_service,
|
|
"_send_webhook_event",
|
|
lambda _url, event: sent_events.append(event),
|
|
)
|
|
monkeypatch.setattr(task_service.time, "sleep", lambda _seconds: None)
|
|
monkeypatch.setattr(
|
|
task_service,
|
|
"get_settings",
|
|
lambda: SimpleNamespace(remotion_service_url="http://remotion.test"),
|
|
)
|
|
monkeypatch.setattr(
|
|
task_service.httpx,
|
|
"get",
|
|
lambda *_args, **_kwargs: _FakeResponse(
|
|
{
|
|
"status": "done",
|
|
"output_path": "projects/1/captioned/video.mp4",
|
|
"callback_delivered": True,
|
|
}
|
|
),
|
|
)
|
|
|
|
task_service.captions_generate_actor.fn(
|
|
job_id=str(uuid.uuid4()),
|
|
webhook_url="http://backend.test/api/tasks/webhook/job-1/",
|
|
video_s3_path="uploads/source.mp4",
|
|
folder="projects/1",
|
|
transcription_json=_make_document_json(),
|
|
style_config=None,
|
|
)
|
|
|
|
done_events = [
|
|
event for event in sent_events if event.status == task_service.JOB_STATUS_DONE
|
|
]
|
|
|
|
assert done_events == []
|