#!/usr/bin/env python3 """ utils/meta_ads_client.py 테스트 스위트 테스트 대상: MetaAdsClient 클래스 전략: 모든 외부 API 호출(facebook_business SDK, requests)을 mock으로 대체 """ import os import sys import tempfile from pathlib import Path from unittest.mock import MagicMock, patch import pytest # sys.path 설정: workspace 루트를 추가하여 utils 패키지 import 가능하게 함 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) # ────────────────────────────────────────────────────────────────────────────── # Fixtures # ────────────────────────────────────────────────────────────────────────────── @pytest.fixture def mock_env(): """4개의 필수 환경변수를 세팅하는 fixture.""" env = { "META_APP_ID": "test_app_id", "META_APP_SECRET": "test_app_secret", "META_ACCESS_TOKEN": "test_token", "META_AD_ACCOUNT_ID": "act_123456", } with patch.dict(os.environ, env, clear=False): yield env @pytest.fixture def mock_api_init(): """FacebookAdsApi.init을 mock으로 대체하는 fixture.""" with patch("utils.meta_ads_client.FacebookAdsApi.init") as mock_init: yield mock_init @pytest.fixture def mock_ad_account(): """AdAccount 클래스를 mock으로 대체하는 fixture.""" with patch("utils.meta_ads_client.AdAccount") as mock_cls: mock_instance = MagicMock() mock_cls.return_value = mock_instance yield mock_cls, mock_instance @pytest.fixture def client(mock_env, mock_api_init, mock_ad_account): """ MetaAdsClient 인스턴스를 반환하는 통합 fixture. load_env_keys와 FacebookAdsApi.init, AdAccount를 모두 mock 처리한다. """ with patch("utils.meta_ads_client.load_env_keys"): from utils.meta_ads_client import MetaAdsClient instance = MetaAdsClient() # 테스트에서 편리하게 쓸 수 있도록 mock_ad_account의 인스턴스를 교체 _, mock_ad_account_instance = mock_ad_account instance._account = mock_ad_account_instance return instance # ────────────────────────────────────────────────────────────────────────────── # 1. test_init_success # ────────────────────────────────────────────────────────────────────────────── class TestInit: """MetaAdsClient.__init__() 테스트""" def test_init_success(self, mock_env, mock_api_init): """환경변수 4개 모두 설정 시 정상 초기화되고 FacebookAdsApi.init이 호출된다.""" with ( patch("utils.meta_ads_client.load_env_keys"), patch("utils.meta_ads_client.AdAccount") as mock_ad_account_cls, ): from utils.meta_ads_client import MetaAdsClient client = MetaAdsClient() # FacebookAdsApi.init이 한 번 호출됨 mock_api_init.assert_called_once() init_kwargs = mock_api_init.call_args # app_id, app_secret, access_token이 올바르게 전달됨 args, kwargs = init_kwargs all_args = list(args) + list(kwargs.values()) assert "test_app_id" in all_args or kwargs.get("app_id") == "test_app_id" or args[0] == "test_app_id" # AdAccount가 생성됨 mock_ad_account_cls.assert_called_once() call_arg = mock_ad_account_cls.call_args[0][0] assert "act_123456" in call_arg def test_init_missing_env(self): """필수 환경변수 누락 시 ValueError가 발생한다.""" # 4개 환경변수를 모두 비워서 테스트 with ( patch.dict(os.environ, {}, clear=True), patch("utils.meta_ads_client.load_env_keys"), patch("utils.meta_ads_client.FacebookAdsApi"), ): from utils.meta_ads_client import MetaAdsClient with pytest.raises((ValueError, KeyError)): MetaAdsClient() # ────────────────────────────────────────────────────────────────────────────── # 3. test_exchange_token (구 exchange_long_lived_token) # ────────────────────────────────────────────────────────────────────────────── class TestExchangeLongLivedToken: """exchange_token() 테스트""" def test_exchange_long_lived_token(self, client, mock_env, mock_api_init): """ requests.get을 mock하여 새 장기 토큰을 받고, 반환값이 토큰 문자열인지 확인한다. FacebookAdsApi.init 재호출은 exchange_token 내부에서 일어나지 않는다. """ mock_response = MagicMock() mock_response.json.return_value = { "access_token": "long_token_xxx", "token_type": "bearer", "expires_in": 5184000, } mock_response.raise_for_status = MagicMock() with patch("utils.meta_ads_client.requests.get", return_value=mock_response) as mock_get: result = client.exchange_token() # 반환값이 str 타입이고 새 토큰 문자열임 assert isinstance(result, str) assert result == "long_token_xxx" # requests.get이 한 번 호출됨 mock_get.assert_called_once() # ────────────────────────────────────────────────────────────────────────────── # 4. test_check_token (구 get_token_info) # ────────────────────────────────────────────────────────────────────────────── class TestGetTokenInfo: """check_token() 테스트""" def test_get_token_info(self, client): """requests.get mock으로 토큰 정보 dict를 올바르게 반환한다.""" expected_data = { "data": { "is_valid": True, "expires_at": 1234567890, "app_id": "test_app_id", "user_id": "999", "scopes": ["ads_read", "ads_management"], } } mock_response = MagicMock() mock_response.json.return_value = expected_data mock_response.raise_for_status = MagicMock() with patch("utils.meta_ads_client.requests.get", return_value=mock_response): result = client.check_token() assert isinstance(result, dict) # check_token은 "data" 키가 있으면 data 내부 dict를 반환 assert result.get("is_valid") is True assert result.get("expires_at") == 1234567890 # ────────────────────────────────────────────────────────────────────────────── # 5. test_get_account_info # ────────────────────────────────────────────────────────────────────────────── class TestGetAccountInfo: """get_account_info() 테스트""" def test_get_account_info(self, client): """AdAccount.api_get mock으로 계정 필드를 올바르게 반환한다.""" expected = { "id": "act_123456", "name": "Test Ad Account", "currency": "KRW", "account_status": 1, } client._account.api_get.return_value = expected result = client.get_account_info() assert result is not None client._account.api_get.assert_called_once() # ────────────────────────────────────────────────────────────────────────────── # 6. test_list_campaigns # ────────────────────────────────────────────────────────────────────────────── class TestListCampaigns: """list_campaigns() 테스트""" def test_list_campaigns(self, client): """AdAccount.get_campaigns mock으로 캠페인 리스트를 반환한다.""" mock_campaign_1 = MagicMock() mock_campaign_1.__getitem__ = MagicMock(side_effect=lambda k: {"id": "camp_001", "name": "Campaign 1"}[k]) mock_campaign_2 = MagicMock() mock_campaign_2.__getitem__ = MagicMock(side_effect=lambda k: {"id": "camp_002", "name": "Campaign 2"}[k]) client._account.get_campaigns.return_value = [mock_campaign_1, mock_campaign_2] result = client.list_campaigns() assert result is not None assert len(result) == 2 client._account.get_campaigns.assert_called_once() # ────────────────────────────────────────────────────────────────────────────── # 7. test_create_campaign # ────────────────────────────────────────────────────────────────────────────── class TestCreateCampaign: """create_campaign() 테스트""" def test_create_campaign(self, client): """account.create_campaign mock으로 캠페인을 생성하고 기본 status가 PAUSED임을 확인한다.""" mock_campaign = MagicMock() mock_campaign.export_all_data.return_value = { "id": "camp_new_001", "name": "New Campaign", "status": "PAUSED", } client._account.create_campaign.return_value = mock_campaign result = client.create_campaign( name="New Campaign", objective="OUTCOME_AWARENESS", ) # create_campaign이 호출됨 client._account.create_campaign.assert_called_once() # params에 status=PAUSED가 포함되었는지 확인 call_args, call_kwargs = client._account.create_campaign.call_args all_args_str = str(call_args) + str(call_kwargs) assert "PAUSED" in all_args_str def test_create_campaign_default_status_paused(self, client): """status 파라미터 미지정 시 기본값이 PAUSED이다.""" mock_campaign = MagicMock() mock_campaign.export_all_data.return_value = { "id": "camp_new_002", "name": "Test Campaign", "status": "PAUSED", } client._account.create_campaign.return_value = mock_campaign client.create_campaign(name="Test Campaign", objective="OUTCOME_TRAFFIC") # create_campaign 호출 인자에 PAUSED가 포함됨 call_args, call_kwargs = client._account.create_campaign.call_args all_args_str = str(call_args) + str(call_kwargs) assert "PAUSED" in all_args_str # ────────────────────────────────────────────────────────────────────────────── # 8. test_update_campaign # ────────────────────────────────────────────────────────────────────────────── class TestUpdateCampaign: """update_campaign() 테스트""" def test_update_campaign(self, client): """Campaign.api_update mock으로 캠페인 업데이트를 수행한다.""" with patch("utils.meta_ads_client.Campaign") as mock_campaign_cls: mock_instance = MagicMock() mock_campaign_cls.return_value = mock_instance mock_instance.api_update.return_value = mock_instance result = client.update_campaign( campaign_id="camp_001", name="Updated Campaign", status="ACTIVE", ) mock_instance.api_update.assert_called_once() # ────────────────────────────────────────────────────────────────────────────── # 9. test_delete_campaign # ────────────────────────────────────────────────────────────────────────────── class TestDeleteCampaign: """delete_campaign() 테스트""" def test_delete_campaign(self, client): """Campaign.api_delete mock으로 캠페인을 삭제하고 True를 반환한다.""" with patch("utils.meta_ads_client.Campaign") as mock_campaign_cls: mock_instance = MagicMock() mock_campaign_cls.return_value = mock_instance mock_instance.api_delete.return_value = True result = client.delete_campaign(campaign_id="camp_001") mock_instance.api_delete.assert_called_once() # 삭제 성공 시 True 반환 assert result is True # ────────────────────────────────────────────────────────────────────────────── # 10. test_upload_image # ────────────────────────────────────────────────────────────────────────────── class TestUploadImage: """upload_image() 테스트""" def test_upload_image(self, client): """AdImage mock으로 이미지를 업로드하고 hash 문자열을 반환한다.""" mock_image = MagicMock() mock_image.export_all_data.return_value = {"hash": "abc123def456"} client._account.create_ad_image.return_value = mock_image with patch("utils.meta_ads_client.Path") as mock_path_cls: mock_path_instance = MagicMock() mock_path_cls.return_value = mock_path_instance mock_path_instance.exists.return_value = True mock_path_instance.name = "test_banner.png" mock_path_instance.__str__ = MagicMock(return_value="/tmp/test_banner.png") result = client.upload_image(image_path="/tmp/test_banner.png") # 반환값이 hash 문자열임 assert isinstance(result, str) assert result == "abc123def456" def test_upload_image_with_nested_response(self, client): """upload_image는 hash 문자열만 반환한다 (중첩 응답 처리 포함).""" mock_image = MagicMock() mock_image.export_all_data.return_value = { "images": { "test_banner.png": { "hash": "abc123def456", "url": "https://example.com/image.jpg", } } } client._account.create_ad_image.return_value = mock_image with patch("utils.meta_ads_client.Path") as mock_path_cls: mock_path_instance = MagicMock() mock_path_cls.return_value = mock_path_instance mock_path_instance.exists.return_value = True mock_path_instance.name = "test_banner.png" mock_path_instance.__str__ = MagicMock(return_value="/tmp/test_banner.png") result = client.upload_image(image_path="/tmp/test_banner.png") # 반환값이 str 타입 (hash 문자열) assert isinstance(result, str) # ────────────────────────────────────────────────────────────────────────────── # 11. test_get_insights # ────────────────────────────────────────────────────────────────────────────── class TestGetInsights: """get_insights() 테스트""" def test_get_insights(self, client): """Campaign.get_insights mock으로 인사이트 데이터를 반환한다.""" mock_insight_1 = MagicMock() mock_insight_1.export_all_data.return_value = { "impressions": "10000", "clicks": "500", "spend": "50000", "reach": "8000", "ctr": "5.0", } with patch("utils.meta_ads_client.Campaign") as mock_campaign_cls: mock_campaign_instance = MagicMock() mock_campaign_cls.return_value = mock_campaign_instance mock_campaign_instance.get_insights.return_value = [mock_insight_1] result = client.get_insights(object_id="123", object_type="campaign") assert result is not None assert len(result) == 1 mock_campaign_instance.get_insights.assert_called_once() def test_get_insights_with_fields(self, client): """기본 필드(impressions, clicks, spend)를 포함하여 인사이트를 조회한다.""" with patch("utils.meta_ads_client.Campaign") as mock_campaign_cls: mock_campaign_instance = MagicMock() mock_campaign_cls.return_value = mock_campaign_instance mock_campaign_instance.get_insights.return_value = [] result = client.get_insights(object_id="123", object_type="campaign") call_args, call_kwargs = mock_campaign_instance.get_insights.call_args # fields가 인자로 전달됨 all_args_str = str(call_args) + str(call_kwargs) # 기본 필드 중 하나 이상이 포함되어야 함 assert any(field in all_args_str for field in ["impressions", "clicks", "spend", "reach"]) # ────────────────────────────────────────────────────────────────────────────── # 13. test_update_env_token # ────────────────────────────────────────────────────────────────────────────── class TestUpdateEnvToken: """update_env_token() 테스트""" def test_update_env_token_replaces_line(self, client): """ 실제 임시 파일을 사용하여 META_ACCESS_TOKEN 라인이 새 토큰으로 교체되는지 확인한다. """ original_content = ( "export META_APP_ID=test_app_id\n" "export META_APP_SECRET=test_app_secret\n" "export META_ACCESS_TOKEN=old_token\n" "export META_AD_ACCOUNT_ID=act_123456\n" ) new_token = "new_long_token_xyz" with tempfile.NamedTemporaryFile(mode="w", suffix=".keys", delete=False) as tmp: tmp.write(original_content) tmp_path = tmp.name try: # client._env_keys_path를 tmp 파일로 설정 client._env_keys_path = tmp_path client.update_env_token(new_token=new_token) updated_content = Path(tmp_path).read_text() assert new_token in updated_content assert "old_token" not in updated_content finally: os.unlink(tmp_path) def test_update_env_token_with_real_tempfile(self, client): """실제 임시 파일을 사용하여 META_ACCESS_TOKEN 라인 교체를 검증한다.""" original_content = ( "export META_APP_ID=test_app_id\n" "export META_APP_SECRET=test_app_secret\n" "export META_ACCESS_TOKEN=old_token\n" "export META_AD_ACCOUNT_ID=act_123456\n" ) new_token = "new_long_token_xyz" with tempfile.NamedTemporaryFile(mode="w", suffix=".keys", delete=False) as tmp: tmp.write(original_content) tmp_path = tmp.name try: # client._env_keys_path를 tmp 파일로 설정 client._env_keys_path = tmp_path client.update_env_token(new_token=new_token) updated_content = Path(tmp_path).read_text() assert new_token in updated_content assert "old_token" not in updated_content finally: os.unlink(tmp_path)