PlanOpticon

planopticon / tests / test_callbacks.py

Source Blame History 114 lines

0981a08…	noreply	1	"""Tests for video_processor.utils.callbacks.WebhookCallback."""
0981a08…	noreply	2
0981a08…	noreply	3	import json
0981a08…	noreply	4	from unittest.mock import patch
0981a08…	noreply	5
0981a08…	noreply	6	import pytest
0981a08…	noreply	7
0981a08…	noreply	8	from video_processor.utils.callbacks import WebhookCallback
0981a08…	noreply	9
0981a08…	noreply	10
0981a08…	noreply	11	@pytest.fixture()
0981a08…	noreply	12	def callback():
0981a08…	noreply	13	return WebhookCallback(url="https://example.com/webhook")
0981a08…	noreply	14
0981a08…	noreply	15
0981a08…	noreply	16	# --- Constructor ---
0981a08…	noreply	17
0981a08…	noreply	18
0981a08…	noreply	19	def test_default_headers():
0981a08…	noreply	20	cb = WebhookCallback(url="https://example.com/hook")
0981a08…	noreply	21	assert cb.headers == {"Content-Type": "application/json"}
0981a08…	noreply	22
0981a08…	noreply	23
0981a08…	noreply	24	def test_custom_headers():
0981a08…	noreply	25	headers = {"Authorization": "Bearer tok", "Content-Type": "application/json"}
0981a08…	noreply	26	cb = WebhookCallback(url="https://example.com/hook", headers=headers)
0981a08…	noreply	27	assert cb.headers["Authorization"] == "Bearer tok"
0981a08…	noreply	28
0981a08…	noreply	29
0981a08…	noreply	30	def test_custom_timeout():
0981a08…	noreply	31	cb = WebhookCallback(url="https://example.com/hook", timeout=5.0)
0981a08…	noreply	32	assert cb.timeout == 5.0
0981a08…	noreply	33
0981a08…	noreply	34
0981a08…	noreply	35	# --- _post ---
0981a08…	noreply	36
0981a08…	noreply	37
0981a08…	noreply	38	@patch("urllib.request.urlopen")
0981a08…	noreply	39	@patch("urllib.request.Request")
0981a08…	noreply	40	def test_post_sends_json_payload(mock_request_cls, mock_urlopen, callback):
0981a08…	noreply	41	callback._post({"event": "test"})
0981a08…	noreply	42
0981a08…	noreply	43	mock_request_cls.assert_called_once()
0981a08…	noreply	44	call_args = mock_request_cls.call_args
0981a08…	noreply	45	data = json.loads(call_args[1]["data"] if "data" in call_args[1] else call_args[0][1])
0981a08…	noreply	46	assert data["event"] == "test"
0981a08…	noreply	47	mock_urlopen.assert_called_once()
0981a08…	noreply	48
0981a08…	noreply	49
0981a08…	noreply	50	@patch("urllib.request.urlopen", side_effect=Exception("Connection refused"))
0981a08…	noreply	51	@patch("urllib.request.Request")
0981a08…	noreply	52	def test_post_logs_failure_does_not_raise(mock_request_cls, mock_urlopen, callback):
0981a08…	noreply	53	# Should not raise
0981a08…	noreply	54	callback._post({"event": "fail_test"})
0981a08…	noreply	55
0981a08…	noreply	56
0981a08…	noreply	57	# --- on_step_start ---
0981a08…	noreply	58
0981a08…	noreply	59
0981a08…	noreply	60	@patch.object(WebhookCallback, "_post")
0981a08…	noreply	61	def test_on_step_start_payload(mock_post, callback):
0981a08…	noreply	62	callback.on_step_start("transcription", 1, 5)
0981a08…	noreply	63
0981a08…	noreply	64	mock_post.assert_called_once_with(
0981a08…	noreply	65	{
0981a08…	noreply	66	"event": "step_start",
0981a08…	noreply	67	"step": "transcription",
0981a08…	noreply	68	"index": 1,
0981a08…	noreply	69	"total": 5,
0981a08…	noreply	70	}
0981a08…	noreply	71	)
0981a08…	noreply	72
0981a08…	noreply	73
0981a08…	noreply	74	# --- on_step_complete ---
0981a08…	noreply	75
0981a08…	noreply	76
0981a08…	noreply	77	@patch.object(WebhookCallback, "_post")
0981a08…	noreply	78	def test_on_step_complete_payload(mock_post, callback):
0981a08…	noreply	79	callback.on_step_complete("analysis", 3, 5)
0981a08…	noreply	80
0981a08…	noreply	81	mock_post.assert_called_once_with(
0981a08…	noreply	82	{
0981a08…	noreply	83	"event": "step_complete",
0981a08…	noreply	84	"step": "analysis",
0981a08…	noreply	85	"index": 3,
0981a08…	noreply	86	"total": 5,
0981a08…	noreply	87	}
0981a08…	noreply	88	)
0981a08…	noreply	89
0981a08…	noreply	90
0981a08…	noreply	91	# --- on_progress ---
0981a08…	noreply	92
0981a08…	noreply	93
0981a08…	noreply	94	@patch.object(WebhookCallback, "_post")
0981a08…	noreply	95	def test_on_progress_payload(mock_post, callback):
0981a08…	noreply	96	callback.on_progress("transcription", 42.5, "Processing chunk 3/7")
0981a08…	noreply	97
0981a08…	noreply	98	mock_post.assert_called_once_with(
0981a08…	noreply	99	{
0981a08…	noreply	100	"event": "progress",
0981a08…	noreply	101	"step": "transcription",
0981a08…	noreply	102	"percent": 42.5,
0981a08…	noreply	103	"message": "Processing chunk 3/7",
0981a08…	noreply	104	}
0981a08…	noreply	105	)
0981a08…	noreply	106
0981a08…	noreply	107
0981a08…	noreply	108	@patch.object(WebhookCallback, "_post")
0981a08…	noreply	109	def test_on_progress_default_message(mock_post, callback):
0981a08…	noreply	110	callback.on_progress("extraction", 100.0)
0981a08…	noreply	111
0981a08…	noreply	112	payload = mock_post.call_args[0][0]
0981a08…	noreply	113	assert payload["message"] == ""
0981a08…	noreply	114	assert payload["percent"] == 100.0

PlanOpticon

Keyboard Shortcuts