from unittest.mock import Mock import numpy as np import pytest from pandas import NA, DataFrame, Series, isna from src.backend.constants import Flags, MiscLower, OutputColumns from src.backend.constants import SnowFlakeColumns as SFCols from src.backend.constants import YTReferenceStatus from src.backend.logic.audit import flags from src.backend.logic.audit.logic import common nans = (None, np.nan, float("nan")) empty = ("", " ", *nans) @pytest.fixture(autouse=True) def mocker_yt_checker(mocker): return mocker.patch("src.backend.logic.audit.flags.mv.YTChecks", autospec=True) class TestFlagChecksMV: _class = flags.FlagChecksMV mock_isrc = "1232432563236" def test_no_asset_id(self): """Inherited from _FlagChecks.""" assert self._class.no_asset_id is flags.base._FlagChecks.no_asset_id @pytest.mark.asyncio @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "ownership,is_ownership_valid,should_flag", [ (np.nan, True, True), # NaN ownership should flag always ({"US"}, True, False), # Complete ownership should not flag ({"US"}, False, True), # Incomplete ownership should flag ], ) async def test_ownership_incomplete( self, mocker, mocker_add_flag_db, mock_flagger, mock_df, has_asset_id, ownership, is_ownership_valid, should_flag, ): """Inherited from _FlagChecks.""" mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame([mv.loc[0].to_dict() | {SFCols.OWNERSHIP: ownership}]) instance = self._class(mv, mock_flagger) mocker.patch.object(instance, "_double_check_yt_cid_ownerships", autospec=True) mocker.patch( "src.backend.logic.audit.logic.common.is_ownership_valid", return_value=(is_ownership_valid, ["mock territories"]), ) await instance.ownership_incomplete() value = mv.loc[0][OutputColumns.AUDIT_FLAG_OWNERSHIP_INCOMPLETE] if has_asset_id: assert value == should_flag else: assert isna(value) if should_flag and has_asset_id: arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != should_flag assert arg1 == mock_flagger assert arg2 == Flags.OWNERSHIP_INCOMPLETE else: assert not mocker_add_flag_db.called @pytest.mark.asyncio @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "video_length,expected_auto_resolve", [(0, True), (np.nan, False), (None, False), (100, False)], ) @pytest.mark.parametrize( "active_reference_ids, should_flag", [ ("active reference id", False), ( { "active reference id", }, False, ), (set(), True), *[(_, True) for _ in empty], ], ) async def test_no_active_references( self, mocker_add_flag_db, mock_flagger, mock_df, mocker_yt_checker, has_asset_id, active_reference_ids, should_flag, video_length, expected_auto_resolve, ): """Inherited from _FlagChecks.""" # Mock coroutine mocker_yt_checker.return_value.get_asset_references.return_value = {} mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame( [ mv.loc[0].to_dict() | { SFCols.ACTIVE_REFERENCE_IDS: active_reference_ids, SFCols.VIDEO_LENGTH: video_length, } ] ) instance = self._class(mv, mock_flagger) await instance.no_active_references() value = mv.loc[0][OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES] if has_asset_id: assert value == should_flag else: assert isna(value) if should_flag and has_asset_id: arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != should_flag assert arg1 == mock_flagger assert arg2 == Flags.NO_ACTIVE_REFERENCES kwargs = mocker_add_flag_db.call_args.kwargs assert (kwargs["resolution"] is not None) == expected_auto_resolve assert (kwargs["resolution_subtype"] is not None) == expected_auto_resolve else: assert not mocker_add_flag_db.called @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "privacy_status", [ MiscLower.PUBLIC, "private", ], ) @pytest.mark.parametrize( "other_owners_claiming, should_flag", [ ("third party claim", True), ( { "claim", }, True, ), *[(_, False) for _ in empty], ], ) def test_third_party_claim( self, mocker_add_flag_db, mock_flagger, mock_df, has_asset_id, privacy_status, other_owners_claiming, should_flag, ): mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame( [ mv.loc[0].to_dict() | { SFCols.VIDEO_PRIVACY_STATUS: privacy_status, SFCols.OTHER_OWNERS_CLAIMING: other_owners_claiming, } ] ) instance = self._class(mv, mock_flagger) instance.third_party_claim() value = mv.loc[0][OutputColumns.AUDIT_FLAG_THIRD_PARTY_CLAIM] expect_flag = should_flag and privacy_status == MiscLower.PUBLIC if expect_flag: assert value is True arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != expect_flag assert arg1 == mock_flagger assert arg2 == Flags.THIRD_PARTY_CLAIM else: assert isna(value) assert not mocker_add_flag_db.called @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "privacy_status", [ MiscLower.PUBLIC, "private", ], ) @pytest.mark.parametrize( "other_owners_claiming, should_flag", [ ("third party claim", False), ( { "claim", }, False, ), *[(_, True) for _ in empty], ], ) def test_must_claim( self, mocker_add_flag_db, mock_flagger, mock_df, has_asset_id, privacy_status, other_owners_claiming, should_flag, ): mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame( [ mv.loc[0].to_dict() | { SFCols.VIDEO_PRIVACY_STATUS: privacy_status, SFCols.OTHER_OWNERS_CLAIMING: other_owners_claiming, } ] ) instance = self._class(mv, mock_flagger) instance.must_claim() value = mv.loc[0][OutputColumns.AUDIT_FLAG_MUST_CLAIM] expect_flag = ( should_flag and not has_asset_id and privacy_status == MiscLower.PUBLIC ) if expect_flag: assert value is True arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != expect_flag assert arg1 == mock_flagger assert arg2 == Flags.MUST_CLAIM else: assert isna(value) assert not mocker_add_flag_db.called @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "privacy_status", [ MiscLower.PUBLIC.lower(), MiscLower.PUBLIC.upper(), "private", ], ) @pytest.mark.parametrize("isrc", [mock_isrc, *empty]) def test_asset_missing_isrc( self, mocker_add_flag_db, mock_flagger, mock_df, has_asset_id, privacy_status, isrc, ): mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame( [mv.loc[0].to_dict() | {SFCols.VIDEO_PRIVACY_STATUS: privacy_status}] ) mv.loc[0, SFCols.ISRC] = isrc instance = self._class(mv, mock_flagger) instance.asset_missing_isrc() value = mv.loc[0][OutputColumns.AUDIT_FLAG_ASSET_MISSING_ISRC] expect_flag = ( has_asset_id and isrc != self.mock_isrc # Has no ISRC and privacy_status.lower() == MiscLower.PUBLIC.lower() ) if expect_flag: assert value is True arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != expect_flag assert arg1 == mock_flagger assert arg2 == Flags.ASSET_MISSING_ISRC else: assert isna(value) assert not mocker_add_flag_db.called @pytest.mark.asyncio @pytest.mark.parametrize( "has_asset_id", [True, False], ) @pytest.mark.parametrize( "match_policy, should_flag", [ *[(policy, False) for policy in common.VALID_POLICIES], ("another policy", True), *[(_, True) for _ in empty], ], ) async def test_bad_match_policy( self, mocker, mocker_add_flag_db, mock_flagger, mock_df, mocker_yt_checker, has_asset_id, match_policy, should_flag, ): mv = mock_df(has_asset_id=has_asset_id) mv = DataFrame([mv.loc[0].to_dict() | {SFCols.MATCH_POLICY: match_policy}]) # Omit match policy population from API mocker_yt_checker.return_value.get_asset_match_policies.return_value = {} instance = self._class(mv, mock_flagger) mocker.patch.object( instance, "_double_check_yt_cid_match_policies", autospec=True ) await instance.bad_match_policy() should_flag = should_flag and has_asset_id value = mv.loc[0][OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY] assert value == should_flag if should_flag: arg0, arg1, arg2 = mocker_add_flag_db.call_args.args assert arg0.empty != should_flag assert arg1 == mock_flagger assert arg2 == Flags.BAD_MATCH_POLICY else: assert not mocker_add_flag_db.called @pytest.mark.asyncio @pytest.mark.parametrize( "df", [ DataFrame([{SFCols.MATCH_POLICY: "mock policy", SFCols.ASSET_ID: np.nan}]), DataFrame([{SFCols.ASSET_ID: "123", SFCols.MATCH_POLICY: np.nan}]), DataFrame({SFCols.ASSET_ID: [], SFCols.MATCH_POLICY: []}), DataFrame( {SFCols.ASSET_ID: [np.nan], SFCols.MATCH_POLICY: ["mock policy"]} ), DataFrame({SFCols.ASSET_ID: ["123"], SFCols.MATCH_POLICY: [np.nan]}), DataFrame(), ], ) async def test_double_check_yt_cid_match_policies_no_rows_to_check(self, df): instance = self._class(df, Mock()) df_before = df.copy() assert await instance._double_check_yt_cid_match_policies(df) is None assert df.equals(df_before), "DataFrame should not be modified" @pytest.mark.asyncio async def test_double_check_yt_cid_match_policies_none_found(self, mocker): df = DataFrame([{SFCols.MATCH_POLICY: "mock policy", SFCols.ASSET_ID: "123"}]) instance = self._class(df, Mock()) df_before = df.copy() mock_get_asset_match_policies = instance._yt_checker.get_asset_match_policies mock_get_asset_match_policies.return_value = [] assert await instance._double_check_yt_cid_match_policies(df) is None assert df.equals(df_before), "DataFrame should not be modified" assert mock_get_asset_match_policies.called class TestBadMatchPolicy: _class = flags.mv._BadMatchPolicy _sample_asset_id = "123" _sample_bad_match_policy = "this match policy won't match the one in YT!" _sample_df = DataFrame( [ { SFCols.ASSET_ID: _sample_asset_id, SFCols.MATCH_POLICY: NA, } ] ) @pytest.fixture def instance(self): return self._class(self._sample_df.copy()) def test_get_asset_ids_missing_match_policies_empty_df(self, instance): instance.mv.drop(instance.mv.index, inplace=True) result = instance._get_asset_ids_missing_match_policies() assert result == [] def test_get_asset_ids_missing_match_policies_no_rows_missing_match_policies( self, instance ): instance.mv.at[0, SFCols.MATCH_POLICY] = "mock policy" result = instance._get_asset_ids_missing_match_policies() assert result == [] def test_get_asset_ids_missing_match_policies(self, instance): result = instance._get_asset_ids_missing_match_policies() assert result == [self._sample_asset_id] @pytest.mark.parametrize("has_asset_id", [True, False]) @pytest.mark.parametrize( "match_policy,expected", [ *((_, True) for _ in empty), # Empty match policy (_sample_bad_match_policy, True), # Bad match policy (list(common.VALID_POLICIES)[0], False), # Valid policy ], ) def test_flag_rows(self, instance, has_asset_id, match_policy, expected): instance.mv.at[0, SFCols.MATCH_POLICY] = match_policy instance.mv.at[0, SFCols.ASSET_ID] = ( self._sample_asset_id if has_asset_id else NA ) instance._flag_rows() result = instance.mv.loc[0][OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY] assert result == (expected if has_asset_id else False) class TestNoActiveReferences: _class = flags.mv._NoActiveReferences @pytest.fixture def instance(self): return self._class( DataFrame( { SFCols.ASSET_ID: ["123"], SFCols.ACTIVE_REFERENCE_IDS: ["active reference id"], SFCols.VIDEO_LENGTH: [100], } ), 1, ) @pytest.mark.parametrize( "_property", [ _class.rows_flagged, _class.rows_flagged_auto_resolvable, _class.rows_flagged_not_auto_resolvable, ], ) def test_properties_require_run(self, instance, _property): """Test that certain properties raise AttributeError if run has not been called. """ with pytest.raises(AttributeError): getattr(instance, _property.__name__) @pytest.mark.parametrize( "video_length_value, expected", [ (0, True), (np.nan, False), (None, False), (29, True), (30, False), ], ) def test_auto_resolvable_mask(self, mocker, instance, video_length_value, expected): instance.auto_resolve_length_lt = 30 mocker.patch.object( type(instance), "rows_flagged", new_callable=mocker.PropertyMock, return_value=DataFrame({SFCols.VIDEO_LENGTH: [video_length_value]}), ) mask = instance._auto_resolvable_mask result = mask.iloc[0] assert result == expected @pytest.mark.parametrize("flagged", [True, False]) def test_rows_flagged(self, instance, flagged): instance._has_been_run = True col = OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES instance.mv.at[0, col] = flagged assert len(instance.rows_flagged) == (1 if flagged else 0) @pytest.mark.parametrize( "mask_value, property_name", [ (True, "rows_flagged_auto_resolvable"), (False, "rows_flagged_not_auto_resolvable"), ], ) def test_rows_flagged_resolvable(self, mocker, instance, mask_value, property_name): row = {"Test": True} sample_df = DataFrame([row]) # Mock `rows_flagged` and `_auto_resolvable_mask`. mocker.patch.object( type(instance), "rows_flagged", return_value=sample_df, ) mocker.patch.object( type(instance), "_auto_resolvable_mask", return_value=Series([mask_value]), ) instance._has_been_run = True assert getattr(instance, property_name).equals(sample_df) @pytest.mark.parametrize( "active,inactive,expected_count", [ (Mock(), Mock(), 0), (Mock(), NA, 0), (NA, Mock(), 0), (NA, NA, 1), ], ) @pytest.mark.parametrize("use_none", [True, False]) # Test with None values as well def test_rows_no_references( self, instance, active, inactive, expected_count, use_none ): instance._has_been_run = True active = None if use_none and active is NA else active inactive = None if use_none and inactive is NA else inactive instance.mv.at[0, SFCols.ACTIVE_REFERENCE_IDS] = active instance.mv.at[0, SFCols.INACTIVE_REFERENCE_IDS] = inactive assert len(instance.rows_no_references) == expected_count @pytest.mark.asyncio async def test_run_sets_has_been_run_to_true(self, instance): await instance.run() assert instance._has_been_run is True @pytest.mark.asyncio async def test_run_flags_rows(self, instance, mocker): mocker_flag_rows = mocker.patch.object(instance, "_flag_rows", autospec=True) await instance.run() assert mocker_flag_rows.called @pytest.mark.asyncio async def test_populate_missing_references(self, instance, mocker): missing_references = [ {"asset_id": "123", "status": YTReferenceStatus.ACTIVE, "id": "id1"}, {"asset_id": "123", "status": YTReferenceStatus.INACTIVE, "id": "id2"}, {"asset_id": "123", "status": YTReferenceStatus.INACTIVE, "id": "id3"}, ] mocker.patch.object( instance, "_fetch_missing_references", autospec=True, return_value=missing_references, ) await instance._populate_missing_references() first_row = instance.mv.loc[0] assert first_row[SFCols.ACTIVE_REFERENCE_IDS] == {"id1"} assert first_row[SFCols.INACTIVE_REFERENCE_IDS] == {"id2", "id3"} @pytest.mark.asyncio async def test_fetch_missing_references(self, mocker): instance = self._class(DataFrame({})) mocker.patch.object( type(instance), "rows_no_references", return_value=DataFrame({SFCols.ASSET_ID: ["123"]}), new_callable=mocker.PropertyMock, ) mocker_get_asset_references = mocker.patch.object( instance._yt_checker, "get_asset_references", ) mocker_get_asset_references.return_value = {"123": ["reference"]} result = list(await instance._fetch_missing_references()) assert result == ["reference"] @pytest.mark.asyncio async def test_fetch_missing_references_nothing_to_fetch(self, mocker): instance = self._class(DataFrame({})) mocker.patch.object( type(instance), "rows_no_references", return_value=DataFrame({}), new_callable=mocker.PropertyMock, ) mocker_get_asset_references = mocker.patch.object( instance._yt_checker, "get_asset_references", ) assert list(await instance._fetch_missing_references()) == [] assert not mocker_get_asset_references.called @pytest.mark.parametrize( "active_reference_ids, expected", [ ("active reference id", False), ( { "active reference id", }, False, ), (set(), True), *[(_, True) for _ in empty], ], ) def test_flag_rows(self, instance, active_reference_ids, expected): instance.mv.at[0, SFCols.ACTIVE_REFERENCE_IDS] = active_reference_ids instance._flag_rows() result = instance.mv.loc[0][OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES] assert result == expected