"""Partner group listing and overlap participant resolution.""" from __future__ import annotations import os from pathlib import Path import pytest pytestmark = pytest.mark.usefixtures("tmp_path") @pytest.fixture def backend(tmp_path: Path): os.environ["KINK_SKIP_HEAVY_WARM"] = "1" from backend.core import Backend db = tmp_path / "t.db" b = Backend(db) b.warm_up_catalog() b._catalog() return b def test_member_sees_partner_owned_group_and_overlap_participants(backend): owner = backend.create_user() member = backend.create_user() assert backend.request_partner_link(owner.id, member.id)[0] assert backend.accept_partner_request(member.id, owner.id) grp = backend.create_partner_group(owner.id, "G", [member.id]) assert grp gid = grp["id"] member_groups = backend.list_partner_groups(member.id) assert any(g["id"] == gid for g in member_groups), "guest should see partner-owned group" pop_m = backend._group_participants(member.id, gid) assert {u["id"] for u in pop_m} == {owner.id, member.id} owner_only = backend.create_user() assert backend._group_participants(owner_only.id, gid) == [] def test_accepting_partner_link_creates_one_pair_group_seen_by_both(backend): left = backend.create_user() right = backend.create_user() assert backend.request_partner_link(left.id, right.id) == (True, "sent") assert backend.accept_partner_request(right.id, left.id) left_groups = backend.list_partner_groups(left.id) right_groups = backend.list_partner_groups(right.id) left_pair_groups = [ g for g in left_groups if g["name"] == "Together" and set(g["participant_ids"]) == {left.id, right.id} ] right_pair_groups = [ g for g in right_groups if g["name"] == "Together" and set(g["participant_ids"]) == {left.id, right.id} ] assert len(left_pair_groups) == 1 assert len(right_pair_groups) == 1 assert left_pair_groups[0]["id"] == right_pair_groups[0]["id"] def test_owner_share_row_is_visible_to_partner_not_owner(backend): owner = backend.create_user() member = backend.create_user() assert backend.request_partner_link(owner.id, member.id) == (True, "sent") assert backend.accept_partner_request(member.id, owner.id) group = backend.create_partner_group(owner.id, "Owner share", [member.id]) assert group assert backend.toggle_share_full_list(owner.id, group["id"], True) owner_view = next(g for g in backend.list_partner_groups(owner.id) if g["id"] == group["id"]) member_view = next(g for g in backend.list_partner_groups(member.id) if g["id"] == group["id"]) assert owner_view["my_share_full_list"] is True assert owner.id not in owner_view["sharing_member_ids"] assert owner.id in member_view["sharing_member_ids"] def test_owner_share_row_not_counted_as_group_member(backend): owner = backend.create_user() gid = backend.create_partner_group(owner.id, "Soloish", [])["id"] assert backend.toggle_share_full_list(owner.id, gid, True) g = next(x for x in backend.list_partner_groups(owner.id) if x["id"] == gid) assert g["member_ids"] == [] assert g["participant_ids"] == [owner.id] assert len(backend._group_participants(owner.id, gid)) == 1 def test_startup_migration_collapses_mirrored_together_groups(tmp_path: Path) -> None: os.environ["KINK_SKIP_HEAVY_WARM"] = "1" from backend.core import Backend db = tmp_path / "mirrored.db" first = Backend(db) left = first.create_user() right = first.create_user() assert first.request_partner_link(left.id, right.id) == (True, "sent") assert first.accept_partner_request(right.id, left.id) mirrored = first.create_partner_group(right.id, "Together", [left.id]) assert mirrored assert first.toggle_share_full_list(right.id, mirrored["id"], True) migrated = Backend(db) left_groups = [ group for group in migrated.list_partner_groups(left.id) if group["name"] == "Together" and set(group["participant_ids"]) == {left.id, right.id} ] right_groups = [ group for group in migrated.list_partner_groups(right.id) if group["name"] == "Together" and set(group["participant_ids"]) == {left.id, right.id} ] assert len(left_groups) == 1 assert len(right_groups) == 1 assert left_groups[0]["id"] == right_groups[0]["id"] assert right.id in left_groups[0]["sharing_member_ids"]