from enum import StrEnum import numpy as np import pytest from pandas import DataFrame from src.backend.constants import MiscLower, OutputColumns from src.backend.constants import SnowFlakeColumns as SFCols from src.backend.logic.audit.flags import conditions class TestHasNonEmptyStr: sample_col = "sample_col" class MockStrEnum(StrEnum): MOCK = "mock" class MockStrEnumEmpty(StrEnum): MOCK = "" @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), (set(), False), (MockStrEnum.MOCK, True), (MockStrEnumEmpty.MOCK, False), ], ) def test_has_non_empty_str(self, value, expected): df = DataFrame({self.sample_col: [value]}) df = conditions.has_non_empty_str(df, self.sample_col) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), ], ) def test_has_isrc(value, expected): df = DataFrame({SFCols.ISRC: [value]}) df = conditions.has_isrc(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), ], ) def test_has_asset_id(value, expected): df = DataFrame({SFCols.ASSET_ID: [value]}) df = conditions.has_asset_id(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), (set(), False), ([], False), ([""], False), (["a"], True), (["1"], True), ([" a1"], True), ({""}, False), ({" "}, False), ({"a"}, True), ({"1"}, True), ({" a1"}, True), ], ) def test_has_other_owners_claiming(value, expected): df = DataFrame({SFCols.OTHER_OWNERS_CLAIMING: [value]}) df = conditions.has_other_owners_claiming(df) assert df[0] == expected class TestHasThirdPartyClaim: @pytest.mark.parametrize( "other_claimants,status,expected", [ ({"other_claimant"}, MiscLower.PUBLIC, True), ({"other_claimant"}, MiscLower.PRIVATE, False), ({"other_claimant"}, MiscLower.UNLISTED, False), (set(), MiscLower.PUBLIC, False), (set(), MiscLower.PRIVATE, False), (set(), MiscLower.UNLISTED, False), (None, MiscLower.PUBLIC, False), (np.nan, MiscLower.PUBLIC, False), ({"other_claimant"}, np.nan, False), ], ) def test_has_third_party_claim(self, other_claimants, status, expected): df = DataFrame( { SFCols.OTHER_OWNERS_CLAIMING: [other_claimants], SFCols.VIDEO_PRIVACY_STATUS: [status], } ) df = conditions.has_third_party_claim(df) assert df[0] == expected @pytest.mark.parametrize("asset_id", [None, np.nan, ""]) @pytest.mark.parametrize( "other_claimants,status,expected", [ ({"other_claimant"}, MiscLower.PUBLIC, False), ({"other_claimant"}, MiscLower.PRIVATE, False), ({"other_claimant"}, MiscLower.UNLISTED, False), (set(), MiscLower.PUBLIC, True), (set(), MiscLower.PRIVATE, False), (set(), MiscLower.UNLISTED, False), (None, MiscLower.PUBLIC, True), (np.nan, MiscLower.PUBLIC, True), ({"other_claimant"}, np.nan, False), ], ) def test_can_claim(self, other_claimants, status, expected, asset_id): df = DataFrame( { SFCols.ASSET_ID: [asset_id], SFCols.OTHER_OWNERS_CLAIMING: [other_claimants], SFCols.VIDEO_PRIVACY_STATUS: [status], } ) df = conditions.has_third_party_claim(df, other_claimants=False) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), ], ) def test_has_ownership(value, expected): df = DataFrame({SFCols.OWNERSHIP: [value]}) df = conditions.has_ownership(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (None, False), (np.nan, False), ("", False), (" ", False), ("a", True), ("1", True), (" a1", True), ], ) def test_has_match_policy(value, expected): df = DataFrame({SFCols.MATCH_POLICY: [value]}) df = conditions.has_match_policy(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (True, True), (False, False), (None, False), (np.nan, False), ("", False), (" ", False), ("a", False), ("1", False), (" a1", False), ], ) def test_has_asset_missing_isrc(value, expected): df = DataFrame({OutputColumns.AUDIT_FLAG_ASSET_MISSING_ISRC: [value]}) df = conditions.has_asset_missing_isrc(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ (True, True), (False, False), (None, False), (np.nan, False), ("", False), (" ", False), ("a", False), ("1", False), (" a1", False), ], ) def test_has_ownership_incomplete(value, expected): df = DataFrame({OutputColumns.AUDIT_FLAG_OWNERSHIP_INCOMPLETE: [value]}) df = conditions.has_ownership_incomplete(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ ("other", False), ("0", False), ("", False), (" ", False), (MiscLower.PUBLIC, True), ], ) def test_has_privacy_status_public(value, expected): df = DataFrame({SFCols.VIDEO_PRIVACY_STATUS: [value]}) df = conditions.has_privacy_status_public(df) assert df[0] == expected @pytest.mark.parametrize( "value,expected", [ ("topic channel", True), ("0", True), ("", False), (" ", False), (123, False), (None, False), (np.nan, False), ], ) def test_has_topic_channel(value, expected): df = DataFrame({SFCols.CHANNEL_DISPLAY_NAME: [value]}) df = conditions.has_topic_channel(df) assert df[0] == expected