You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
326 lines
15 KiB
Python
326 lines
15 KiB
Python
# backend/tests/test_normalize_golden.py — C5 정규화 골든(provider → 내부 모델 고정 매핑)
|
|
import base64
|
|
|
|
from app.connectors.base import RawRecord
|
|
from app.connectors.calendar.normalize import account_calendar_id, normalize_event
|
|
from app.connectors.mail.normalize import normalize_email
|
|
from app.models import ConnectorAccount, ConnectorDomain
|
|
|
|
|
|
def _b64(s: str) -> str:
|
|
return base64.urlsafe_b64encode(s.encode("utf-8")).decode("utf-8").rstrip("=")
|
|
|
|
|
|
def _acct(aid, domain, name="x", tone="blue"):
|
|
return ConnectorAccount(id=aid, domain=ConnectorDomain(domain), name=name, tone=tone)
|
|
|
|
|
|
def test_gmail_email_golden():
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
payload = {
|
|
"id": "gmail-1",
|
|
"labelIds": ["UNREAD", "IMPORTANT", "CATEGORY_PERSONAL", "STARRED"],
|
|
"snippet": "미리보기 텍스트",
|
|
"internalDate": "1718000000000",
|
|
"payload": {
|
|
"headers": [
|
|
{"name": "From", "value": "hyunwoo@lumi.co"},
|
|
{"name": "To", "value": "jiwoo@lumi.co"},
|
|
{"name": "Subject", "value": "온보딩 시안 v3"},
|
|
],
|
|
"parts": [
|
|
{"mimeType": "text/plain", "body": {"data": _b64("본문 라인1\n본문 라인2")}},
|
|
{"filename": "spec.pdf", "body": {}},
|
|
],
|
|
},
|
|
}
|
|
norm = normalize_email(acct, RawRecord("gmail-1", payload, etag="hist1"), provider="gmail")
|
|
f = norm.fields
|
|
assert norm.entity_type == "email"
|
|
assert f["account"] == "personal"
|
|
assert f["from_key"] == "hyunwoo@lumi.co"
|
|
assert f["to"] == "jiwoo@lumi.co"
|
|
assert f["subject"] == "온보딩 시안 v3"
|
|
assert f["body"] == ["본문 라인1", "본문 라인2"]
|
|
assert f["labels"] == ["중요", "개인"]
|
|
assert f["read"] is False # UNREAD 존재
|
|
assert f["starred"] is True
|
|
assert f["has_attach"] is True
|
|
# 수신 시각: internalDate(epoch ms) → aware datetime(정렬·표시 기준).
|
|
assert f["received_at"] is not None
|
|
assert f["received_at"].year == 2024 # 1718000000000ms = 2024-06-10
|
|
|
|
|
|
def test_gmail_plain_strips_conditional_comments_and_vml():
|
|
"""text/plain 파트에 MSO 조건부 주석·VML 블록·<url> 래핑이 섞인 (Google 알림류) 메일.
|
|
본문에 HTML 잔재가 남지 않고 URL 은 언랩되어 보존되어야 한다."""
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
body = (
|
|
"<!--[if !mso]><!-->\n"
|
|
"<!--[if false]><!-->\n"
|
|
"본인의 계정으로 로그인했습니다\n"
|
|
"<https://c.gle/AOPyTOKEN>magnific.com\n"
|
|
"<!--[if mso]>\n"
|
|
'<v:roundrect xmlns:v="urn:schemas-microsoft-com:vml"\n'
|
|
'style="height:48px;" fillcolor="#0b57d0">\n'
|
|
"<w:anchorlock/>\n"
|
|
"<![endif]-->\n"
|
|
"계정으로 이동\n"
|
|
"© 2026 Google LLC"
|
|
)
|
|
payload = {
|
|
"id": "gmail-2",
|
|
"labelIds": ["CATEGORY_UPDATES"],
|
|
"snippet": "스니펫",
|
|
"internalDate": "1718000000000",
|
|
"payload": {
|
|
"mimeType": "text/plain",
|
|
"body": {"data": _b64(body)},
|
|
},
|
|
}
|
|
norm = normalize_email(acct, RawRecord("gmail-2", payload, etag="h"), provider="gmail")
|
|
text = "\n".join(norm.fields["body"])
|
|
assert "<!--" not in text and "<![" not in text # 조건부 주석 제거
|
|
assert "v:roundrect" not in text and "<w:" not in text # VML 제거
|
|
assert "fillcolor" not in text and "anchorlock" not in text
|
|
assert "본인의 계정으로 로그인했습니다" in text # 평문 보존
|
|
assert "https://c.gle/AOPyTOKEN" in text # URL 언랩·보존
|
|
assert "© 2026 Google LLC" in text # 엔티티 디코딩
|
|
|
|
|
|
def test_gmail_captures_html_body_and_strips_active_content():
|
|
"""text/html 파트가 있으면 원본 HTML 을 body_html 로 보존(서식/이미지) 하되,
|
|
script/on* 등 능동 콘텐츠는 제거하고 인라인 cid 첨부는 content_id 와 함께 수집한다."""
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
html = (
|
|
'<div style="color:red">안녕하세요 <b>홍길동</b>님'
|
|
'<img src="https://ex.com/a.png"><img src="cid:logo123">'
|
|
'<script>alert(1)</script><a href="javascript:evil()" onclick="x()">링크</a></div>'
|
|
)
|
|
payload = {
|
|
"id": "gmail-h",
|
|
"labelIds": [],
|
|
"snippet": "s",
|
|
"internalDate": "1718000000000",
|
|
"payload": {
|
|
"mimeType": "multipart/alternative",
|
|
"parts": [
|
|
{"mimeType": "text/plain", "body": {"data": _b64("안녕하세요 홍길동님")}},
|
|
{"mimeType": "text/html", "body": {"data": _b64(html)}},
|
|
{
|
|
"mimeType": "image/png",
|
|
"filename": "logo.png",
|
|
"headers": [{"name": "Content-ID", "value": "<logo123>"}],
|
|
"body": {"attachmentId": "att-logo", "size": 99},
|
|
},
|
|
],
|
|
},
|
|
}
|
|
norm = normalize_email(acct, RawRecord("gmail-h", payload, etag="h"), provider="gmail")
|
|
f = norm.fields
|
|
bh = f["body_html"]
|
|
assert "<b>홍길동</b>" in bh # 서식 보존
|
|
assert 'src="https://ex.com/a.png"' in bh # 이미지 보존
|
|
assert "cid:logo123" in bh # cid 는 읽기 시 치환(정규화 단계선 유지)
|
|
assert "<script" not in bh and "alert(1)" not in bh # script 제거
|
|
assert "onclick" not in bh # on* 핸들러 제거
|
|
assert "javascript:" not in bh # javascript: 스킴 제거
|
|
# 인라인 cid 이미지가 content_id 와 함께 첨부로 수집됨
|
|
inline = [a for a in f["attachments"] if a.get("content_id") == "logo123"]
|
|
assert inline and inline[0]["inline"] is True and inline[0]["id"] == "att-logo"
|
|
|
|
|
|
def test_outlook_captures_html_body():
|
|
"""Outlook html 본문(Prefer html)이 body_html 로 보존된다."""
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
payload = {
|
|
"id": "ol-h",
|
|
"from": {"emailAddress": {"address": "a@b.com"}},
|
|
"subject": "공지",
|
|
"body": {"contentType": "html", "content": "<p>안내 <b>본문</b></p><script>x()</script>"},
|
|
"toRecipients": [{"emailAddress": {"address": "me@x.com"}}],
|
|
"attachments": [],
|
|
}
|
|
payload["receivedDateTime"] = "2026-06-18T11:57:01Z"
|
|
norm = normalize_email(acct, RawRecord("ol-h", payload, etag="h"), provider="outlook")
|
|
bh = norm.fields["body_html"]
|
|
assert "<b>본문</b>" in bh and "<script" not in bh
|
|
# 수신 시각: receivedDateTime(ISO) → aware datetime.
|
|
rcv = norm.fields["received_at"]
|
|
assert rcv is not None and rcv.year == 2026 and rcv.month == 6 and rcv.day == 18
|
|
|
|
|
|
def test_email_category_and_list_time():
|
|
"""라벨 → 카테고리 매핑(프로모션/소셜/업데이트/기본)과 상대시각 산출."""
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from app.routers.mail import _category_for, _list_time
|
|
|
|
assert _category_for(["프로모션"]) == "promotions"
|
|
assert _category_for(["소셜"]) == "social"
|
|
assert _category_for(["업데이트"]) == "updates"
|
|
assert _category_for(["포럼"]) == "updates"
|
|
assert _category_for(["중요", "개인"]) == "primary"
|
|
assert _category_for([]) == "primary"
|
|
# 상대시각: 방금 / N분 전, received_at 없으면 fallback 사용.
|
|
now = datetime.now(UTC)
|
|
assert _list_time(now, "x") == "방금"
|
|
assert _list_time(now - timedelta(minutes=8), "x") == "8분 전"
|
|
assert _list_time(None, "8분 전") == "8분 전"
|
|
|
|
|
|
def test_gmail_plain_shortens_tracking_url_and_decodes_uc_entity():
|
|
"""초장문 트래킹 URL 은 host/… 로 축약, 대문자 엔티티(&NBSP;)는 디코딩."""
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
long_url = "http://url3015.example.com/ls/click?upn=" + "A1b2C3d4" * 40
|
|
body = f"[Nellis Auction]{long_url}\n가격&NBSP;업데이트\n실제 안내 내용입니다"
|
|
payload = {
|
|
"id": "gmail-3",
|
|
"labelIds": [],
|
|
"snippet": "s",
|
|
"internalDate": "1718000000000",
|
|
"payload": {"mimeType": "text/plain", "body": {"data": _b64(body)}},
|
|
}
|
|
norm = normalize_email(acct, RawRecord("gmail-3", payload, etag="h"), provider="gmail")
|
|
text = "\n".join(norm.fields["body"])
|
|
assert long_url not in text # 원본 장문 URL 사라짐
|
|
assert "http://url3015.example.com/…" in text # host/… 로 축약
|
|
assert "가격 업데이트" in text # &NBSP; → 공백
|
|
assert "실제 안내 내용입니다" in text # 본문 보존
|
|
|
|
|
|
def test_gmail_folder_routing():
|
|
"""labelIds → 시스템 폴더(스팸/휴지통/받은편지함). 카테고리는 폴더 아님."""
|
|
from app.connectors.mail.normalize import _gmail_folder
|
|
|
|
assert _gmail_folder(["SPAM"]) == "spam"
|
|
assert _gmail_folder(["TRASH"]) == "trash"
|
|
assert _gmail_folder(["INBOX", "CATEGORY_PROMOTIONS"]) == "inbox"
|
|
assert _gmail_folder(["SENT"]) == "sent"
|
|
|
|
|
|
def test_email_in_folder_predicate():
|
|
"""폴더 슬러그 필터: 받은편지함=집중만, 카테고리/라벨/스팸/휴지통/커스텀 분기."""
|
|
from app.models import Email
|
|
from app.routers.mail import _email_in_folder
|
|
|
|
inbox = Email(id="a", account="x", from_key="f", folder="inbox", labels=[])
|
|
promo = Email(id="b", account="x", from_key="f", folder="inbox", labels=["프로모션"])
|
|
spam = Email(id="c", account="x", from_key="f", folder="spam", labels=[])
|
|
labeled = Email(id="d", account="x", from_key="f", folder="inbox", labels=["Shopping"])
|
|
custom = Email(id="e", account="x", from_key="f", folder="of:XYZ", labels=[])
|
|
|
|
assert _email_in_folder(inbox, "inbox") is True
|
|
assert _email_in_folder(promo, "inbox") is False # 프로모션은 집중(받은편지함)서 빠짐
|
|
assert _email_in_folder(promo, "others") is True # → 기타로
|
|
assert _email_in_folder(promo, "cat:promotions") is True
|
|
assert _email_in_folder(spam, "spam") is True
|
|
assert _email_in_folder(spam, "inbox") is False # 스팸은 받은편지함에 안 나옴
|
|
assert _email_in_folder(labeled, "label:Shopping") is True
|
|
assert _email_in_folder(custom, "of:XYZ") is True
|
|
assert _email_in_folder(inbox, "of:XYZ") is False
|
|
|
|
|
|
def test_sender_display_name():
|
|
"""보낸사람 표시 이름: '이름 <메일>'의 이름, 없으면 도메인 추정, 메일주소 분리."""
|
|
from app.connectors.mail.normalize import _outlook_from, sender_parts
|
|
|
|
assert sender_parts("Costco Wholesale <c@digital.costco.com>") == (
|
|
"Costco Wholesale",
|
|
"c@digital.costco.com",
|
|
)
|
|
# 표시 이름 없음 → 도메인(SLD) 기반 추정(서브도메인 제거).
|
|
assert sender_parts("noreply@e.coupang.com") == ("Coupang", "noreply@e.coupang.com")
|
|
assert sender_parts("info@trupanion.com")[0] == "Trupanion"
|
|
# Outlook 발신자: 이름 보존('이름 <메일>'), 이름 없으면 주소만.
|
|
assert _outlook_from({"name": "MS Team", "address": "a@ms.com"}) == "MS Team <a@ms.com>"
|
|
assert _outlook_from({"name": "", "address": "n@x.com"}) == "n@x.com"
|
|
|
|
|
|
def test_outlook_inference_and_other_folder():
|
|
"""Outlook inferenceClassification(focused/other) 정규화 + '기타' 폴더 라우팅."""
|
|
from app.connectors.mail.normalize import normalize_email
|
|
from app.models import Email
|
|
from app.routers.mail import _email_in_folder
|
|
|
|
acct = _acct("ca-mail-personal", "mail")
|
|
payload = {
|
|
"id": "ol-i",
|
|
"from": {"emailAddress": {"name": "쿠팡", "address": "x@coupang.com"}},
|
|
"subject": "광고",
|
|
"body": {"contentType": "text", "content": "내용"},
|
|
"toRecipients": [{"emailAddress": {"address": "me@x.com"}}],
|
|
"attachments": [],
|
|
"inferenceClassification": "other",
|
|
"receivedDateTime": "2026-06-18T11:57:01Z",
|
|
}
|
|
norm = normalize_email(acct, RawRecord("ol-i", payload, etag="i"), provider="outlook")
|
|
assert norm.fields["inference"] == "other"
|
|
assert norm.fields["from_key"] == "쿠팡 <x@coupang.com>"
|
|
|
|
# 집중(inbox)/기타(others) 탭 분류 — Outlook inference + Gmail 카테고리 통합.
|
|
ol_focused = Email(id="f", account="x", from_key="f", folder="inbox", inference="focused")
|
|
ol_other = Email(id="o", account="x", from_key="f", folder="inbox", inference="other")
|
|
gm_primary = Email(id="g", account="x", from_key="f", folder="inbox", labels=[])
|
|
gm_promo = Email(id="p", account="x", from_key="f", folder="inbox", labels=["프로모션"])
|
|
# 집중: Outlook focused + Gmail 기본(primary)
|
|
assert _email_in_folder(ol_focused, "inbox") is True
|
|
assert _email_in_folder(gm_primary, "inbox") is True
|
|
assert _email_in_folder(ol_other, "inbox") is False
|
|
assert _email_in_folder(gm_promo, "inbox") is False
|
|
# 기타: Outlook other + Gmail 비기본(프로모션 등)
|
|
assert _email_in_folder(ol_other, "others") is True
|
|
assert _email_in_folder(gm_promo, "others") is True
|
|
assert _email_in_folder(ol_focused, "others") is False
|
|
assert _email_in_folder(gm_primary, "others") is False
|
|
|
|
|
|
def test_google_calendar_event_golden():
|
|
acct = _acct("ca-cal-google", "calendar")
|
|
acct.provider = "google_calendar"
|
|
acct.external_account_id = "me@gmail.com"
|
|
payload = {
|
|
"id": "ev-1",
|
|
"summary": "분기 전략 미팅",
|
|
"location": "대회의실",
|
|
"start": {"dateTime": "2026-06-10T14:00:00+09:00"},
|
|
"end": {"dateTime": "2026-06-10T15:00:00+09:00"},
|
|
"attendees": [{"email": "a@x.co"}, {"email": "b@x.co"}],
|
|
}
|
|
norm = normalize_event(acct, RawRecord("ev-1", payload), provider="google_calendar")
|
|
f = norm.fields
|
|
assert norm.entity_type == "event"
|
|
assert f["day"] == 10 and f["start"] == "14:00" and f["end"] == "15:00"
|
|
assert f["title"] == "분기 전략 미팅" and f["loc"] == "대회의실"
|
|
# 이벤트는 연결된 Google 계정 캘린더(계정별 슬러그)로 귀속.
|
|
assert f["cal"] == account_calendar_id(acct) and f["cal"].startswith("gcal-")
|
|
assert f["people"] == "a@x.co, b@x.co"
|
|
|
|
|
|
def test_outlook_calendar_event_golden():
|
|
acct = _acct("ca-cal-outlook", "calendar")
|
|
acct.provider = "outlook_calendar"
|
|
acct.external_account_id = "me@corp.com"
|
|
payload = {
|
|
"id": "AAMk-ev-1",
|
|
"subject": "주간 스탠드업",
|
|
"bodyPreview": "지난주 회고 + 이번주 계획",
|
|
"location": {"displayName": "3층 포커스룸"},
|
|
"start": {"dateTime": "2026-06-12T10:30:00.0000000", "timeZone": "Korea Standard Time"},
|
|
"end": {"dateTime": "2026-06-12T11:00:00.0000000", "timeZone": "Korea Standard Time"},
|
|
"attendees": [
|
|
{"emailAddress": {"address": "lead@corp.com", "name": "리드"}},
|
|
{"emailAddress": {"address": "me@corp.com"}},
|
|
],
|
|
"@odata.etag": 'W/"abc"',
|
|
}
|
|
norm = normalize_event(acct, RawRecord("AAMk-ev-1", payload), provider="outlook_calendar")
|
|
f = norm.fields
|
|
assert norm.entity_type == "event"
|
|
assert f["day"] == 12 and f["start"] == "10:30" and f["end"] == "11:00"
|
|
assert f["title"] == "주간 스탠드업" and f["loc"] == "3층 포커스룸"
|
|
# Outlook 이벤트도 연결된 계정 캘린더(계정별 슬러그)로 귀속.
|
|
assert f["cal"] == account_calendar_id(acct) and f["cal"].startswith("ocal-")
|
|
assert f["note"] == "지난주 회고 + 이번주 계획"
|
|
assert f["people"] == "lead@corp.com, me@corp.com"
|