test_meta_ads_conversions.py
1 """Meta Ads Conversions API (CAPI) ユニットテスト 2 3 send_event / send_purchase_event / send_lead_event および 4 ハッシュ化ユーティリティ（hash_email, hash_phone, normalize_user_data）をテストする。 5 """ 6 7 from __future__ import annotations 8 9 import hashlib 10 from unittest.mock import AsyncMock 11 12 import pytest 13 14 from mureo.meta_ads._hash_utils import ( 15 hash_email, 16 hash_phone, 17 normalize_user_data, 18 ) 19 from mureo.meta_ads._conversions import ConversionsMixin 20 21 22 # --------------------------------------------------------------------------- 23 # ヘルパー: ConversionsMixinをテスト可能にするモッククラス 24 # --------------------------------------------------------------------------- 25 26 27 def _make_conversions_client() -> ConversionsMixin: 28 """ConversionsMixinにモック _post を付与したインスタンスを生成""" 29 30 class MockClient(ConversionsMixin): 31 def __init__(self) -> None: 32 self._post = AsyncMock( # type: ignore[assignment] 33 return_value={"events_received": 1, "fbtrace_id": "trace123"} 34 ) 35 36 return MockClient() 37 38 39 # =========================================================================== 40 # hash_email テスト 41 # =========================================================================== 42 43 44 @pytest.mark.unit 45 class TestHashEmail: 46 def test_basic(self) -> None: 47 """メールをSHA-256ハッシュ化""" 48 result = hash_email("Test@Example.COM") 49 expected = hashlib.sha256("test@example.com".encode()).hexdigest() 50 assert result == expected 51 52 def test_strips_whitespace(self) -> None: 53 """前後の空白を除去してからハッシュ化""" 54 result = hash_email(" user@test.com ") 55 expected = hashlib.sha256("user@test.com".encode()).hexdigest() 56 assert result == expected 57 58 59 # =========================================================================== 60 # hash_phone テスト 61 # =========================================================================== 62 63 64 @pytest.mark.unit 65 class TestHashPhone: 66 def test_basic(self) -> None: 67 """電話番号を数字のみに正規化してSHA-256ハッシュ化""" 68 result = hash_phone("+81-90-1234-5678") 69 expected = hashlib.sha256("819012345678".encode()).hexdigest() 70 assert result == expected 71 72 def test_strips_spaces_and_parens(self) -> None: 73 """スペース・括弧・ハイフンを除去""" 74 result = hash_phone("(090) 1234-5678") 75 expected = hashlib.sha256("09012345678".encode()).hexdigest() 76 assert result == expected 77 78 79 # =========================================================================== 80 # normalize_user_data テスト 81 # =========================================================================== 82 83 84 @pytest.mark.unit 85 class TestNormalizeUserData: 86 def test_hashes_em_and_ph(self) -> None: 87 """em(email)とph(phone)を自動ハッシュ化""" 88 user_data = { 89 "em": "user@example.com", 90 "ph": "+81901234567", 91 "client_ip_address": "1.2.3.4", 92 "client_user_agent": "Mozilla/5.0", 93 } 94 result = normalize_user_data(user_data) 95 # emとphはハッシュ化される 96 assert result["em"] == hash_email("user@example.com") 97 assert result["ph"] == hash_phone("+81901234567") 98 # 非PIIフィールドはそのまま 99 assert result["client_ip_address"] == "1.2.3.4" 100 assert result["client_user_agent"] == "Mozilla/5.0" 101 102 def test_already_hashed_skips(self) -> None: 103 """既にSHA-256ハッシュ済み（64文字の16進数）ならスキップ""" 104 already_hashed = hashlib.sha256("test@example.com".encode()).hexdigest() 105 user_data = { 106 "em": already_hashed, 107 "ph": hashlib.sha256("09012345678".encode()).hexdigest(), 108 } 109 result = normalize_user_data(user_data) 110 assert result["em"] == already_hashed 111 assert result["ph"] == user_data["ph"] 112 113 def test_list_values_hashed(self) -> None: 114 """emやphがリスト形式の場合も各要素をハッシュ化""" 115 user_data = { 116 "em": ["user1@example.com", "user2@example.com"], 117 } 118 result = normalize_user_data(user_data) 119 assert isinstance(result["em"], list) 120 assert result["em"][0] == hash_email("user1@example.com") 121 assert result["em"][1] == hash_email("user2@example.com") 122 123 def test_handles_fn_ln_and_other_pii_fields(self) -> None: 124 """fn(名), ln(姓)等のPIIフィールドもハッシュ化""" 125 user_data = { 126 "fn": "Taro", 127 "ln": "Yamada", 128 "ct": "Tokyo", 129 "st": "Tokyo", 130 "zp": "1000001", 131 "country": "jp", 132 } 133 result = normalize_user_data(user_data) 134 assert result["fn"] == hashlib.sha256("taro".encode()).hexdigest() 135 assert result["ln"] == hashlib.sha256("yamada".encode()).hexdigest() 136 assert result["ct"] == hashlib.sha256("tokyo".encode()).hexdigest() 137 assert result["st"] == hashlib.sha256("tokyo".encode()).hexdigest() 138 assert result["zp"] == hashlib.sha256("1000001".encode()).hexdigest() 139 assert result["country"] == hashlib.sha256("jp".encode()).hexdigest() 140 141 142 # =========================================================================== 143 # ConversionsMixin.send_event テスト 144 # =========================================================================== 145 146 147 @pytest.mark.unit 148 class TestSendEvent: 149 @pytest.fixture() 150 def client(self) -> ConversionsMixin: 151 return _make_conversions_client() 152 153 @pytest.mark.asyncio 154 async def test_send_event(self, client: ConversionsMixin) -> None: 155 """正常にイベントを送信""" 156 events = [ 157 { 158 "event_name": "Purchase", 159 "event_time": 1700000000, 160 "action_source": "website", 161 "user_data": { 162 "em": hashlib.sha256("test@example.com".encode()).hexdigest(), 163 "client_ip_address": "1.2.3.4", 164 }, 165 "custom_data": {"currency": "USD", "value": 100.0}, 166 } 167 ] 168 result = await client.send_event("pixel123", events) 169 170 assert result["events_received"] == 1 171 client._post.assert_called_once() # type: ignore[union-attr] 172 call_args = client._post.call_args # type: ignore[union-attr] 173 assert "/pixel123/events" in call_args[0][0] 174 175 # dataパラメータにイベントが含まれる 176 post_data = call_args[1].get("data") or call_args[0][1] 177 assert "data" in post_data 178 179 @pytest.mark.asyncio 180 async def test_send_event_with_test_code(self, client: ConversionsMixin) -> None: 181 """テストイベントコード付きで送信""" 182 events = [ 183 { 184 "event_name": "Lead", 185 "event_time": 1700000000, 186 "action_source": "website", 187 "user_data": {"client_ip_address": "1.2.3.4"}, 188 } 189 ] 190 result = await client.send_event( 191 "pixel123", events, test_event_code="TEST12345" 192 ) 193 194 assert result["events_received"] == 1 195 call_args = client._post.call_args # type: ignore[union-attr] 196 post_data = call_args[1].get("data") or call_args[0][1] 197 assert post_data.get("test_event_code") == "TEST12345" 198 199 @pytest.mark.asyncio 200 async def test_send_event_api_error(self, client: ConversionsMixin) -> None: 201 """APIエラー時にRuntimeErrorを伝搬""" 202 client._post = AsyncMock( # type: ignore[assignment] 203 side_effect=RuntimeError("Meta API request failed") 204 ) 205 events = [ 206 { 207 "event_name": "Purchase", 208 "event_time": 1700000000, 209 "action_source": "website", 210 "user_data": {"client_ip_address": "1.2.3.4"}, 211 } 212 ] 213 with pytest.raises(RuntimeError, match="Meta API"): 214 await client.send_event("pixel123", events) 215 216 217 # =========================================================================== 218 # ConversionsMixin.send_purchase_event テスト 219 # =========================================================================== 220 221 222 @pytest.mark.unit 223 class TestSendPurchaseEvent: 224 @pytest.fixture() 225 def client(self) -> ConversionsMixin: 226 return _make_conversions_client() 227 228 @pytest.mark.asyncio 229 async def test_send_purchase_event(self, client: ConversionsMixin) -> None: 230 """購入イベントを正しい形式で送信""" 231 user_data = { 232 "em": "buyer@example.com", 233 "client_ip_address": "1.2.3.4", 234 } 235 result = await client.send_purchase_event( 236 pixel_id="pixel123", 237 event_time=1700000000, 238 user_data=user_data, 239 currency="JPY", 240 value=9800.0, 241 content_ids=["product_001"], 242 event_source_url="https://example.com/checkout", 243 ) 244 245 assert result["events_received"] == 1 246 call_args = client._post.call_args # type: ignore[union-attr] 247 post_data = call_args[1].get("data") or call_args[0][1] 248 249 # dataフィールドのイベント確認 250 import json 251 252 events = json.loads(post_data["data"]) 253 event = events[0] 254 assert event["event_name"] == "Purchase" 255 assert event["event_time"] == 1700000000 256 assert event["action_source"] == "website" 257 assert event["event_source_url"] == "https://example.com/checkout" 258 assert event["custom_data"]["currency"] == "JPY" 259 assert event["custom_data"]["value"] == 9800.0 260 assert event["custom_data"]["content_ids"] == ["product_001"] 261 262 263 # =========================================================================== 264 # ConversionsMixin.send_lead_event テスト 265 # =========================================================================== 266 267 268 @pytest.mark.unit 269 class TestSendLeadEvent: 270 @pytest.fixture() 271 def client(self) -> ConversionsMixin: 272 return _make_conversions_client() 273 274 @pytest.mark.asyncio 275 async def test_send_lead_event(self, client: ConversionsMixin) -> None: 276 """リードイベントを正しい形式で送信""" 277 user_data = { 278 "em": "lead@example.com", 279 "client_ip_address": "10.0.0.1", 280 } 281 result = await client.send_lead_event( 282 pixel_id="pixel456", 283 event_time=1700001000, 284 user_data=user_data, 285 event_source_url="https://example.com/contact", 286 test_event_code="TEST99", 287 ) 288 289 assert result["events_received"] == 1 290 call_args = client._post.call_args # type: ignore[union-attr] 291 post_data = call_args[1].get("data") or call_args[0][1] 292 293 import json 294 295 events = json.loads(post_data["data"]) 296 event = events[0] 297 assert event["event_name"] == "Lead" 298 assert event["event_time"] == 1700001000 299 assert event["action_source"] == "website" 300 assert event["event_source_url"] == "https://example.com/contact" 301 assert post_data.get("test_event_code") == "TEST99"