"""Tests for Handlers.""" import json from datetime import datetime from http import HTTPStatus from typing import Any from unittest.mock import MagicMock, patch from urllib.parse import urlencode import pytest from botocore.errorfactory import ClientError from flask import Response, g from jsonschema import ValidationError from requests.exceptions import ConnectionError from sqlalchemy.exc import SQLAlchemyError from werkzeug.exceptions import BadRequest, Conflict, HTTPException, NotFound from video import config, handlers from video.api import app from video.constants import ( error as error_constants, header, job_io_fields, job_statuses, job_types, ) from video.exceptions import ( AccessForbidden, ApprovalNotFound, ArtistNotFound, ArtistOwnershipError, AutoCarveoutNotFound, InvalidRequest, JobNotFound, OARequired, ProductManagerNotFound, ProductOwnershipError, ProductVideoNotFound, ProductVideoSizingRuleNotFound, ProjectNotFound, ProjectOwnershipError, ReleaseNotFound, SubaccountNotFound, SubmissionNotAllowed, TrackNotFound, VideoAssetNotFound, VideoTypeNotFound, VideoValidationError, ) from video.logic import ( access_control, approval as approval_logic, asset as asset_logic, cloudfront as cloudfront_logic, context as context_logic, ingest as ingest_logic, job as job_logic, metadata as metadata_logic, product as product_logic, stream as stream_logic, thumbnails as thumbnails_logic, track as track_metadata_logic, video_asset as video_asset_logic, video_dashboard_item as video_dashboard_item_logic, video_product_copy as video_product_copy_logic, video_type as video_type_logic, ) from video.models.sql import queries from video.models.sql.classes import release from tests.unit.factories.ows.video import VideoJobFactory def test_invalid_request_exception_handler() -> None: """Test handler for InvalidRequest exceptions.""" invalid_request_exception = InvalidRequest("No good!") with app.test_request_context(): g.log = MagicMock(exception=MagicMock()) server_response = handlers.http_exception_handler(invalid_request_exception) assert server_response.status_code == HTTPStatus.BAD_REQUEST response_message = json.loads(server_response.data.decode()) assert response_message["message"] == "No good!" assert response_message["code"] == "bad_request" def test_validation_error_exception_handler() -> None: """Test handler for ValidationError exceptions.""" validation_error = ValidationError( validator="nut", validator_value="coconut", message="YumMMMMMMmmmMMmm!", ) with app.test_request_context(): server_response = handlers.validation_error_exception_handler(validation_error) assert server_response.status_code == HTTPStatus.BAD_REQUEST response_message = json.loads(server_response.data.decode()) assert response_message["message"] == { "validator": "nut", "validator_value": "coconut", "message": "YumMMMMMMmmmMMmm!", } assert response_message["code"] == error_constants.ERROR_CODE_VALIDATION_ERROR def test_sqlalchemy_error_exception_handler() -> None: """Test sqlalchemy_error_exception_handler.""" exception = SQLAlchemyError() with app.test_request_context(): g.log = MagicMock(exception=MagicMock()) server_response = handlers.sqlalchemy_error_exception_handler(exception) g.log.exception.assert_called_once_with(exception) assert server_response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR response_message = json.loads(server_response.data.decode()) assert response_message["message"] is None assert response_message["code"] == "internal_error" def test_exception_handler() -> None: """Verify exception_handler returns 500 status code and json payload.""" exception = Exception("something went wrong") with app.test_request_context(): g.log = MagicMock(exception=MagicMock()) server_response = handlers.exception_handler(exception) g.log.exception.assert_called_once_with(exception) assert server_response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR response_message = json.loads(server_response.data.decode()) assert ( response_message["message"] == "I have erred and am unable to complete your request." ) assert response_message["code"] == "internal_error" def test_botocore_client_error_no_such_key() -> None: """NoSuchKey ClientError returns 400.""" error = ClientError({"Error": {"Code": "NoSuchKey"}}, "GetObject") with app.test_request_context(): g.log = MagicMock(warning=MagicMock(), exception=MagicMock()) server_response = handlers.botocore_client_error_exception_handler(error) g.log.warning.assert_called_once_with(error) g.log.exception.assert_not_called() assert server_response.status_code == HTTPStatus.BAD_REQUEST response_message = json.loads(server_response.data.decode()) assert ( response_message["message"] == "Oopsies, I don't know how to respond to that ¯\\_(ツ)_/¯" ) assert response_message["code"] == error_constants.ERROR_CODE_BAD_REQUEST def test_botocore_client_error_other() -> None: """Non-NoSuchKey ClientError delegates to exception_handler (500).""" error = ClientError({"Error": {"Code": "AccessDenied"}}, "GetObject") with app.test_request_context(): g.log = MagicMock(exception=MagicMock()) server_response = handlers.botocore_client_error_exception_handler(error) g.log.exception.assert_called_once_with(error) assert server_response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR @pytest.mark.parametrize( "mock_exception, status_code, message, error_code", [ ( NotFound("not found error"), HTTPStatus.NOT_FOUND, "not found error", "not_found", ), ( BadRequest("bad request"), HTTPStatus.BAD_REQUEST, "bad request", "bad_request", ), (Conflict("conflict"), HTTPStatus.CONFLICT, "conflict", "conflict"), (InvalidRequest(), HTTPStatus.BAD_REQUEST, "bad request", "bad_request"), ( VideoValidationError(["badChannel"]), HTTPStatus.BAD_REQUEST, ["badChannel"], "validation_error", ), (ApprovalNotFound(), HTTPStatus.NOT_FOUND, "approval not found", "not_found"), ( AutoCarveoutNotFound(), HTTPStatus.NOT_FOUND, "auto carveout not found", "not_found", ), ( ProductManagerNotFound(), HTTPStatus.NOT_FOUND, "product manager not found", "not_found", ), ( ProductVideoNotFound(), HTTPStatus.NOT_FOUND, "product video not found", "not_found", ), ( ProductVideoSizingRuleNotFound(), HTTPStatus.NOT_FOUND, "product video sizing rule not found", "not_found", ), (ReleaseNotFound(), HTTPStatus.NOT_FOUND, "release not found", "not_found"), ( VideoAssetNotFound(), HTTPStatus.NOT_FOUND, "video asset not found", "not_found", ), ( VideoTypeNotFound(), HTTPStatus.NOT_FOUND, "video type not found", "not_found", ), (TrackNotFound(), HTTPStatus.NOT_FOUND, "track not found", "not_found"), (JobNotFound(), HTTPStatus.NOT_FOUND, "job not found", "not_found"), (ArtistNotFound(), HTTPStatus.NOT_FOUND, "artist not found", "not_found"), ( SubaccountNotFound(), HTTPStatus.NOT_FOUND, "subaccount not found", "not_found", ), (ProjectNotFound(), HTTPStatus.NOT_FOUND, "project not found", "not_found"), ( OARequired(), HTTPStatus.UNAUTHORIZED, "Must be oa", "UNAUTHORIZED", ), ( AccessForbidden(), HTTPStatus.FORBIDDEN, "access forbidden", "authorization_error", ), (SubmissionNotAllowed(), HTTPStatus.FORBIDDEN, "", "authorization_error"), ( ArtistOwnershipError(), HTTPStatus.FORBIDDEN, "artist ownership verification failed", "authorization_error", ), ( ProductOwnershipError(), HTTPStatus.FORBIDDEN, "product ownership verification failed", "authorization_error", ), ( ProjectOwnershipError(), HTTPStatus.FORBIDDEN, "project ownership verification failed", "authorization_error", ), ], ) def test_http_exception_handler( app_context: Any, mock_exception: HTTPException, status_code: int, message: object, error_code: str, ) -> None: """Verify http_exception_handler returns correct status code and message.""" server_response = handlers.http_exception_handler(mock_exception) server_response_json = json.loads(server_response.data.decode()) assert server_response.status_code == status_code assert server_response_json["message"] == message assert server_response_json["code"] == error_code def test_request_exception_handler() -> None: """RequestException returns 502 with error details.""" error = ConnectionError("connection refused") with app.test_request_context(): g.log = MagicMock(exception=MagicMock()) server_response = handlers.request_exception_handler(error) g.log.exception.assert_called_once_with(error) assert server_response.status_code == HTTPStatus.BAD_GATEWAY response_message = json.loads(server_response.data.decode()) assert response_message["code"] == "ConnectionError" assert response_message["message"] == "connection refused" user_forbidden_message = { "message": "User is forbidden", "code": "authorization_error", } def generate_test_manage_job_test_cases() -> list[Any]: """Generate test_manage_job test cases.""" test_cases: list[Any] = [] for job_type in list(job_types.JOB_TYPES): test_cases = test_cases + [ ( "test create job: {}".format(job_type), { "inputs": VideoJobFactory.build( job_status=job_statuses.SETUP, job_type=job_type ).inputs, "context": { "product_id": 112428, }, "type": job_type, }, None, HTTPStatus.OK, VideoJobFactory.build(job_type=job_type).to_dict(), ), ( "test change status: {}".format(job_type), { "id": 111, "status": job_statuses.PROGRESSING, }, { "type": job_type, "status": job_statuses.SUBMITTED, }, HTTPStatus.OK, VideoJobFactory.build( job_type=job_type, job_status=job_statuses.SUBMITTED ).to_dict(), ), ( "test add inputs: {}".format(job_type), { "id": 111, "inputs": VideoJobFactory.build( job_status=job_statuses.SETUP, job_type=job_type ).inputs, }, { "type": job_type, "status": job_statuses.SETUP, }, HTTPStatus.OK, VideoJobFactory.build( job_type=job_type, job_status=job_statuses.SETUP ).to_dict(), ), ( "test add outputs: {}".format(job_type), { "id": 111, "outputs": VideoJobFactory.build( job_status=job_statuses.PROGRESSING, job_type=job_type ).outputs, }, { "type": job_type, "status": job_statuses.PROGRESSING, }, HTTPStatus.OK, VideoJobFactory.build( job_type=job_type, job_status=job_statuses.PROGRESSING ).to_dict(), ), ] return test_cases @pytest.mark.parametrize( ( "test_description", "request_data", "existing_job", "expected_status_code", "expected_data", ), generate_test_manage_job_test_cases(), ) def test_manage_job( mocker: Any, test_description: str, request_data: dict[str, Any], existing_job: Any, expected_status_code: int, expected_data: dict[str, Any], ) -> None: """Test jobs handler.""" mocker.patch.object( access_control, "check_access_to_product", autospec=True, ) mocker.patch.object( access_control, "check_access_to_job", autospec=True, ) mocker.patch.object( job_logic, "persist_new_job_data", return_value=expected_data, autospec=True, ) mocker.patch.object( context_logic, "add_to_context", autospec=True, ) mocker.patch.object( queries, "get_jobs", return_value=[existing_job], autospec=True, ) with app.test_request_context( data=json.dumps(request_data), headers={ "Content-Type": "application/json", }, ): create_job_response = handlers.manage_job() assert expected_status_code == create_job_response.status_code data = json.loads(create_job_response.get_data().decode()) assert expected_data == data @pytest.mark.parametrize("job_id", [0, -1, 1.7, True, False]) def test_manage_job_invalid_id_raises(job_id: object) -> None: """Test that invalid id values are rejected by schema validation.""" with pytest.raises(ValidationError): with app.test_request_context( data=json.dumps({"id": job_id}), headers={"Content-Type": "application/json"}, ): handlers.manage_job() def test_before_request_hook(mocker: Any) -> None: """Test before_request hook.""" create_context_mock = mocker.patch.object( context_logic, "create_context", autospec=True ) with app.test_request_context(): handlers.before_request() create_context_mock.assert_called_once_with() def test_after_request_hook() -> None: """Test after_request_hook.""" resp = Response("passee") assert handlers.after_request(resp) is resp test_get_job_video_job_a = VideoJobFactory.build().to_dict() @pytest.mark.parametrize( ( "test_description", "get_jobs_logic_response", "expected_status_code", "expected_data", ), [ ( "test get_jobs logic response.", [test_get_job_video_job_a], HTTPStatus.OK, test_get_job_video_job_a, ), ( "test empty get_jobs logic response.", [], HTTPStatus.NOT_FOUND, {"message": "Job not found", "code": "not_found"}, ), ], ) def test_get_job( mocker: Any, test_description: str, get_jobs_logic_response: list[Any], expected_status_code: int, expected_data: dict[str, Any], ) -> None: """Test get_job.""" get_jobs_mock = mocker.patch.object( job_logic, "get_jobs", return_value=get_jobs_logic_response ) with app.test_request_context(): get_jobs_response = handlers.get_job(123) get_jobs_mock.assert_called_once_with({"id": 123}) assert expected_status_code == get_jobs_response.status_code data = json.loads(get_jobs_response.get_data().decode()) assert expected_data == data test_get_jobs_video_job_a = VideoJobFactory.build().to_dict() test_get_jobs_video_job_b = VideoJobFactory.build().to_dict() test_get_jobs_video_job_c = VideoJobFactory.build().to_dict() test_get_jobs_video_jobs = [ test_get_jobs_video_job_a, test_get_jobs_video_job_b, test_get_jobs_video_job_c, ] test_jobs_video_job_filters = VideoJobFactory.build().to_dict() @pytest.mark.parametrize( ( "test_description", "query_params", "parsed_query_params", "get_jobs_logic_response", "expected_status_code", "expected_data", ), [ ( "test get_jobs with no query params and no jobs found", {}, { "id": None, "parent_id": None, }, [], HTTPStatus.OK, { "items": [], "pagination": { "type": "standard", "offset": "0", "limit": config.QUERY_RECORD_LIMIT, "total_records": 0, }, }, ), ( "test get_jobs with no query params", {}, { "id": None, "parent_id": None, }, test_get_jobs_video_jobs, HTTPStatus.OK, { "items": test_get_jobs_video_jobs, "pagination": { "type": "standard", "offset": "0", "limit": config.QUERY_RECORD_LIMIT, "total_records": len(test_get_jobs_video_jobs), }, }, ), ( "test get_jobs with all query params", { "job_id": test_jobs_video_job_filters["id"], "job_parent_id": test_jobs_video_job_filters["parent_id"], }, { "id": test_jobs_video_job_filters["id"], "parent_id": test_jobs_video_job_filters["parent_id"], }, test_get_jobs_video_jobs, HTTPStatus.OK, { "items": test_get_jobs_video_jobs, "pagination": { "type": "standard", "offset": "0", "limit": config.QUERY_RECORD_LIMIT, "total_records": len(test_get_jobs_video_jobs), }, }, ), ( "test get_jobs with deprecated job type in response.", {}, { "id": None, "parent_id": None, }, [{"type": job_types.DEPRECATED_JOB_TYPES[0]}], HTTPStatus.OK, { "items": [{"type": job_types.DEPRECATED_JOB_TYPES[0]}], "pagination": { "type": "standard", "offset": "0", "limit": config.QUERY_RECORD_LIMIT, "total_records": 1, }, }, ), ], ) def test_get_jobs( mocker: Any, test_description: str, query_params: dict[str, Any], parsed_query_params: dict[str, Any], get_jobs_logic_response: list[Any], expected_status_code: int, expected_data: dict[str, Any], ) -> None: """Test get_job.""" get_jobs_mock = mocker.patch.object( job_logic, "get_jobs", return_value=get_jobs_logic_response ) with app.test_request_context(query_string=urlencode(query_params)): get_jobs_response = handlers.get_jobs() get_jobs_mock.assert_called_once_with(parsed_query_params) assert expected_status_code == get_jobs_response.status_code data = json.loads(get_jobs_response.get_data().decode()) assert expected_data == data @pytest.mark.parametrize( ( "test_description", "dashboard_item_id", "query_params", "get_new_assets_logic_response", "expected_status_code", "expected_data", ), [ ( "test get_new_assets logic response.", 123, { "offset": 0, }, [ { "dashboard_item_id": 124, "asset_id": 12345, "product_id": 54321, } ], HTTPStatus.OK, { "items": [ { "dashboard_item_id": 124, "asset_id": 12345, "product_id": 54321, } ], "pagination": {"type": "standard", "offset": 0, "total_records": 99}, }, ), ( "test empty get_new_assets logic response.", 123, { "offset": 0, }, [], HTTPStatus.OK, { "items": [], "pagination": {"type": "standard", "offset": 0, "total_records": 99}, }, ), ( "test get_new_assets offset.", 123, { "offset": 1, }, [ { "dashboard_item_id": 125, "asset_id": 12345, "product_id": 54321, } ], HTTPStatus.OK, { "items": [ { "dashboard_item_id": 125, "asset_id": 12345, "product_id": 54321, } ], "pagination": {"type": "standard", "offset": 1, "total_records": 99}, }, ), ], ) def test_get_new_assets( mocker: Any, test_description: str, dashboard_item_id: int, query_params: dict[str, Any], get_new_assets_logic_response: list[Any], expected_status_code: int, expected_data: dict[str, Any], ) -> None: """Test get_new_assets.""" get_new_assets_mock = mocker.patch.object( video_dashboard_item_logic, "get_new_assets", return_value=(get_new_assets_logic_response, 99), ) with app.test_request_context(query_string=urlencode(query_params)): get_new_assets_response = handlers.get_new_assets(dashboard_item_id) get_new_assets_mock.assert_called_once_with( dashboard_item_id, query_params.get("offset") ) assert expected_status_code == get_new_assets_response.status_code data = json.loads(get_new_assets_response.get_data().decode()) assert expected_data == data def test_get_max_dashboard_item_id(mocker: Any) -> None: """Test get_max_dashboard_item_id.""" get_max_dashboard_item_id_mock = mocker.patch.object( video_dashboard_item_logic, "get_max_dashboard_item_id", return_value=12345 ) with app.test_request_context(): get_max_dashboard_item_id_response = handlers.get_max_dashboard_item_id() get_max_dashboard_item_id_mock.assert_called_once() assert get_max_dashboard_item_id_response.status_code == HTTPStatus.OK data = json.loads(get_max_dashboard_item_id_response.get_data().decode()) expected_data = { "max_dashboard_item_id": 12345, } assert data == expected_data def test_get_hls_streaming_url(mocker: Any) -> None: """Test get hls streaming url endpoint.""" product_id = 1 get_hls_streaming_url_mock = mocker.patch.object( stream_logic, "get_hls_streaming_url", return_value="https://example.com/stream.m3u8", ) with app.test_request_context(): result = handlers.get_hls_streaming_url(product_id) get_hls_streaming_url_mock.assert_called_once_with( product_id, referrer=None, user_id=None ) assert result.status_code == HTTPStatus.OK def test_create_product(mocker: Any) -> None: """Create product.""" create_product_mock = mocker.patch.object( product_logic, "create_product", return_value={} ) with app.test_request_context( data=json.dumps({"project_id": 12}), headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", }, ): create_product_response = handlers.create_product() create_product_mock.assert_called_once_with({"project_id": 12}, "vendor", "12") assert create_product_response.status_code == HTTPStatus.OK @pytest.mark.parametrize( ("payload", "account_type", "account_id"), [ ({"account_id": 1000}, "vendor", 1000), ({"subaccount_id": 2000}, "subaccount", 2000), ], ) def test_create_product_by_oa_user( mocker: Any, payload: dict[str, Any], account_type: str, account_id: int ) -> None: """Create product by oa user.""" create_product_mock = mocker.patch.object( product_logic, "create_product", return_value={} ) payload["project_id"] = 12 with app.test_request_context( data=json.dumps(payload), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12345", }, ): create_product_response = handlers.create_product() create_product_mock.assert_called_once_with(payload, account_type, account_id) assert create_product_response.status_code == HTTPStatus.OK def test_delete_product(mocker: Any) -> None: """Delete product.""" product_id = 123 delete_mock = mocker.patch.object(product_logic, "delete", return_value={}) with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", } ): create_product_response = handlers.delete_product(product_id) delete_mock.assert_called_once_with(product_id, "vendor", "12") assert create_product_response.status_code == HTTPStatus.OK def test_save_metadata(mocker: Any) -> None: """Save metadata.""" save_metadata_mock = mocker.patch.object( metadata_logic, "save_metadata", return_value={"a": datetime(2018, 1, 12)}, ) product_id = 12 with app.test_request_context( data=json.dumps({"type_of_video": "Live Performance"}), headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", header.ORCHARD_USER_ID: "alw:123", }, ): save_metadata_response = handlers.save_metadata(product_id) save_metadata_mock.assert_called_once_with( product_id, {"type_of_video": "Live Performance"}, "vendor", "12", "alw:123" ) assert save_metadata_response.status_code == HTTPStatus.OK assert save_metadata_response.get_data().decode() == '{"a": "2018-01-12T00:00:00"}' @pytest.mark.parametrize( ("payload", "account_type", "account_id", "save_metadata_payload"), [ ( {"account_id": 1000, "subaccount_id": None}, "vendor", 1000, { "account_id": 1000, "subaccount_id": None, "type_of_video": "Live Performance", }, ), ( {"account_id": None, "subaccount_id": 2000}, "subaccount", 2000, { "account_id": None, "subaccount_id": 2000, "type_of_video": "Live Performance", }, ), ( {"account_id": 1000, "subaccount_id": 2000}, "subaccount", 2000, { "account_id": 1000, "subaccount_id": 2000, "type_of_video": "Live Performance", }, ), ({}, None, None, {"type_of_video": "Live Performance"}), ], ) def test_save_metadata_by_oa_user( mocker: Any, payload: dict[str, Any], account_type: Any, account_id: Any, save_metadata_payload: dict[str, Any], ) -> None: """Save metadata by oa_user.""" save_metadata_mock = mocker.patch.object( metadata_logic, "save_metadata", return_value={"a": datetime(2018, 1, 12)}, ) product_id = 12 payload["type_of_video"] = "Live Performance" with app.test_request_context( data=json.dumps(payload), headers={"Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:123"}, ): save_metadata_response = handlers.save_metadata(product_id) save_metadata_mock.assert_called_once_with( product_id, save_metadata_payload, account_type, account_id, "oa:123" ) assert save_metadata_response.status_code == HTTPStatus.OK assert save_metadata_response.get_data().decode() == '{"a": "2018-01-12T00:00:00"}' def test_get_tracks_metadata_with_tuids(mocker: Any) -> None: """Get tracks metadata.""" product_id = 123 get_tracks_metadata_mock = mocker.patch.object( track_metadata_logic, "get_tracks_metadata", return_value={} ) with app.test_request_context(path="/?tuids=123,3423"): result = handlers.get_tracks_metadata(product_id) get_tracks_metadata_mock.assert_called_once_with(product_id, [123, 3423]) assert result.status_code == HTTPStatus.OK def test_get_tracks_metadata_without_tuids(mocker: Any) -> None: """Get tracks metadata.""" product_id = 123 get_tracks_metadata_mock = mocker.patch.object( track_metadata_logic, "get_tracks_metadata", return_value={} ) with app.test_request_context(path="/"): result = handlers.get_tracks_metadata(product_id) get_tracks_metadata_mock.assert_called_once_with(product_id, None) assert result.status_code == HTTPStatus.OK def test_save_track_metadata(mocker: Any) -> None: """Save track metadata.""" create_track_metadata_mock = mocker.patch.object( track_metadata_logic, "create_track_metadata", return_value={} ) product_id = 12 data = { "fps": 30, "resolution": "720HD", "aspect_ratio": "16:9", "color": "color", "channel": 2, "duration_minutes": 23, "duration_seconds": 45, "closed_caption_exists": 0, "closed_caption_reason_id": 1, } with app.test_request_context( data=json.dumps(data), headers={"Content-Type": "application/json"} ): save_track_metadata_response = handlers.create_track_metadata(product_id) create_track_metadata_mock.assert_called_once_with(product_id, data) assert save_track_metadata_response.status_code == HTTPStatus.CREATED def test_load_metadata(mocker: Any) -> None: """Load metadata.""" load_metadata_mock = mocker.patch.object( metadata_logic, "load_metadata", return_value={"a": datetime(2018, 1, 12)}, ) product_id = 12 with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", } ): load_metadata_response = handlers.load_metadata(product_id) load_metadata_mock.assert_called_once_with(product_id, "vendor", "12") assert load_metadata_response.status_code == HTTPStatus.OK assert load_metadata_response.get_data().decode() == '{"a": "2018-01-12T00:00:00"}' def test_get_available_channels(mocker: Any) -> None: """Get available channels.""" get_available_channels_mock = mocker.patch.object( metadata_logic, "get_available_channels", return_value=[], ) product_id = 12 with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", } ): get_channels_response = handlers.get_available_channels(product_id) get_available_channels_mock.assert_called_once_with( product_id, "vendor", "12", None ) assert get_channels_response.status_code == HTTPStatus.OK assert get_channels_response.get_data().decode() == "[]" def test_get_available_channels_oa_user(mocker: Any) -> None: """Get available channels for OA user.""" get_available_channels_mock = mocker.patch.object( metadata_logic, "get_available_channels", return_value=[], ) product_id = 12 with app.test_request_context( headers={"Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12345"} ): get_channels_response = handlers.get_available_channels(product_id) get_available_channels_mock.assert_called_once_with( product_id, None, None, "oa:12345" ) assert get_channels_response.status_code == HTTPStatus.OK assert get_channels_response.get_data().decode() == "[]" def test_get_available_channels_for_project(mocker: Any) -> None: """Get available channels.""" get_available_channels_by_project_id_mock = mocker.patch.object( metadata_logic, "get_available_channels_by_project_id", return_value=[], ) project_id = 123 with app.test_request_context( headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12345", }, data=json.dumps({"account_id": "10001"}), ): get_channels_response = handlers.get_available_channels_by_project_id( project_id ) get_available_channels_by_project_id_mock.assert_called_once_with( project_id, "vendor", "10001" ) assert get_channels_response.status_code == HTTPStatus.OK assert get_channels_response.get_data().decode() == "[]" def test_get_available_channels_for_project_bad_headers(mocker: Any) -> None: """Get available channels with bad headers.""" mocker.patch.object( metadata_logic, "get_available_channels_by_project_id", return_value=[], ) project_id = 123 with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_ID: "12345", header.GRASS_ACCOUNT_TYPE: "vendor", } ): get_channels_response = handlers.get_available_channels_by_project_id( project_id ) assert get_channels_response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR response_message = json.loads(get_channels_response.get_data().decode()) assert response_message["message"] == "bad_grass_request_error" def test_get_approvals(mocker: Any) -> None: """Get approvals.""" get_mock = mocker.patch.object(approval_logic, "get", return_value="response") product_id = 12 with app.test_request_context( headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12", } ): get_approvals_response = handlers.get_approvals(product_id) get_mock.assert_called_once_with(product_id) assert get_approvals_response.get_data().decode() == "response" def test_change_approvals(mocker: Any) -> None: """Change approvals.""" product_id = 12 product_vendor_id = 12321 change_mock = mocker.patch.object( approval_logic, "change", autospec=True, return_value={"a": datetime(2018, 1, 12)}, ) mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={"vendor_id": product_vendor_id}, ) body = {"approval_type": "final", "value": "abc"} with app.test_request_context( data=json.dumps(body), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12", }, ): change_approvals_response = handlers.change_approval(product_id) change_mock.assert_called_once_with(product_id, "oa:12", body) assert ( change_approvals_response.get_data().decode() == '{"a": "2018-01-12T00:00:00"}' ) @pytest.mark.parametrize( "body, error_message", [ ({}, "'approval_type' is a required property"), ({"approval_type": "final"}, "'value' is a required property"), ( {"approval_type": "new", "value": "nope"}, "'new' is not one of ['final', 'rejection', 'release', 'content']", ), ( {"approval_type": "final", "value": "nope", "bypass_validation": "yes"}, "'yes' is not of type 'boolean'", ), ( {"approval_type": "final", "value": "nope", "extra": "data"}, "Additional properties are not allowed ('extra' was unexpected)", ), ( {"approval_type": "final", "value": "nope", "upc": False}, "False is not of type 'string', 'integer', 'null'", ), ], ) def test_change_approvals_validation( mocker: Any, body: dict[str, Any], error_message: str ) -> None: """Change approval body validation errors.""" product_id = 12 product_vendor_id = 12321 mocker.patch.object( approval_logic, "change", autospec=True, return_value={"a": datetime(2018, 1, 12)}, ) mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={"vendor_id": product_vendor_id}, ) with pytest.raises(ValidationError) as e: with app.test_request_context( data=json.dumps(body), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12", }, ): handlers.change_approval(product_id) assert e.value.message == error_message def test_change_approvals_no_access(mocker: Any) -> None: """Change approvals.""" mocker.patch.object(approval_logic, "change", return_value="response") product_id = 12 with pytest.raises(OARequired): with app.test_request_context( data=json.dumps({"": ""}), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "alw:12", }, ): handlers.change_approval(product_id) def test_change_approvals_validation_errors(mocker: Any) -> None: """Change approval raises VideoValidationError when logic detects validation errors.""" product_id = 12 product_vendor_id = 12321 mocker.patch.object( approval_logic, "change", autospec=True, side_effect=VideoValidationError(["badChannel"]), ) mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={"vendor_id": product_vendor_id}, ) body = {"approval_type": "final", "value": True} with pytest.raises(VideoValidationError) as exc_info: with app.test_request_context( data=json.dumps(body), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:12", }, ): handlers.change_approval(product_id) assert exc_info.value.validation_errors == ["badChannel"] def test_submit_product(mocker: Any) -> None: """Submit product.""" submit_mock = mocker.patch.object( product_logic, "submit", return_value={"a": datetime(2018, 1, 12)}, ) product_id = 12 with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", } ): submit_response = handlers.submit(product_id) submit_mock.assert_called_once_with(product_id, "vendor", "12") assert submit_response.status_code == HTTPStatus.OK assert submit_response.get_data().decode() == '{"a": "2018-01-12T00:00:00"}' def test_setup_workflows_not_found_error(mocker: Any) -> None: """Setup workflows.""" setup_ingest_from_browser_workflow_mock = mocker.patch.object( job_logic, "setup_ingest_from_browser_workflow", autospec=True ) with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", } ): create_workflow_response = handlers.setup_workflow("NONEXISTENT_WORKFLOW") assert not setup_ingest_from_browser_workflow_mock.called assert create_workflow_response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize( "workflow_job_type, workflow_function_name", [(job_types.WORKFLOW_INGEST_FROM_BROWSER, "setup_ingest_from_browser_workflow")], ) def test_setup_workflow_ingest_from_browser_success( mocker: Any, workflow_job_type: str, workflow_function_name: str ) -> None: """Test setup work flows.""" expected_workflow_response = [ { "context": { "correlation_id": "3163f53e-e2c9-11e8-865e-0242ac110002.1", "datetime": "2018-11-07T20:10:38Z", "id": 3750, "subaccount_id": None, "user_id": "alw:17411", "vendor_id": 16156, }, "id": 5992, "inputs": None, "outputs": None, "parent_id": 5986, "status": "SUBMITTED", "type": "detect_max_volume", } ] workflow_function_name_mock = mocker.patch.object( job_logic, workflow_function_name, autospec=True, return_value=expected_workflow_response, ) with app.test_request_context( headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: "12", }, data=json.dumps( { "type": workflow_job_type, "status": job_statuses.SUBMITTED, } ), ): create_workflow_response = handlers.setup_workflow(workflow_job_type) data = json.loads(create_workflow_response.get_data().decode()) workflow_function_name_mock.assert_called_once() assert create_workflow_response.status_code == HTTPStatus.OK assert data == expected_workflow_response def test_setup_workflow_approval_success(mocker: Any) -> None: """Test setup work flows.""" product_id = 12 product_vendor_id = 12321 orchard_user_id = "oa:345345" expected_workflow_response = [ { "context": { "correlation_id": "3163f53e-e2c9-11e8-865e-0242ac110002.1", "datetime": "2018-11-07T20:10:38Z", "id": 3750, "subaccount_id": None, "user_id": orchard_user_id, "vendor_id": None, }, "id": 5992, "inputs": None, "outputs": None, "parent_id": 5986, "status": "SUBMITTED", "type": "workflow_approval", } ] with app.test_request_context( headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: orchard_user_id, }, data=json.dumps( { "type": job_types.WORKFLOW_APPROVAL, "status": job_statuses.SUBMITTED, "context": {"product_id": product_id}, } ), ): mocker.patch.object( release, "get_by_id_with_vendor", autospec=True, return_value={"vendor_id": product_vendor_id}, ) setup_workflow_mock = mocker.patch.object( job_logic, "setup_workflow", autospec=True, return_value=expected_workflow_response, ) create_workflow_response = handlers.setup_workflow(job_types.WORKFLOW_APPROVAL) data = json.loads(create_workflow_response.get_data().decode()) setup_workflow_mock.assert_called_once() assert create_workflow_response.status_code == HTTPStatus.OK assert data == expected_workflow_response @pytest.mark.parametrize( ("test_description", "caller_headers", "expected_status", "dispatched"), [ ("internal service", {}, HTTPStatus.OK, True), ("oa admin", {header.ORCHARD_USER_ID: "oa:5"}, HTTPStatus.OK, True), ( "workstation vendor", {header.GRASS_ACCOUNT_TYPE: header.GRASS_ACCOUNT_TYPE_VENDOR}, HTTPStatus.FORBIDDEN, False, ), ( "workstation label", {header.ORCHARD_USER_ID: "alw:5"}, HTTPStatus.FORBIDDEN, False, ), ], ) def test_setup_workflow_ingestion_reencode_caller_gate( test_description: str, caller_headers: dict[str, str], expected_status: HTTPStatus, dispatched: bool, mocker: Any, ) -> None: """Only OA admins and internal services may launch the reencode workflow.""" request_data = { job_io_fields.WORKFLOW_INGEST_JOB_ID: 55, "overrides": {job_io_fields.VIDEO_INNER_HEIGHT_PIXELS: 1080}, "context": {"product_id": 12}, } setup_mock = mocker.patch.object( job_logic, "setup_ingestion_reencode_workflow", autospec=True, return_value=[], ) with app.test_request_context( headers={"Content-Type": "application/json", **caller_headers}, data=json.dumps(request_data), ): response = handlers.setup_workflow(job_types.WORKFLOW_INGESTION_REENCODE) assert response.status_code == expected_status if dispatched: setup_mock.assert_called_once_with(request_data) else: setup_mock.assert_not_called() def test_get_available_thumbnails(mocker: Any) -> None: """Test get available thumbnails endpoint.""" job_id = 1 thumbnails = [ "https://cdn.example.com/thumb1.jpg", "https://cdn.example.com/thumb2.jpg", ] get_available_thumbnails_mock = mocker.patch.object( thumbnails_logic, "get_available_thumbnails", return_value=thumbnails ) set_signed_cookies_mock = mocker.patch.object( cloudfront_logic, "set_signed_cookies", return_value=None ) with app.test_request_context(): result = handlers.get_available_thumbnails(job_id) get_available_thumbnails_mock.assert_called_once_with(job_id, image_size=None) set_signed_cookies_mock.assert_called_once() assert result.status_code == HTTPStatus.OK data = json.loads(result.get_data().decode()) assert data == {"items": thumbnails} def test_get_product_thumbnail(mocker: Any) -> None: """Test get product thumbnail endpoint.""" product_id = 1 get_product_thumbnail_mock = mocker.patch.object( thumbnails_logic, "get_product_thumbnail", return_value="https://example.com/thumb.jpg", ) with app.test_request_context(): result = handlers.get_product_thumbnail(product_id) get_product_thumbnail_mock.assert_called_once_with( product_id, image_size=None, expire_at=0 ) assert result.status_code == HTTPStatus.OK def test_generate_upload_token_for_custom_thumbnail(mocker: Any) -> None: """Test generate upload token endpoint.""" data = {"workflow_job_id": 123, "filename": "123.jpg"} generate_upload_token_for_custom_thumbnail_mock = mocker.patch.object( thumbnails_logic, "generate_upload_token_for_custom_thumbnail", return_value={}, ) with app.test_request_context( data=json.dumps(data), headers={"Content-Type": "application/json"} ): result = handlers.generate_upload_token_for_custom_thumbnail() generate_upload_token_for_custom_thumbnail_mock.assert_called_once_with(data) assert result.status_code == HTTPStatus.OK def test_bulk_ingest_metadata(mocker: Any) -> None: """Test bulk ingest metadata endpoint.""" data = [{"release_id": 1}] bulk_ingest_metadata_mock = mocker.patch.object( metadata_logic, "bulk_ingest_metadata", return_value={} ) with app.test_request_context( data=json.dumps(data), headers={ "Content-Type": "application/json", header.ORCHARD_USER_ID: "oa:1", }, ): bulk_ingest_metadata_response = handlers.bulk_ingest_metadata() bulk_ingest_metadata_mock.assert_called_once_with(data) assert bulk_ingest_metadata_response.status_code == HTTPStatus.OK def test_ingest(mocker: Any) -> None: """Test ingest endpoint.""" files = [ {"file_path": "s3://bucket/video1.mp4"}, {"file_path": "s3://bucket/video2.mp4"}, ] ingest_mock = mocker.patch.object(ingest_logic, "ingest", return_value=files) with app.test_request_context(json={}): result = handlers.ingest() ingest_mock.assert_called_once() assert result.status_code == HTTPStatus.OK data = json.loads(result.get_data().decode()) assert data == {"items": files} def test_get_cloud_transfer_progress(mocker: Any) -> None: """Test get cloud transfer progress.""" mocker.patch.object( job_logic, "get_cloud_transfer_progress", return_value={"abc": 123}, autospec=True, ) with app.test_request_context(): resp = handlers.get_cloud_transfer_progress(123, "asdf") assert json.loads(resp.get_data().decode()) == {"abc": 123} assert resp.status_code == HTTPStatus.OK def test_get_product_assets(mocker: Any) -> None: """Test get product assets endpoint.""" product_ids = [123, 3423] assets = [{"asset_id": 1, "product_id": 123}, {"asset_id": 2, "product_id": 3423}] get_product_assets_mock = mocker.patch.object( asset_logic, "get_product_assets", return_value=assets ) with app.test_request_context(path="/?product_ids=123,3423"): result = handlers.get_product_assets() get_product_assets_mock.assert_called_once_with(product_ids) assert result.status_code == HTTPStatus.OK data = json.loads(result.get_data().decode()) assert data == assets def test_get_product_assets_invalid_ids(mocker: Any) -> None: """Test get product assets returns 400 when product_ids are non-integer.""" mocker.patch.object(asset_logic, "get_product_assets") with app.test_request_context(path="/?product_ids=abc,123"): result = handlers.get_product_assets() assert result.status_code == HTTPStatus.BAD_REQUEST data = json.loads(result.get_data().decode()) assert data["code"] == error_constants.ERROR_CODE_VALIDATION_ERROR @patch("video.handlers.g") @patch("video.handlers.is_jwt_identity_authorized") @pytest.mark.parametrize( "query_params, expected_product_ids, expected_asset_types", [ ( {"product_ids": "123,3423,987754", "asset_types": "video_master,h264_hd"}, [123, 3423, 987754], ["VIDEO_MASTER", "H264_HD"], ), ( {"product_ids": "1", "asset_types": ""}, [1], [], ), ( {"product_ids": "1234,5678", "asset_types": "h264_SD"}, [1234, 5678], ["H264_SD"], ), ( {"product_ids": "123,3423,987754", "asset_types": "video_MASTER,H264_hd"}, [123, 3423, 987754], ["VIDEO_MASTER", "H264_HD"], ), ], ) def test_get_product_assets_bulk( is_jwt_identity_authorized_mock: Any, g_mock: Any, query_params: dict[str, Any], expected_product_ids: list[Any], expected_asset_types: list[Any], mocker: Any, app_context: Any, ) -> None: get_product_assets_mock = mocker.patch.object( asset_logic, "get_product_assets", return_value={} ) with app.test_request_context(query_string=urlencode(query_params)): g_mock.request_context.jwt_identity_id = "test_jwt" is_jwt_identity_authorized_mock.return_value = True result = handlers.get_video_asset_details_for_bulk_products() get_product_assets_mock.assert_called_once_with( expected_product_ids, expected_asset_types ) assert result.status_code == HTTPStatus.OK @patch("video.handlers.g") @patch("video.handlers.is_jwt_identity_authorized") @pytest.mark.parametrize( "query_params", [ ({"product_ids": "123,a,987754", "asset_types": "video_master,h264_hd"}), ({"product_ids": "a,3!,invalid", "asset_types": "video_master,h264_hd"}), ({"product_ids": "", "asset_types": "video_master"}), ({"asset_types": ""}), ({"product_ids": "1234,5678", "asset_types": "h111_SD"}), ({"product_ids": "", "asset_types": "video_MASTER,H264_hd"}), ({"product_ids": ",".join([str(x) for x in range(0, 186)])}), ({}), ], ) def test_get_product_assets_bulk_error( is_jwt_identity_authorized_mock: Any, g_mock: Any, query_params: dict[str, Any], mocker: Any, app_context: Any, ) -> None: get_product_assets_mock = mocker.patch.object( asset_logic, "get_product_assets", return_value={} ) with app.test_request_context(query_string=urlencode(query_params)): g_mock.request_context.jwt_identity_id = "test_jwt" is_jwt_identity_authorized_mock.return_value = True result = handlers.get_video_asset_details_for_bulk_products() get_product_assets_mock.assert_not_called() assert result.status_code == HTTPStatus.BAD_REQUEST @patch("video.handlers.g") @patch("video.handlers.is_jwt_identity_authorized") @pytest.mark.parametrize( "query_params, expected_product_ids, expected_asset_types", [ ( {"product_ids": "123,3423,987754", "asset_types": "video_master,h264_hd"}, [123, 3423, 987754], ["VIDEO_MASTER", "H264_HD"], ), ( {"product_ids": "1", "asset_types": ""}, [1], [], ), ( {"product_ids": "1234,5678", "asset_types": "h264_SD"}, [1234, 5678], ["H264_SD"], ), ( {"product_ids": "123,3423,987754", "asset_types": "video_MASTER,H264_hd"}, [123, 3423, 987754], ["VIDEO_MASTER", "H264_HD"], ), ], ) def test_get_product_assets_bulk_unauthorized_error( is_jwt_identity_authorized_mock: Any, g_mock: Any, query_params: dict[str, Any], expected_product_ids: list[Any], expected_asset_types: list[Any], mocker: Any, app_context: Any, ) -> None: get_product_assets_mock = mocker.patch.object( asset_logic, "get_product_assets", return_value={} ) with app.test_request_context(query_string=urlencode(query_params)): g_mock.request_context.jwt_identity_id = "test_jwt" is_jwt_identity_authorized_mock.return_value = False result = handlers.get_video_asset_details_for_bulk_products() get_product_assets_mock.assert_not_called() assert result.status_code == HTTPStatus.FORBIDDEN def test_upsert_video_asset(mocker: Any) -> None: """Test upsert video asset endpoint.""" new_video_asset = {"product_id": 1, "asset_type": "type", "asset_path": "path"} upsert_video_asset_mock = mocker.patch.object( video_asset_logic, "upsert_video_asset", return_value={} ) with app.test_request_context( data=json.dumps(new_video_asset), headers={"Content-Type": "application/json"}, path="/video-asset", ): result = handlers.upsert_video_asset() upsert_video_asset_mock.assert_called_once_with(new_video_asset) assert result.status_code == HTTPStatus.OK def test_bulk_upsert_video_asset(mocker: Any) -> None: """Test upsert video asset endpoint.""" new_video_asset = [ { "product_id": 1, "asset_type": "video_master", "tuid": 1, "asset_path": "/path/to/video1.mp4", }, { "product_id": 2, "asset_type": "H264_HD", "tuid": 2, "asset_path": "/path/to/video2.mp4", }, { "product_id": 3, "asset_type": "video_image_S5", "tuid": 3, "asset_path": "/path/to/image1.png", }, ] bulk_upsert_video_assets_mock = mocker.patch.object( video_asset_logic, "bulk_upsert_video_assets", return_value={} ) with app.test_request_context( data=json.dumps(new_video_asset), headers={"Content-Type": "application/json"}, path="/video-asset/bulk-ingest", ): result = handlers.bulk_upsert_video_asset() bulk_upsert_video_assets_mock.assert_called_once_with(new_video_asset) assert result.status_code == HTTPStatus.OK def test_create_bundle(mocker: Any) -> None: """Test create bundle endpoint.""" payload = { "audioProductId": 4763149, "bundleType": "AUDIO_VIDEO", "volumes": [["4763145", "4763147"], ["4763146", "4763148"]], "selectedThumbnailId": "4763147", } audio_product_id = payload["audioProductId"] bundle_type = payload["bundleType"] selected_thumbnail_id = payload["selectedThumbnailId"] volume_data = payload["volumes"] create_bundle_mock = mocker.patch.object( video_product_copy_logic, "create_bundle", return_value=audio_product_id, ) with app.test_request_context( data=json.dumps(payload), headers={"Content-Type": "application/json"}, path="/create_bundle", ): result = handlers.create_bundle() create_bundle_mock.assert_called_once_with( audio_product_id, bundle_type, selected_thumbnail_id, volume_data ) assert result.status_code == HTTPStatus.CREATED data = json.loads(result.get_data().decode()) assert data == {"product_id": audio_product_id} def test_migrate_artist(mocker: Any) -> None: """Migrate product artist.""" migrate_artist_mock = mocker.patch.object( product_logic, "migrate_artist", return_value={} ) data = {"artist_id": 12} with app.test_request_context( data=json.dumps(data), headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: 12, }, ): migrate_artist_response = handlers.migrate_artist(1234) migrate_artist_mock.assert_called_once_with(12, 1234, "vendor", "12") assert migrate_artist_response.status_code == HTTPStatus.OK def test_disassociate_tracks(mocker: Any) -> None: """Disassociate tracks from video products.""" disassociate_tracks_mock = mocker.patch.object( product_logic, "disassociate_tracks", return_value=None ) data = {"track_ids": [100, 200]} with app.test_request_context( data=json.dumps(data), headers={"Content-Type": "application/json"}, ): disassociate_tracks_response = handlers.disassociate_tracks() disassociate_tracks_mock.assert_called_once_with([100, 200]) assert disassociate_tracks_response.status_code == HTTPStatus.OK assert disassociate_tracks_response.get_data() == b"" def test_disassociate_tracks_invalid_payload(mocker: Any) -> None: """Reject a request with an empty track id list.""" disassociate_tracks_mock = mocker.patch.object(product_logic, "disassociate_tracks") data: dict[str, Any] = {"track_ids": []} with app.test_request_context( data=json.dumps(data), headers={"Content-Type": "application/json"}, ): with pytest.raises(ValidationError): handlers.disassociate_tracks() disassociate_tracks_mock.assert_not_called() def test_migrate_project(mocker: Any) -> None: """Migrate project of product.""" migrate_project_mock = mocker.patch.object( metadata_logic, "migrate_project", return_value={} ) data = {"project_id": 12} with app.test_request_context( data=json.dumps(data), headers={ "Content-Type": "application/json", header.GRASS_ACCOUNT_TYPE: "vendor", header.GRASS_ACCOUNT_ID: 12, }, ): migrate_project_response = handlers.migrate_project(1234) migrate_project_mock.assert_called_once_with(1234, 12, "vendor", "12") assert migrate_project_response.status_code == HTTPStatus.OK def test_get_all_video_types(mocker: Any) -> None: """Get All Track Types.""" get_all_video_types_mock = mocker.patch.object( video_type_logic, "get_all_video_types", return_value={} ) with app.test_request_context(): get_all_video_types_response = handlers.get_all_video_types() get_all_video_types_mock.assert_called_once() assert get_all_video_types_response.status_code == HTTPStatus.OK