"""Feed domain — invariant tests. Covers the contracts the production system relies on: * visibility: ``department`` posts are hidden from cross-department viewers * soft-delete: deleted posts never surface in list / detail / search * likes: UNIQUE(post_id, person_id) — double-like is a no-op * blessings: UNIQUE(celebration_id, sender) — double-send returns ConflictError * @ mentions: insert one ``feed_mentions`` row per mentioned id * celebration cron: idempotent rerun + leap-year fallback (Feb 29 → Feb 28) * timezone: hire-anniversary fires in the person's local TZ """ from __future__ import annotations import uuid from datetime import date, datetime from unittest.mock import AsyncMock from zoneinfo import ZoneInfo # The celebration cron computes ``today`` in the person's local TZ # (resolve_timezone(person)). All test persons here use ``country='CN'`` # → Asia/Shanghai. Tests must compute the matching wall-clock date in # the SAME TZ — otherwise a CI run that straddles 16:00 UTC (= 00:00 # SHA) produces a birthday/hire_date of UTC-today which is SHA-yesterday, # the cron looks for SHA-today's MM-DD, finds no match, all assertions # go to zero. Same root cause as ``feedback_cron_must_respect_local_tz``. _SHA = ZoneInfo("Asia/Shanghai") import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ConflictError, ForbiddenError from app.domains.feed.celebration_service import ( _matches_md, _years_since, scan_today_celebrations, ) from app.domains.feed.models import ( POST_TYPE_AGGREGATE_BIRTHDAY, FeedCelebration, FeedComment, FeedLike, FeedMention, FeedPost, ) from app.domains.feed.service import FeedService from app.identity.person_models import IdentityOrgMembership, IdentityPerson # ── Helpers ──────────────────────────────────────────────────────── async def _mk_person( db: AsyncSession, *, email: str, name: str, department: str | None = None, birthday: date | None = None, hire_date: date | None = None, manager_person_id: str | None = None, hrbp_person_id: str | None = None, country: str | None = None, ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=email.split("@")[0], work_email=email, display_name=name, employment_status="active", is_china_employee=True, birthday=birthday, hire_date=hire_date, country=country, ) if manager_person_id: p.manager_person_id = uuid.UUID(manager_person_id) if hrbp_person_id: p.hrbp_person_id = uuid.UUID(hrbp_person_id) db.add(p) await db.flush() if department: membership = IdentityOrgMembership( id=uuid.uuid4(), person_id=p.id, source_system="successfactors", org_node_source_id=department.lower(), org_node_type="department", org_node_name=department, is_primary=True, membership_status="active", ) db.add(membership) await db.flush() # Refresh with eager attributes so subsequent property reads # (``primary_department``, ``role_names`` via ``user_roles``) don't # trip MissingGreenlet on lazy-load outside of an async context. await db.refresh( p, attribute_names=["org_memberships", "user_roles"], ) return p # ── Tests ────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_visibility_department_hides_other_dept(db_session: AsyncSession) -> None: alice = await _mk_person(db_session, email="alice@x.com", name="Alice", department="Eng") bob = await _mk_person(db_session, email="bob@x.com", name="Bob", department="Eng") eve = await _mk_person(db_session, email="eve@x.com", name="Eve", department="Sales") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="dept-only", visibility="department", category="general", mentioned_person_ids=[], attachments=[], ) # Alice sees it posts, _ = await svc.list_posts(viewer=alice, tab="public", limit=20) assert post.id in [p.id for p in posts] # Bob (same dept) sees it posts_bob, _ = await svc.list_posts(viewer=bob, tab="public", limit=20) assert post.id in [p.id for p in posts_bob] # Eve (other dept) does NOT see it posts_eve, _ = await svc.list_posts(viewer=eve, tab="public", limit=20) assert post.id not in [p.id for p in posts_eve] @pytest.mark.asyncio async def test_soft_delete_removes_from_list(db_session: AsyncSession) -> None: alice = await _mk_person(db_session, email="alice@x.com", name="Alice") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="hello", visibility="public", category="general", mentioned_person_ids=[], attachments=[], ) await svc.soft_delete_post(post_id=post.id, viewer=alice) posts, _ = await svc.list_posts(viewer=alice, tab="public", limit=20) assert post.id not in [p.id for p in posts] @pytest.mark.asyncio async def test_like_unique_idempotent(db_session: AsyncSession) -> None: alice = await _mk_person(db_session, email="alice@x.com", name="Alice") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="hi", visibility="public", category="general", mentioned_person_ids=[], attachments=[], ) n1, changed1 = await svc.toggle_like(post_id=post.id, person=alice, like=True) n2, changed2 = await svc.toggle_like(post_id=post.id, person=alice, like=True) assert n1 == 1 and n2 == 1 # First call inserted the like row; the repeat is a no-op so the # caller must NOT fire another "X liked your post" notification. assert changed1 is True and changed2 is False n3, changed3 = await svc.toggle_like(post_id=post.id, person=alice, like=False) n4, changed4 = await svc.toggle_like(post_id=post.id, person=alice, like=False) assert n3 == 0 and n4 == 0 assert changed3 is True and changed4 is False # Re-like for the row-count assertion below. await svc.toggle_like(post_id=post.id, person=alice, like=True) rows = (await db_session.execute( select(FeedLike).where(FeedLike.post_id == post.id) )).scalars().all() assert len(rows) == 1 @pytest.mark.asyncio async def test_mentions_inserted(db_session: AsyncSession) -> None: alice = await _mk_person(db_session, email="alice@x.com", name="Alice") bob = await _mk_person(db_session, email="bob@x.com", name="Bob") eve = await _mk_person(db_session, email="eve@x.com", name="Eve") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="hey @Bob @Eve", visibility="public", category="general", mentioned_person_ids=[str(bob.id), str(eve.id), str(alice.id)], # self-mention dropped attachments=[], ) rows = (await db_session.execute( select(FeedMention).where(FeedMention.post_id == post.id) )).scalars().all() mentioned = {str(r.mentioned_person_id) for r in rows} assert mentioned == {str(bob.id), str(eve.id)} @pytest.mark.asyncio async def test_comment_mentions_capped(db_session: AsyncSession) -> None: """Regression (2026-07-08): a single comment must not write an unbounded number of FeedMention rows — cap at MAX_MENTIONS_PER_POST (mirrors create_user_post), bounding both the rows and the Feishu mention fan-out.""" from app.domains.feed.service import MAX_MENTIONS_PER_POST alice = await _mk_person(db_session, email="a2@x.com", name="Alice2") post = await svc_create(db_session, alice) svc = FeedService(db_session) # Ask to mention far more than the cap (real persons — FeedMention FKs). people = [ await _mk_person(db_session, email=f"m{i}@x.com", name=f"M{i}") for i in range(MAX_MENTIONS_PER_POST + 5) ] many = [str(p.id) for p in people] comment = await svc.add_comment( post_id=str(post.id), author=alice, body="ping everyone", mentioned_person_ids=many, reply_to_comment_id=None, ) rows = (await db_session.execute( select(FeedMention).where(FeedMention.comment_id == comment.id) )).scalars().all() assert len(rows) == MAX_MENTIONS_PER_POST async def svc_create(db_session, author): """Small helper: a public post to hang a comment off.""" post, _created = await FeedService(db_session).create_user_post( author=author, body="parent", visibility="public", category="general", mentioned_person_ids=[], attachments=[], ) return post @pytest.mark.asyncio async def test_celebration_cron_idempotent(db_session: AsyncSession) -> None: today = datetime.now(_SHA).date() person = await _mk_person( db_session, email="a@x.com", name="Anna", birthday=today, hire_date=date(today.year - 3, today.month, today.day), country="CN", ) r1 = await scan_today_celebrations(db_session) r2 = await scan_today_celebrations(db_session) cels = (await db_session.execute( select(FeedCelebration).where(FeedCelebration.person_id == str(person.id)) )).scalars().all() assert len(cels) == 2 # birthday + anniversary assert r1["new_birthdays"] + r1["new_anniversaries"] >= 2 assert r2["new_birthdays"] == 0 and r2["new_anniversaries"] == 0 # System posts present, exactly one per celebration sys_posts = (await db_session.execute( select(FeedPost).where(FeedPost.celebration_id.in_([c.id for c in cels])) )).scalars().all() assert len(sys_posts) == 2 @pytest.mark.asyncio async def test_leap_day_fallback() -> None: # In a non-leap year, Feb 29 birthday matches Feb 28. leap_anchor = date(2020, 2, 29) today_nonleap = date(2026, 2, 28) assert _matches_md(today_nonleap, leap_anchor) is True # In a leap year, the fallback does NOT fire on the 28th — the # real 29th match handles it. today_leap = date(2024, 2, 28) assert _matches_md(today_leap, leap_anchor) is False today_leap_29 = date(2024, 2, 29) assert _matches_md(today_leap_29, leap_anchor) is True def test_years_since() -> None: assert _years_since(date(2026, 6, 1), date(2020, 6, 1)) == 6 assert _years_since(date(2026, 5, 31), date(2020, 6, 1)) == 5 @pytest.mark.asyncio async def test_blessing_unique_idempotent(db_session: AsyncSession) -> None: today = datetime.now(_SHA).date() target = await _mk_person( db_session, email="t@x.com", name="Target", birthday=today, country="CN", ) sender = await _mk_person(db_session, email="s@x.com", name="Sender") await scan_today_celebrations(db_session) cel = (await db_session.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(target.id), FeedCelebration.celebration_type == "birthday", ) )).scalar_one() svc = FeedService(db_session) await svc.send_blessing( celebration_id=cel.id, sender=sender, sticker_key="cake", message="happy bday!", ) with pytest.raises(ConflictError): await svc.send_blessing( celebration_id=cel.id, sender=sender, sticker_key="gift", message="oops twice", ) @pytest.mark.asyncio async def test_list_blessings_sender_omits_work_email(db_session: AsyncSession) -> None: """Regression (#1660·SEC): the blessing SENDER projection must not leak ``work_email``. ``/celebrations/{id}`` is id-enumerable by any authenticated user, so echoing sender emails turned it into a roster-email harvester. Sender exposes name / avatar / department only.""" today = datetime.now(_SHA).date() target = await _mk_person( db_session, email="target@x.com", name="Target", birthday=today, country="CN", ) sender = await _mk_person(db_session, email="sender@x.com", name="Sender") await scan_today_celebrations(db_session) cel = (await db_session.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(target.id), FeedCelebration.celebration_type == "birthday", ) )).scalar_one() svc = FeedService(db_session) await svc.send_blessing( celebration_id=cel.id, sender=sender, sticker_key="cake", message="happy bday!", ) blessings = await svc.list_blessings(celebration_id=cel.id) assert len(blessings) == 1 s = blessings[0]["sender"] assert "work_email" not in s assert s["display_name"] == "Sender" assert set(s) == {"id", "display_name", "avatar_url", "department"} @pytest.mark.asyncio async def test_blessing_self_forbidden(db_session: AsyncSession) -> None: today = datetime.now(_SHA).date() target = await _mk_person( db_session, email="t@x.com", name="Target", birthday=today, country="CN", ) await scan_today_celebrations(db_session) cel = (await db_session.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(target.id), FeedCelebration.celebration_type == "birthday", ) )).scalar_one() svc = FeedService(db_session) with pytest.raises(ForbiddenError): await svc.send_blessing( celebration_id=cel.id, sender=target, sticker_key="cake", message="hello self", ) @pytest.mark.asyncio async def test_blessing_creates_mirror_comment(db_session: AsyncSession) -> None: today = datetime.now(_SHA).date() target = await _mk_person( db_session, email="t@x.com", name="Target", birthday=today, country="CN", ) sender = await _mk_person(db_session, email="s@x.com", name="Sender") await scan_today_celebrations(db_session) cel = (await db_session.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(target.id), FeedCelebration.celebration_type == "birthday", ) )).scalar_one() svc = FeedService(db_session) b = await svc.send_blessing( celebration_id=cel.id, sender=sender, sticker_key="cake", message="have a great one!", ) # Mirror comment exists + back-pointer set. assert b.mirror_comment_id is not None mirror = await db_session.get(FeedComment, b.mirror_comment_id) assert mirror is not None assert mirror.sticker_key == "cake" # System post's comment_count incremented. sys_post = (await db_session.execute( select(FeedPost).where(FeedPost.celebration_id == cel.id) )).scalar_one() assert sys_post.comment_count == 1 # Python-side ``created_at`` populated on both blessing + mirror # comment — same MissingGreenlet guard as ``add_comment`` / # ``create_user_post`` (PR #658). Without this, the router's # ``_iso_utc(blessing.created_at)`` after ``db.commit()`` triggers # a lazy server-default refresh that raises in async context → 500 # AFTER the blessing was already committed (surfaces as "send # button errors but blessing arrived", 2026-06-02 round). assert b.created_at is not None assert mirror.created_at is not None assert mirror.updated_at is not None @pytest.mark.asyncio async def test_birthday_hidden_pref_skips(db_session: AsyncSession) -> None: from app.domains.feed.models import ( BIRTHDAY_VISIBILITY_HIDDEN, PREF_BIRTHDAY_VISIBILITY, IdentityPersonPreference, ) today = datetime.now(_SHA).date() person = await _mk_person( db_session, email="p@x.com", name="Hidden", birthday=today, country="CN", ) db_session.add(IdentityPersonPreference( person_id=str(person.id), pref_key=PREF_BIRTHDAY_VISIBILITY, pref_value=BIRTHDAY_VISIBILITY_HIDDEN, )) await db_session.flush() r = await scan_today_celebrations(db_session) assert r["skipped_hidden"] >= 1 cels = (await db_session.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(person.id), FeedCelebration.celebration_type == "birthday", ) )).scalars().all() assert cels == [] @pytest.mark.asyncio async def test_hr_role_can_manage_announcements(db_session: AsyncSession) -> None: """Regression: ``_is_admin`` checked ``'hr_admin'`` (wrong) instead of ``'hr'`` (the real system role token from ``HR_ROLE``), locking HR out of /admin/announcements despite the user explicitly putting announcement management in the HR menu. """ import uuid as _uuid from app.auth.models import AuthRole, AuthUserRole from app.core.rbac import HR_ROLE from app.domains.feed.service import FeedService, _is_admin # Seed the ``hr`` role row + grant it to a fresh person. hr_role = (await db_session.execute( select(AuthRole).where(AuthRole.name == HR_ROLE) )).scalar_one_or_none() if hr_role is None: hr_role = AuthRole(id=_uuid.uuid4(), name=HR_ROLE, description="HR") db_session.add(hr_role) await db_session.flush() hr_person = await _mk_person( db_session, email="hr@x.com", name="HR Person", ) db_session.add(AuthUserRole( user_id=hr_person.id, role_id=hr_role.id, )) await db_session.flush() await db_session.refresh(hr_person, attribute_names=["user_roles"]) assert _is_admin(hr_person) is True svc = FeedService(db_session) ann = await svc.create_announcement( author=hr_person, title="welcome", body_markdown="hello team", category="general", color="default", is_pinned=False, publish_at=None, expire_at=None, status="published", attachments_json=None, ) assert ann.id is not None @pytest.mark.asyncio async def test_dept_visibility_widens_to_public_when_author_has_no_dept( db_session: AsyncSession, ) -> None: """Regression: a user with no primary department picking visibility=department would have created a row only they could see (department_snapshot=NULL never matches). Service widens to public instead. """ alice = await _mk_person(db_session, email="alice@x.com", name="Alice") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="dept-but-no-dept", visibility="department", category="general", mentioned_person_ids=[], attachments=[], ) assert post.visibility == "public" assert post.department_snapshot is None @pytest.mark.asyncio async def test_aggregate_banner_updated_with_count(db_session: AsyncSession) -> None: today = datetime.now(_SHA).date() await _mk_person( db_session, email="a@x.com", name="A", birthday=today, country="CN", ) await _mk_person( db_session, email="b@x.com", name="B", birthday=today, country="CN", ) await scan_today_celebrations(db_session) banners = (await db_session.execute( select(FeedPost).where(FeedPost.post_type == POST_TYPE_AGGREGATE_BIRTHDAY) )).scalars().all() assert len(banners) == 1 assert "2" in banners[0].body @pytest.mark.asyncio async def test_list_comments_is_bounded(db_session: AsyncSession) -> None: """Regression: list_comments must cap the result so a hot post can't force an unbounded materialization. With ≤limit comments all are returned; with a smaller limit the result is capped to that count.""" alice = await _mk_person(db_session, email="alice@x.com", name="Alice") svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="p", visibility="public", category="general", mentioned_person_ids=[], attachments=[], ) for i in range(5): await svc.add_comment( post_id=post.id, author=alice, body=f"c{i}", mentioned_person_ids=[], reply_to_comment_id=None, ) await db_session.flush() # Small limit caps the result. capped = await svc.list_comments(post_id=post.id, viewer=alice, limit=3) assert len(capped) == 3 # Generous limit returns everything (common case unaffected). full = await svc.list_comments(post_id=post.id, viewer=alice, limit=200) assert len(full) == 5 # All comments present (exact ordering among same-second inserts is not # asserted to avoid timestamp-precision flakiness). assert {c["body"] for c in full} == {"c0", "c1", "c2", "c3", "c4"} # ── #2124 attachment object-key ownership ─────────────────────────── def _key(purpose: str, uploader_id) -> str: return f"dev/{purpose}/2026/07/31/{uploader_id}/photo.png" @pytest.mark.asyncio async def test_post_cannot_attach_another_namespaces_object( db_session: AsyncSession, ) -> None: """The whole attack in one call. Before the gate, ``s3_key`` was stored verbatim (the only constraint was its length) and hydration minted a 10-minute presigned GET for whatever was in the column — so any active employee could post an ``inventory_evidence`` / ``vendor_assessment`` / ``self_check`` key and read the download URL out of their own post. That bypasses ``/uploads/presign-get``'s per-purpose ACL, whose docstring says in as many words that bare authentication is not enough because "the presigned URL bypasses Helix auth once minted". """ from app.core.exceptions import ValidationError mallory = await _mk_person( db_session, email="mallory@x.com", name="Mallory", department="Eng", ) victim = await _mk_person( db_session, email="victim@x.com", name="Victim", department="Eng", ) svc = FeedService(db_session) with pytest.raises(ValidationError): await svc.create_user_post( author=mallory, body="nice photo", visibility="public", category="general", mentioned_person_ids=[], attachments=[{"s3_key": _key("inventory_evidence", victim.id)}], ) @pytest.mark.asyncio async def test_post_cannot_attach_someone_elses_feed_upload( db_session: AsyncSession, ) -> None: """Right purpose, wrong uploader — still refused. ``presign_put`` stamps the uploader's id into the key path, which is the only ownership evidence the key carries. """ from app.core.exceptions import ValidationError mallory = await _mk_person( db_session, email="m2@x.com", name="M2", department="Eng", ) other = await _mk_person( db_session, email="o2@x.com", name="O2", department="Eng", ) svc = FeedService(db_session) with pytest.raises(ValidationError): await svc.create_user_post( author=mallory, body="not mine", visibility="public", category="general", mentioned_person_ids=[], attachments=[{"s3_key": _key("feed_post_image", other.id)}], ) @pytest.mark.asyncio async def test_post_rejects_duplicate_attachment_upload_keys( db_session: AsyncSession, ) -> None: from app.core.exceptions import ValidationError alice = await _mk_person( db_session, email="duplicate@x.com", name="Duplicate", department="Eng", ) key = _key("feed_post_image", alice.id) with pytest.raises(ValidationError, match="duplicated"): await FeedService(db_session).create_user_post( author=alice, body="one object", visibility="public", category="general", mentioned_person_ids=[], attachments=[{"s3_key": key}, {"s3_key": key}], ) @pytest.mark.asyncio async def test_post_accepts_the_authors_own_feed_upload( db_session: AsyncSession, monkeypatch, ) -> None: """The gate must not break the normal path.""" from datetime import UTC, timedelta from app.integrations.models import StorageUploadGrant alice = await _mk_person( db_session, email="a3@x.com", name="A3", department="Eng", ) key = _key("feed_post_image", alice.id) db_session.add(StorageUploadGrant( person_id=alice.id, purpose="feed_post_image", object_key=key, content_type="image/jpeg", declared_size_bytes=42, checksum_sha256="47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=", status="issued", expires_at=datetime.now(UTC) + timedelta(minutes=10), )) await db_session.flush() monkeypatch.setattr( "app.core.storage_uploads.stat_object", AsyncMock(return_value={ "size_bytes": 42, "content_type": "image/jpeg", "checksum_sha256": "47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=", }), ) svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="my photo", visibility="public", category="general", mentioned_person_ids=[], attachments=[{ "s3_key": key, "size_bytes": 1, "mime_type": "image/png", }], ) assert post.id from app.domains.feed.models import FeedAttachment attachment = await db_session.scalar( select(FeedAttachment).where(FeedAttachment.post_id == post.id), ) assert attachment is not None and attachment.mime_type == "image/jpeg" @pytest.mark.asyncio async def test_legacy_foreign_key_is_never_presigned( db_session: AsyncSession, ) -> None: """Rows written before the gate existed must not be signed either. The create path is closed now, but the line that actually mints the credential is in hydration — and the table still holds whatever was accepted yesterday. """ from app.domains.feed.models import FeedAttachment alice = await _mk_person( db_session, email="a4@x.com", name="A4", department="Eng", ) svc = FeedService(db_session) post, _ = await svc.create_user_post( author=alice, body="legacy", visibility="public", category="general", mentioned_person_ids=[], attachments=[], ) # Simulate the pre-fix row: written straight into the table. db_session.add(FeedAttachment( post_id=post.id, s3_key=_key("inventory_evidence", alice.id), mime_type="image/png", position=0, )) await db_session.flush() [dto] = await svc.hydrate_posts([post], viewer=alice) assert dto["attachments"][0]["download_url"] is None, ( "hydration signed an object the create path would refuse today" )