"""Test queries.""" import concurrent.futures import copy from datetime import datetime from typing import Any from unittest.mock import MagicMock import pytest from flask import g from pytest_mock import MockerFixture from syrupy.assertion import SnapshotAssertion from video.api import app from video.constants import field_types, job_statuses, job_types from video.models.sql import queries from video.models.sql.classes.context import Context from video.models.sql.classes.context_field import ContextField from video.models.sql.classes.job import Job from video.models.sql.classes.job_input import JobInput from video.models.sql.classes.job_output import JobOutput from video.models.sql.classes.job_status import JobStatus from video.models.sql.classes.release import Release from video.models.sql.classes.track import Track from video.models.sql.classes.track_video import TrackVideo from video.models.sql.classes.video_asset import VideoAsset from video.models.sql.classes.video_dashboard_item import VideoDashboardItem from video.utils import typer from tests.unit import ar_db_utils, db_utils from tests.unit.factories.ows import video as video_job_requests job_a_vendor_id = 123 job_a_subaccount_id = 456 job_a_user_id = "alw:322" job_a_job_id = 111 job_b_job_id = 222 job_a_type = job_types.TRANSFER_FROM_BROWSER_TO_S3 job_b_type = job_types.TRANSFER_FROM_BROWSER_TO_S3 job_a_parent_id = 123 job_b_parent_id = 124 job_a_product_id = 234 job_b_product_id = 345 job_a_correlation_id = "c9f0409c-8d46-11e8-858c-acbc32b98ae7" job_b_correlation_id = "c8a90fb6-9fed-11e8-9e50-acbc32b98ae7" job_a_context_id = 456 job_b_context_id = 457 job_a = Job( primary_key=job_a_job_id, job_type=job_a_type, parent_id=job_a_parent_id, context_id=job_a_context_id, ) job_a_status = JobStatus( job_id=job_a_job_id, status=job_statuses.SUBMITTED, context_id=job_a_context_id, ) job_a_inputs = JobInput( job_id=job_a_job_id, name="applesauce", value=typer.stringify("bananas"), field_type=field_types.STRING, context_id=job_a_context_id, ) job_a_context = Context(primary_key=job_a_context_id) job_a_context_field_correlation_id = ContextField( context_id=job_a_context_id, name="correlation_id", value=typer.stringify(job_a_correlation_id), field_type=field_types.STRING, ) job_a_context_field_product_id = ContextField( context_id=job_a_context_id, name="product_id", value=typer.stringify(job_a_product_id), field_type=field_types.INTEGER, ) job_a_context_field_vendor_id = ContextField( context_id=job_a_context_id, name="vendor_id", value=typer.stringify(job_a_vendor_id), field_type=field_types.INTEGER, ) job_a_context_field_subaccount_id = ContextField( context_id=job_a_context_id, name="subaccount_id", value=typer.stringify(job_a_subaccount_id), field_type=field_types.INTEGER, ) job_a_context_field_user_id = ContextField( context_id=job_a_context_id, name="user_id", value=typer.stringify(job_a_user_id), field_type=field_types.STRING, ) job_b = Job( primary_key=job_b_job_id, job_type=job_b_type, parent_id=job_b_parent_id, context_id=job_b_context_id, ) job_b_status = JobStatus( job_id=job_b_job_id, status=job_statuses.SUBMITTED, context_id=job_b_context_id, ) job_b_inputs = JobInput( job_id=job_b_job_id, name="applesauce", value=typer.stringify("bananas"), field_type=field_types.STRING, context_id=job_b_context_id, ) job_b_outputs = JobOutput( job_id=job_b_job_id, name="potato", value=typer.stringify("tomato"), field_type=field_types.STRING, context_id=job_b_context_id, ) job_b_context = Context(primary_key=job_b_context_id) job_b_context_field_correlation_id = ContextField( context_id=job_b_context_id, name="correlation_id", value=typer.stringify(job_b_correlation_id), field_type=field_types.STRING, ) job_b_context_field_product_id = ContextField( context_id=job_b_context_id, name="product_id", value=typer.stringify(job_b_product_id), field_type=field_types.INTEGER, ) basic_model_set = [ job_a_context, job_b_context, job_a_context_field_correlation_id, job_b_context_field_correlation_id, job_a_context_field_product_id, job_b_context_field_product_id, job_a_context_field_vendor_id, job_a_context_field_subaccount_id, job_a_context_field_user_id, job_a, job_b, job_a_status, job_b_status, job_a_inputs, job_b_inputs, job_b_outputs, ] job_a_expected_response = { "type": job_types.TRANSFER_FROM_BROWSER_TO_S3, "context": { "datetime": "{{REPLACE_ME}}", "correlation_id": job_a_correlation_id, "product_id": job_a_product_id, "id": job_a_context_id, "vendor_id": job_a_vendor_id, "subaccount_id": job_a_subaccount_id, "user_id": job_a_user_id, }, "outputs": None, "parent_id": job_a_parent_id, "status": "SUBMITTED", "inputs": {"applesauce": "bananas"}, "id": job_a_job_id, } job_b_expected_response = { "type": job_types.TRANSFER_FROM_BROWSER_TO_S3, "context": { "datetime": "{{REPLACE_ME}}", "correlation_id": job_b_correlation_id, "product_id": job_b_product_id, "id": job_b_context_id, }, "outputs": {"potato": "tomato"}, "parent_id": job_b_parent_id, "status": "SUBMITTED", "inputs": {"applesauce": "bananas"}, "id": job_b_job_id, } video_dashboard_item_a = VideoDashboardItem( dashboard_item_id=123, asset_id=12345, upc=123456789, ) video_dashboard_item_b = VideoDashboardItem( dashboard_item_id=124, asset_id=12346, upc=123456788, ) video_dashboard_item_c = VideoDashboardItem( dashboard_item_id=125, asset_id=12347, upc=123456787, ) releases_a = Release( release_id=54321, upc=123456789, project_id=456, ) releases_b = Release( release_id=54322, upc=123456788, project_id=456, ) releases_c = Release( release_id=54323, upc=123456787, project_id=456, ) video_dashboard_and_releases_model_set = [ video_dashboard_item_a, video_dashboard_item_b, video_dashboard_item_c, releases_a, releases_b, releases_c, ] video_dashboard_item_a_expected_response = [ { "dashboard_item_id": 124, "asset_id": 12346, "product_id": 54322, }, { "dashboard_item_id": 125, "asset_id": 12347, "product_id": 54323, }, ] video_dashboard_item_b_expected_response = [ { "dashboard_item_id": 125, "asset_id": 12347, "product_id": 54323, } ] video_dashboard_item_c_expected_response: list[dict[str, Any]] = [] video_assets = [ VideoAsset( product_id=123, asset_type="video_master", asset_path="video_master_s3_path", last_updated=datetime.strptime("2024-12-09 10:03:55", "%Y-%m-%d %H:%M:%S"), duration=30.733, ), VideoAsset( product_id=234, asset_type="video_image_S10", asset_path="video_image_S10_s3_path", last_updated=datetime.strptime("2024-12-09 10:03:55", "%Y-%m-%d %H:%M:%S"), ), ] video_tracks_metadata = [ Track( track_id=1, product_id=123, upc=123, volume_number=1, track_number=1, duration_minutes=1, duration_seconds=0, closed_caption_exists=0, closed_caption_reason_id=1, ), TrackVideo( track_id=1, color="color", channel="2", fps="30", resolution="1080HD", aspect_ratio="16:9", ), Track( track_id=2, product_id=1234, upc=1234, volume_number=1, track_number=1, duration_minutes=2, duration_seconds=0, closed_caption_exists=0, closed_caption_reason_id=1, ), TrackVideo( track_id=2, color="color", channel="2", fps="25", resolution="1080HD", aspect_ratio="16:9", ), ] def _fake_submit(func: Any, **kwargs: Any) -> Any: mock = MagicMock() mock.result.return_value = func(**kwargs) return mock @pytest.mark.parametrize( ( "existing_job", "models_to_seed", "job_request", ), [ pytest.param( {}, copy.deepcopy([job_a_context, job_a_context_field_product_id]), video_job_requests.VideoJobCreateFactory.build( job_parent_id=None, context={"product_id": job_a_product_id}, ).to_dict(), id="test create", ), pytest.param( job_a_expected_response, copy.deepcopy(basic_model_set), video_job_requests.VideoJobChangeStatusFactory.build( job_id=job_a_job_id, job_status=job_statuses.PROGRESSING, ).to_dict(), id="test change status", ), pytest.param( job_a_expected_response, copy.deepcopy(basic_model_set), video_job_requests.VideoJobAddInputsFactory.build( job_id=job_a_job_id, job_type=job_a_type ).to_dict(), id="test add inputs", ), pytest.param( job_a_expected_response, copy.deepcopy(basic_model_set), video_job_requests.VideoJobAddOutputsFactory.build( job_id=job_a_job_id, job_type=job_a_type ).to_dict(), id="test add outputs", ), ], ) @db_utils.test_schema def test_persist_job_data( existing_job: Any, models_to_seed: Any, job_request: Any, mocker: MockerFixture, ) -> None: """Test persist_job_data.""" db_utils.seed_models(models_to_seed) mock_executor: MagicMock = mocker.patch.object( concurrent.futures, "ThreadPoolExecutor", autospec=True ) mock_executor.return_value.__enter__.return_value.submit = _fake_submit with app.test_request_context(): g.context = { "id": job_a_context_id, } job_ids = queries.persist_job_data(job_request) get_jobs_response = queries.get_jobs({"id": job_ids[0]})[0] expected_response = { "id": existing_job.get("id", get_jobs_response.get("id")), "parent_id": existing_job.get("parent_id", job_request.get("parent_id")), "type": existing_job.get("type", job_request.get("type")), "status": job_request.get("status", existing_job.get("status")), "context": existing_job.get("context", get_jobs_response.get("context")), "inputs": dict( existing_job.get("inputs") or {}, **(job_request.get("inputs") or {}) ) or None, "outputs": dict( existing_job.get("outputs") or {}, **(job_request.get("outputs") or {}) ) or None, } expected_response["context"]["datetime"] = get_jobs_response["context"]["datetime"] assert expected_response == get_jobs_response @pytest.mark.parametrize( ( "job_data_filters", "context", "models_to_seed", "expected_response", ), [ pytest.param( {"id": job_a_job_id}, { "vendor_id": job_a_vendor_id, "subaccount_id": job_a_subaccount_id, }, copy.deepcopy(basic_model_set), [job_a_expected_response], id="search by job_id a", ), pytest.param( {"id": job_b_job_id}, {}, copy.deepcopy(basic_model_set), [job_b_expected_response], id="search by job_id b", ), pytest.param( {}, {}, copy.deepcopy(basic_model_set), [job_a_expected_response, job_b_expected_response], id="search by nothing", ), pytest.param( {"parent_id": job_a_parent_id}, {}, copy.deepcopy(basic_model_set), [job_a_expected_response], id="search by parent_id", ), pytest.param( {"job_type": job_types.TRANSFER_FROM_BROWSER_TO_S3}, {}, copy.deepcopy(basic_model_set), [job_a_expected_response, job_b_expected_response], id="search by job_type", ), ], ) @db_utils.test_schema def test_get_job( job_data_filters: Any, context: Any, models_to_seed: Any, expected_response: Any, mocker: MockerFixture, ) -> None: """Test get job.""" mock_executor: MagicMock = mocker.patch.object( concurrent.futures, "ThreadPoolExecutor", autospec=True ) mock_executor.return_value.__enter__.return_value.submit = _fake_submit db_utils.seed_models(models_to_seed) with app.test_request_context(): g.context = context resp = queries.get_jobs(job_data_filters) for exp_ele, resp_ele in zip(expected_response, resp, strict=True): exp_ele["context"]["datetime"] = resp_ele["context"]["datetime"] assert expected_response == resp @db_utils.test_schema def test_persist_context() -> None: """Test persist context.""" with app.test_request_context(): context_id = queries.persist_context( { "whateves": "stuff", } ) queries.persist_context( { "id": context_id, "whateves": "stuffff", "applesauce": "bananas", } ) queries.persist_context( { "id": context_id, "whateves": "stufffffffff", } ) queries.persist_context( { "whateves": "stuf", } ) with app.test_request_context(): g.context = {"id": context_id} job_ids = queries.persist_job_data({"type": "test"}) with app.test_request_context(): get_context_response = queries.get_context(job_id=job_ids[0]) assert get_context_response == { "id": context_id, "applesauce": "bananas", "whateves": "stufffffffff", "datetime": get_context_response["datetime"], } @pytest.mark.parametrize( ( "dashboard_item_id", "offset", "models_to_seed", "expected_response", ), [ pytest.param( 123, 0, copy.deepcopy(video_dashboard_and_releases_model_set), video_dashboard_item_a_expected_response, id="test get new assets two results", ), pytest.param( 124, 0, copy.deepcopy(video_dashboard_and_releases_model_set), video_dashboard_item_b_expected_response, id="test get new assets one result", ), pytest.param( 125, 0, copy.deepcopy(video_dashboard_and_releases_model_set), video_dashboard_item_c_expected_response, id="test get new assets no result", ), ], ) @ar_db_utils.test_schema def test_get_new_assets( dashboard_item_id: Any, offset: Any, models_to_seed: Any, expected_response: Any, ) -> None: """Test get new assets.""" ar_db_utils.seed_models(models_to_seed) response = queries.get_new_assets(dashboard_item_id, offset) assert expected_response == response @pytest.mark.parametrize( ( "dashboard_item_id", "models_to_seed", "expected_response", ), [ pytest.param( 123, copy.deepcopy(video_dashboard_and_releases_model_set), 2, id="test get new assets two results", ), pytest.param( 124, copy.deepcopy(video_dashboard_and_releases_model_set), 1, id="test get new assets one result", ), pytest.param( 125, copy.deepcopy(video_dashboard_and_releases_model_set), 0, id="test get new assets no result", ), ], ) @ar_db_utils.test_schema def test_get_new_assets_count( dashboard_item_id: Any, models_to_seed: Any, expected_response: Any, ) -> None: """Test get new assets count query.""" ar_db_utils.seed_models(models_to_seed) response = queries.get_new_assets_count(dashboard_item_id) assert expected_response == response @ar_db_utils.test_schema def test_get_max_dashboard_item_id() -> None: """Test get max dashboard item id.""" models_to_seed = copy.deepcopy(video_dashboard_and_releases_model_set) ar_db_utils.seed_models(models_to_seed) response = queries.get_max_dashboard_item_id() assert response == 125 @db_utils.test_schema def test_get_product_assets(snapshot: SnapshotAssertion) -> None: """Test get_product_assets.""" db_utils.seed_models(video_assets) assert queries.get_product_assets([123, 234]) == snapshot @ar_db_utils.test_schema def test_get_video_tracks_metadata(snapshot: SnapshotAssertion) -> None: """Test get_video_tracks_metadata.""" ar_db_utils.seed_models(video_tracks_metadata) assert queries.get_video_tracks_metadata([1, 2]) == snapshot