from unittest.mock import Mock import pytest from pandas import NA from src.backend.logic.audit import flags from src.backend.logic.audit.flags.mv import Df, SFCols from tests_backend.conftest import TEST_YT_ASSET_ID class TestFlagChecksMV: _class = flags.FlagChecksMV @pytest.fixture def instance(self, sample_df): return self._class(sample_df, Mock()) @pytest.fixture def sample_df(self): return Df([{SFCols.ASSET_ID: TEST_YT_ASSET_ID}]) @pytest.mark.asyncio async def test_double_check_yt_cid_ownerships(self, instance, sample_df): # These territories are in YT for the provided Asset ID as of # the test creation date. If this test fails, check if they have changed. sample_df[SFCols.OWNERSHIP] = [None] # Initialize the column with None sample_df.at[0, SFCols.OWNERSHIP] = { "AD", "AE", "AF", "AG", "AI", "AL", "AM", "AO", "AQ", "AR", "AS", "AT", "AU", "AW", "AX", "AZ", "BA", "BB", "BD", "BE", "BF", "BG", "BH", "BI", "BJ", "BL", "BM", "BN", "BO", "BQ", "BR", "BS", "BT", "BV", "BW", "BY", "BZ", "CA", "CC", "CD", "CF", "CG", "CH", "CI", "CK", "CL", "CM", "CN", "CO", "CR", "CU", "CV", "CW", "CX", "CY", "CZ", "DE", "DJ", "DK", "DM", "DO", "DZ", "EC", "EE", "EG", "EH", "ER", "ES", "ET", "FI", "FJ", "FK", "FM", "FO", "FR", "GA", "GB", "GD", "GE", "GF", "GG", "GH", "GI", "GL", "GM", "GN", "GP", "GQ", "GR", "GS", "GT", "GU", "GW", "GY", "HK", "HM", "HN", "HR", "HT", "HU", "ID", "IE", "IL", "IM", "IN", "IO", "IQ", "IR", "IS", "IT", "JE", "JM", "JO", "JP", "KE", "KG", "KH", "KI", "KM", "KN", "KP", "KR", "KW", "KY", "KZ", "LA", "LB", "LC", "LI", "LK", "LR", "LS", "LT", "LU", "LV", "LY", "MA", "MC", "MD", "ME", "MF", "MG", "MH", "MK", "ML", "MM", "MN", "MO", "MP", "MQ", "MR", "MS", "MT", "MU", "MV", "MW", "MX", "MY", "MZ", "NA", "NC", "NE", "NF", "NG", "NI", "NL", "NO", "NP", "NR", "NU", "NZ", "OM", "PA", "PE", "PF", "PG", "PH", "PK", "PL", "PM", "PN", "PR", "PS", "PT", "PW", "PY", "QA", "RE", "RO", "RS", "RU", "RW", "SA", "SB", "SC", "SD", "SE", "SG", "SH", "SI", "SJ", "SK", "SL", "SM", "SN", "SO", "SR", "SS", "ST", "SV", "SX", "SY", "SZ", "TC", "TD", "TF", "TG", "TH", "TJ", "TK", "TL", "TM", "TN", "TO", "TR", "TT", "TV", "TW", "TZ", "UA", "UG", "UM", "US", "UY", "UZ", "VA", "VC", "VE", "VG", "VI", "VN", "VU", "WF", "WS", "YE", "YT", "ZA", "ZM", "ZW", } await instance._double_check_yt_cid_ownerships(sample_df) assert len(sample_df) == 0, ( "Provided sample Asset ID expected to be false positive. If this " "test fails, check whether the provided Asset ID is still a false " "positive." ) @pytest.mark.asyncio async def test_double_check_yt_cid_ownerships_no_flagged_rows( self, instance, sample_df ): initial_df = sample_df.copy() await instance._double_check_yt_cid_ownerships(Df()) assert sample_df.equals(initial_df) @pytest.mark.asyncio async def test_double_check_yt_cid_match_policies(self, instance, sample_df): sample_df.loc[0, SFCols.MATCH_POLICY] = ( "this match policy won't match the one in YT!" ) await instance._double_check_yt_cid_match_policies(sample_df) assert sample_df.empty, ( "Provided sample Asset ID expected to be false positive. If this " "test fails, check whether the provided Asset ID is still a false " "positive." ) class TestBadMatchPolicy: _class = flags.mv._BadMatchPolicy _sample_df = Df( [ { SFCols.ASSET_ID: TEST_YT_ASSET_ID, SFCols.MATCH_POLICY: NA, } ] ) @pytest.fixture def instance(self): return self._class(self._sample_df.copy()) @pytest.mark.asyncio async def test_populate_missing_match_policies(self, instance): await instance._populate_missing_match_policies() match_policy = instance.mv.at[0, SFCols.MATCH_POLICY] assert isinstance(match_policy, str) and bool( match_policy ), "Expected match policy to be populated (as a string)" class TestNoActiveReferences: _class = flags.mv._NoActiveReferences _sample_df = Df( [ { SFCols.ASSET_ID: TEST_YT_ASSET_ID, SFCols.ACTIVE_REFERENCE_IDS: NA, SFCols.INACTIVE_REFERENCE_IDS: NA, } ] ) @pytest.fixture def instance(self): return self._class(self._sample_df.copy(), 30) def test_status_cols_are_defined(self, instance): assert SFCols.ACTIVE_REFERENCE_IDS in instance.mv.columns assert SFCols.INACTIVE_REFERENCE_IDS in instance.mv.columns @pytest.mark.asyncio async def test_populate_missing_references(self, instance): instance._has_been_run = True await instance._populate_missing_references() assert isinstance( instance.mv.at[0, SFCols.ACTIVE_REFERENCE_IDS], set ), "Expected active references to be populated (as a set)" @pytest.mark.asyncio async def test_fetch_missing_references(self, instance): instance._has_been_run = True result = list(await instance._fetch_missing_references()) assert result, "Expected references to be fetched for the asset" assert all(ref.asset_id == TEST_YT_ASSET_ID for ref in result)