Untitled

"""testing common_utils"""
import os
import sys
import unittest
from unittest.mock import patch, MagicMock

import pandas as pd

sys.path.append("../../src")
sys.modules["awswrangler"] = MagicMock()
sys.modules["data_reservoir"] = MagicMock()
sys.modules["data_reservoir.common_services"] = MagicMock()
sys.modules["data_reservoir.master_data"] = MagicMock()
sys.modules["data_reservoir.s3_services"] = MagicMock()
sys.modules["requests"] = MagicMock()
os.environ["env"] = "dev"
os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
os.environ['reservoir_bucket'] = "test_bucket"
os.environ["to_emails"] = "test_to_emails"
os.environ["signature"] = "test_signature"
os.environ["cc_emails"] = "test_cc_emails"
from src import common_utils


class Response:
    def __init__(self):
        self.status_code=200


class LambdaTest(unittest.TestCase):
    """Class for passing lambda test portion of pipeline."""
    def test_get_ffsdm_securities(self) -> None:
        """Testing get_ffsdm_securities"""
        patch('src.common_utils.os').start()
        mock_smf_df = patch('src.common_utils.get_securities_master').start()
        mock_smf_df.side_effect = [
            pd.DataFrame([{
                "CADIS_ID": "12345",
                "SecurityID": "HS1MYR",
                "MajorIssueGroupDesc": "Equities"
            }]),
            pd.DataFrame([{
                "UserDefFieldName": "ETF",
                "UserDefFieldValue": "Yes",
                "SecurityID": "HS1MYR"
            }])
        ]
        mock_response = common_utils.get_ffsdm_securities()
        self.assertEqual(
            list(mock_response[0].columns),
            ['CADIS_ID', 'SecurityID', 'MajorIssueGroupDesc']
        )
        patch.stopall()

    
    def test_write_invmnt_trxn_data(self):
        """Test write_invmnt_trxn_data"""
        mock_common_utils_patcher = patch('src.common_utils.requests.post').start()
        mock_common_utils_patcher.return_value = Response()
        mock_response = common_utils.write_invmnt_trxn_data({"body": ["data"]}, "base_url", "token")
        self.assertEqual(200, mock_response.status_code)
        patch.stopall()

    def test_get_src_with_master_data(self):
        input_df = pd.DataFrame({
            "securityid": ["HS1MYR", "securityid"],
            "portfolio": ["AMOY", "AMOY"],
            "brokercode": ["Code1", "code1"]
        })
        mock_smf_df1 = patch("src.common_utils.get_securities_master").start()
        mock_smf_df1.return_value = pd.DataFrame([{
                "CADIS_ID": "12345",
                "SecurityID": "HS1MYR",
                "MajorIssueGroupDesc": "OtherInvestments"
            }])
        mock_smf_udf_df = patch("data_reservoir.common_services.read_s3_csv").start()
        mock_smf_udf_df.return_value = pd.DataFrame([{
                "UserDefFieldName": "ETF",
                "UserDefFieldValue": "Yes",
                "SecurityID": "HS1MYR"
            }])

        mock_pmf_df = patch("src.common_utils.get_portfolio_master").start()
        mock_pmf_df.return_value = pd.DataFrame({
            "PortfolioCode": ["Code","Code1"],
            "PortfolioMasterId": ["1", "2"],


        })
        mock_broker_df = patch("src.common_utils.get_borkercode_master").start()
        mock_broker_df.return_value = pd.DataFrame({
            "MappingValue": ["value","value1"],
            "BrokerEsdlId": ["Id1","Id1"],
            "EnterpriseSystemName": ["ePAM SE ASIA", "ePAM SE ASIA1"]
        })
        patch("src.common_utils.notify_missing_securities").start()
        patch("src.common_utils.notify_missing_portfolios").start()
        patch("src.common_utils.notify_missing_brokercodes").start()
        event_json = {
            "as_of_date": "2024-07-22",
            "yyyy": "2024",
            "mm": "06",
            "dd": "19",
            "schema_name": "esdl_ffsdm_trade",
            "table_name": "esdl_ffsdm_trade",
            "database_name": "pgi_reservoir_my_fa_epam",
            "raise_missing_entries": "No"
        }
        common_utils.get_src_with_master_data(input_df, event_json)