diff --git a/backend/app/affiliate_tree_ingestion.py b/backend/app/affiliate_tree_ingestion.py index d073fc32d..08308c61d 100644 --- a/backend/app/affiliate_tree_ingestion.py +++ b/backend/app/affiliate_tree_ingestion.py @@ -3,6 +3,7 @@ from __future__ import annotations from typing import Any +from uuid import UUID import asyncpg @@ -15,25 +16,42 @@ async def fetch_affiliate_forest(conn: asyncpg.Connection, post_id: str) -> list[dict[str, Any]]: - """Ancestor forest of every organization this post's Keymen touch.""" - aliases = await fetch_corroborated_organization_aliases(conn) - entity_rows = await conn.fetch( - """ - select corporate_entity_id, parent_entity_id, entity_name, entity_level_code - from corporate_entity - """ + """Ancestor forest of only the organizations this post's Keymen touch. + + Read the post's stored affiliations without alias decoration first. Only + unresolved organization names from that post may participate in identity + resolution. After the bounded hierarchy is known, a second bounded alias + snapshot covers exactly those touched entity names so existing alias chips + remain available without loading the global organization-alias catalog. + """ + raw_keymen = await fetch_post_keymen(conn, post_id, organization_aliases=()) + unresolved_names = tuple( + sorted( + { + affiliation["organization_name"].strip() + for person in raw_keymen + for affiliation in person["affiliations"] + if affiliation["corporate_entity_id"] is None + and affiliation["organization_name"].strip() + } + ) ) - entities = tuple( - CorporateEntityRow( - entity_id=str(row["corporate_entity_id"]), - parent_entity_id=str(row["parent_entity_id"]) if row["parent_entity_id"] is not None else None, - entity_name=row["entity_name"], - entity_level_code=row["entity_level_code"], + resolution_aliases = ( + await fetch_corroborated_organization_aliases( + conn, + organization_names=unresolved_names, ) - for row in entity_rows + if unresolved_names + else () + ) + keymen = ( + await fetch_post_keymen(conn, post_id, organization_aliases=resolution_aliases) + if resolution_aliases + else raw_keymen ) + leaves: list[AffiliationLeaf] = [] - for person in await fetch_post_keymen(conn, post_id, organization_aliases=aliases): + for person in keymen: for affiliation in person["affiliations"]: leaves.append( AffiliationLeaf( @@ -44,11 +62,75 @@ async def fetch_affiliate_forest(conn: asyncpg.Connection, post_id: str) -> list corporate_entity_id=affiliation["corporate_entity_id"], ) ) + + resolved_entity_ids = sorted( + { + UUID(leaf.corporate_entity_id) + for leaf in leaves + if leaf.corporate_entity_id is not None + }, + key=str, + ) + entity_rows = [] + if resolved_entity_ids: + entity_rows = await conn.fetch( + """ + with recursive affiliate_entity as ( + select corporate_entity_id, parent_entity_id, entity_name, entity_level_code + from corporate_entity + where corporate_entity_id = any($1::uuid[]) + + union + + select parent.corporate_entity_id, + parent.parent_entity_id, + parent.entity_name, + parent.entity_level_code + from corporate_entity parent + join affiliate_entity child + on child.parent_entity_id = parent.corporate_entity_id + ) + select corporate_entity_id, parent_entity_id, entity_name, entity_level_code + from affiliate_entity + order by entity_name, corporate_entity_id + """, + resolved_entity_ids, + ) + entities = tuple( + CorporateEntityRow( + entity_id=str(row["corporate_entity_id"]), + parent_entity_id=str(row["parent_entity_id"]) if row["parent_entity_id"] is not None else None, + entity_name=row["entity_name"], + entity_level_code=row["entity_level_code"], + ) + for row in entity_rows + ) + + display_alias_names = tuple( + sorted( + set(unresolved_names) + | { + row["entity_name"].strip() + for row in entity_rows + if row["entity_name"].strip() + } + ) + ) + if not display_alias_names: + display_aliases = () + elif display_alias_names == unresolved_names: + display_aliases = resolution_aliases + else: + display_aliases = await fetch_corroborated_organization_aliases( + conn, + organization_names=display_alias_names, + ) + forest = [node.to_dict() for node in build_affiliate_forest(entities, tuple(leaves))] await _attach_lookup_labels(conn, forest) attach_organization_aliases( forest, - aliases, + display_aliases, entity_id_key="entity_id", ) return forest diff --git a/backend/app/organization_name_resolution_ingestion.py b/backend/app/organization_name_resolution_ingestion.py index daf825758..6797f18e6 100644 --- a/backend/app/organization_name_resolution_ingestion.py +++ b/backend/app/organization_name_resolution_ingestion.py @@ -110,31 +110,63 @@ async def resolve_organization_name( async def fetch_corroborated_organization_aliases( conn: asyncpg.Connection, + *, + organization_names: tuple[str, ...] | None = None, ) -> tuple[OrganizationNameAlias, ...]: - """Load corroborated pairs with a unique current catalog target, if any. + """Load corroborated aliases, optionally bounded to observed names. - Pending and uncorroborated rows stay out. The statement is a static - literal; only the status code is bound. Same-named catalog rows fail + ``organization_names`` narrows the resolution rows and catalog-name join + before they enter application memory. Callers that need the complete + corroborated alias catalog may omit it and retain the existing behavior. + Pending and uncorroborated rows stay out, and same-named catalog rows fail closed with a null target id. """ - rows = await conn.fetch( - """ - select resolution.raw_organization_name, - resolution.resolved_organization_name, - case when count(distinct entity.corporate_entity_id) = 1 - then min(entity.corporate_entity_id::text) - else null - end as corporate_entity_id - from organization_name_resolution as resolution - left join corporate_entity as entity - on entity.entity_name = resolution.raw_organization_name - or entity.entity_name = resolution.resolved_organization_name - where resolution.verification_status_code = $1 - group by resolution.raw_organization_name, - resolution.resolved_organization_name - """, - STATUS_CORROBORATED, - ) + if organization_names is None: + rows = await conn.fetch( + """ + select resolution.raw_organization_name, + resolution.resolved_organization_name, + case when count(distinct entity.corporate_entity_id) = 1 + then min(entity.corporate_entity_id::text) + else null + end as corporate_entity_id + from organization_name_resolution as resolution + left join corporate_entity as entity + on entity.entity_name = resolution.raw_organization_name + or entity.entity_name = resolution.resolved_organization_name + where resolution.verification_status_code = $1 + group by resolution.raw_organization_name, + resolution.resolved_organization_name + """, + STATUS_CORROBORATED, + ) + else: + names = sorted({name.strip() for name in organization_names if name.strip()}) + if not names: + return () + rows = await conn.fetch( + """ + select resolution.raw_organization_name, + resolution.resolved_organization_name, + case when count(distinct entity.corporate_entity_id) = 1 + then min(entity.corporate_entity_id::text) + else null + end as corporate_entity_id + from organization_name_resolution as resolution + left join corporate_entity as entity + on entity.entity_name = resolution.raw_organization_name + or entity.entity_name = resolution.resolved_organization_name + where resolution.verification_status_code = $1 + and ( + resolution.raw_organization_name = any($2::text[]) + or resolution.resolved_organization_name = any($2::text[]) + ) + group by resolution.raw_organization_name, + resolution.resolved_organization_name + """, + STATUS_CORROBORATED, + names, + ) return tuple( OrganizationNameAlias( alt_label=row["raw_organization_name"], diff --git a/lineageweave/affiliate_tree.py b/lineageweave/affiliate_tree.py index 2f3dcdeb6..69362061b 100644 --- a/lineageweave/affiliate_tree.py +++ b/lineageweave/affiliate_tree.py @@ -49,7 +49,15 @@ class AffiliatePerson: @dataclass(frozen=True) class AffiliateNode: - """One organization in the rendered forest, with people and children.""" + """One organization in the rendered forest, with people and children. + + ``resolved`` says whether the source entity reference was backed by an + available ``corporate_entity`` row. An unresolved node may still retain + the source ``entity_id`` so two unavailable references with the same + display name are not silently collapsed. ``hierarchy_issue`` is present + only when an available entity's parent pointer could not safely become a + tree edge. + """ entity_id: str | None entity_name: str @@ -57,10 +65,11 @@ class AffiliateNode: resolved: bool people: tuple[AffiliatePerson, ...] children: tuple["AffiliateNode", ...] + hierarchy_issue: str | None = None def to_dict(self) -> dict: - """JSON shape the product API and React panel consume.""" - return { + """Return the JSON shape consumed by the product API and React panel.""" + payload = { "entity_id": self.entity_id, "entity_name": self.entity_name, "entity_level_code": self.entity_level_code, @@ -75,6 +84,9 @@ def to_dict(self) -> dict: ], "children": [child.to_dict() for child in self.children], } + if self.hierarchy_issue is not None: + payload["hierarchy_issue"] = self.hierarchy_issue + return payload def _people_for(affiliations: tuple[AffiliationLeaf, ...]) -> tuple[AffiliatePerson, ...]: @@ -89,15 +101,40 @@ def _people_for(affiliations: tuple[AffiliationLeaf, ...]) -> tuple[AffiliatePer return tuple(sorted(unique.values(), key=lambda person: (person.person_name, person.person_id))) +def _index_entities( + entities: tuple[CorporateEntityRow, ...] | list[CorporateEntityRow], +) -> dict[str, CorporateEntityRow]: + """Index canonical entity rows without letting input order select truth. + + Duplicate ``entity_id`` rows make parent pointers and display attributes + ambiguous. Silently retaining the last row would make database/input order + choose which corporate entity representation is rendered, so the boundary + fails closed before any hierarchy is materialized. + """ + indexed: dict[str, CorporateEntityRow] = {} + for row in entities: + if row.entity_id in indexed: + raise ValueError(f"duplicate corporate entity id: {row.entity_id}") + indexed[row.entity_id] = row + return indexed + + def _needed_entity_ids( entities: dict[str, CorporateEntityRow], leaf_ids: set[str], ) -> set[str]: - """Every ancestor of a resolved leaf, walking ``parent_entity_id``.""" + """Return every available ancestor of a resolved affiliation leaf. + + The local ``seen`` set makes malformed cycles finite without letting one + leaf suppress the ancestor walk for another leaf. Missing parents stop the + walk; they are disclosed later instead of being invented as entity rows. + """ needed: set[str] = set() for leaf_id in leaf_ids: current = leaf_id - while current and current not in needed: + seen: set[str] = set() + while current and current not in seen: + seen.add(current) row = entities.get(current) if row is None: break @@ -106,19 +143,83 @@ def _needed_entity_ids( return needed +def _entity_sort_key(entities: dict[str, CorporateEntityRow], entity_id: str) -> tuple[str, str]: + """Return the stable buyer-visible order used for roots and cycle breaks.""" + row = entities[entity_id] + return (row.entity_name, row.entity_id) + + +def _safe_parent_links( + entities: dict[str, CorporateEntityRow], + needed: set[str], +) -> tuple[dict[str, str | None], dict[str, str]]: + """Convert parent pointers into an acyclic forest without hiding defects. + + Self-parent pointers and unavailable parents become roots immediately. For + a longer directed parent cycle, exactly one edge is ignored: the + lexicographically first entity by ``(entity_name, entity_id)`` becomes the + disclosed root. This preserves every authorized entity while making the + result independent of database or input iteration order. + """ + parent_by_id: dict[str, str | None] = {} + issue_by_id: dict[str, str] = {} + + for entity_id in needed: + parent_id = entities[entity_id].parent_entity_id + if parent_id == entity_id: + parent_by_id[entity_id] = None + issue_by_id[entity_id] = "self_parent_ignored" + elif parent_id and parent_id not in entities: + parent_by_id[entity_id] = None + issue_by_id[entity_id] = "parent_not_available" + elif parent_id and parent_id in needed: + parent_by_id[entity_id] = parent_id + else: + parent_by_id[entity_id] = None + + finalized: set[str] = set() + for start_id in sorted(needed, key=lambda entity_id: _entity_sort_key(entities, entity_id)): + if start_id in finalized: + continue + path: list[str] = [] + path_index: dict[str, int] = {} + current: str | None = start_id + while current is not None and current not in finalized: + cycle_start = path_index.get(current) + if cycle_start is not None: + cycle = path[cycle_start:] + cycle_root = min(cycle, key=lambda entity_id: _entity_sort_key(entities, entity_id)) + parent_by_id[cycle_root] = None + issue_by_id[cycle_root] = "cycle_parent_ignored" + break + path_index[current] = len(path) + path.append(current) + current = parent_by_id[current] + finalized.update(path) + + return parent_by_id, issue_by_id + + def build_affiliate_forest( entities: tuple[CorporateEntityRow, ...] | list[CorporateEntityRow], affiliations: tuple[AffiliationLeaf, ...] | list[AffiliationLeaf], ) -> tuple[AffiliateNode, ...]: - """Ancestor forest covering every affiliation on a post. + """Build the ancestor forest covering every affiliation on a post. - Resolved leaves pull in their parents. An entity that is neither a - leaf nor an ancestor of one is omitted. Unresolved organization - names become extra roots with ``resolved=False``. + Resolved leaves pull in their available parents. An entity that is neither + a leaf nor an ancestor of one is omitted. Unresolved organization + observations become roots with ``resolved=False``; an unavailable source + entity reference remains attached to that root instead of being erased. + Malformed parent pointers never make an otherwise authorized affiliation + disappear: the unsafe edge is omitted deterministically and the affected + root carries ``hierarchy_issue``. Conflicting duplicate canonical entity + identities fail closed rather than allowing input order to select a row. """ - entity_by_id = {row.entity_id: row for row in entities} + entity_by_id = _index_entities(entities) resolved_leaves = [ - leaf for leaf in affiliations if leaf.corporate_entity_id and leaf.corporate_entity_id in entity_by_id + leaf + for leaf in affiliations + if leaf.corporate_entity_id and leaf.corporate_entity_id in entity_by_id ] needed = _needed_entity_ids( entity_by_id, @@ -132,16 +233,15 @@ def build_affiliate_forest( continue people_by_entity.setdefault(entity_id, []).append(leaf) + parent_by_id, hierarchy_issue_by_id = _safe_parent_links(entity_by_id, needed) children_of: dict[str | None, list[str]] = {} for entity_id in needed: - parent_id = entity_by_id[entity_id].parent_entity_id - root_parent = parent_id if parent_id in needed else None - children_of.setdefault(root_parent, []).append(entity_id) + children_of.setdefault(parent_by_id[entity_id], []).append(entity_id) for child_ids in children_of.values(): - child_ids.sort(key=lambda entity_id: (entity_by_id[entity_id].entity_name, entity_id)) + child_ids.sort(key=lambda entity_id: _entity_sort_key(entity_by_id, entity_id)) def _build(entity_id: str) -> AffiliateNode: - """Implement the _build operation for this channel.""" + """Materialize one already-cycle-safe hierarchy node recursively.""" row = entity_by_id[entity_id] return AffiliateNode( entity_id=row.entity_id, @@ -150,28 +250,32 @@ def _build(entity_id: str) -> AffiliateNode: resolved=True, people=_people_for(tuple(people_by_entity.get(entity_id, ()))), children=tuple(_build(child_id) for child_id in children_of.get(entity_id, ())), + hierarchy_issue=hierarchy_issue_by_id.get(entity_id), ) resolved_roots = tuple(_build(entity_id) for entity_id in children_of.get(None, ())) - unresolved_by_name: dict[str, list[AffiliationLeaf]] = {} + unresolved_by_reference: dict[tuple[str | None, str], list[AffiliationLeaf]] = {} for leaf in affiliations: if leaf.corporate_entity_id and leaf.corporate_entity_id in entity_by_id: continue name = leaf.organization_name.strip() if not name: continue - unresolved_by_name.setdefault(name, []).append(leaf) + unresolved_by_reference.setdefault((leaf.corporate_entity_id, name), []).append(leaf) unresolved_roots = tuple( AffiliateNode( - entity_id=None, + entity_id=entity_id, entity_name=name, entity_level_code=None, resolved=False, people=_people_for(tuple(leaves)), children=(), ) - for name, leaves in sorted(unresolved_by_name.items()) + for (entity_id, name), leaves in sorted( + unresolved_by_reference.items(), + key=lambda item: (item[0][1], item[0][0] or ""), + ) ) return resolved_roots + unresolved_roots diff --git a/tests/test_affiliate_tree.py b/tests/test_affiliate_tree.py index 7a88fbb81..436ececd5 100644 --- a/tests/test_affiliate_tree.py +++ b/tests/test_affiliate_tree.py @@ -3,7 +3,8 @@ from __future__ import annotations import asyncio -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, call +from uuid import UUID import backend.app.affiliate_tree_ingestion as ingestion from lineageweave.affiliate_tree import ( @@ -112,24 +113,160 @@ def test_to_dict_is_the_api_shape() -> None: assert payload["children"][0]["people"][0]["person_name"] == "Ada West" -def test_affiliate_forest_reuses_one_corroborated_alias_load(monkeypatch) -> None: - """One request shares its alias snapshot with Keyman and forest hydration.""" - aliases = (OrganizationNameAlias("DC", "Demo Corp", "demo-id"),) +def test_affiliate_forest_skips_alias_catalog_when_no_keymen(monkeypatch) -> None: + """A post with no affiliations must not load the organization alias catalog.""" class _Connection: - async def fetch(self, query: str, *_args: object): - assert "from corporate_entity" in query - return [] + async def fetch(self, _query: str, *_args: object): + raise AssertionError("no corporate hierarchy query is needed without resolved affiliations") conn = _Connection() - fetch_aliases = AsyncMock(return_value=aliases) + fetch_aliases = AsyncMock(return_value=()) fetch_keymen = AsyncMock(return_value=[]) monkeypatch.setattr(ingestion, "fetch_corroborated_organization_aliases", fetch_aliases) monkeypatch.setattr(ingestion, "fetch_post_keymen", fetch_keymen) assert asyncio.run(ingestion.fetch_affiliate_forest(conn, "post-1")) == [] - fetch_aliases.assert_awaited_once_with(conn) - fetch_keymen.assert_awaited_once_with(conn, "post-1", organization_aliases=aliases) + fetch_aliases.assert_not_awaited() + fetch_keymen.assert_awaited_once_with(conn, "post-1", organization_aliases=()) + + +def test_affiliate_forest_bounds_alias_lookup_to_unresolved_names(monkeypatch) -> None: + """Resolution and display aliases stay bounded to this post's names.""" + demo_id = UUID("00000000-0000-0000-0000-000000000002") + aliases = (OrganizationNameAlias("Demo Co", "Demo Corp", str(demo_id)),) + raw_keymen = [ + { + "person_id": "ada", + "person_name": "Ada West", + "person_side_code": "our_side", + "affiliations": [ + { + "organization_name": "Demo Co", + "corporate_entity_id": None, + } + ], + } + ] + resolved_keymen = [ + { + "person_id": "ada", + "person_name": "Ada West", + "person_side_code": "our_side", + "affiliations": [ + { + "organization_name": "Demo Corp", + "corporate_entity_id": str(demo_id), + } + ], + } + ] + + class _Connection: + async def fetch(self, query: str, *_args: object): + assert "with recursive affiliate_entity" in query.lower() + return [ + { + "corporate_entity_id": demo_id, + "parent_entity_id": None, + "entity_name": "Demo Corp", + "entity_level_code": "company", + } + ] + + conn = _Connection() + fetch_aliases = AsyncMock(return_value=aliases) + fetch_keymen = AsyncMock(side_effect=[raw_keymen, resolved_keymen]) + attach_labels = AsyncMock() + monkeypatch.setattr(ingestion, "fetch_corroborated_organization_aliases", fetch_aliases) + monkeypatch.setattr(ingestion, "fetch_post_keymen", fetch_keymen) + monkeypatch.setattr(ingestion, "_attach_lookup_labels", attach_labels) + + forest = asyncio.run(ingestion.fetch_affiliate_forest(conn, "post-1")) + + assert fetch_aliases.await_args_list == [ + call(conn, organization_names=("Demo Co",)), + call(conn, organization_names=("Demo Co", "Demo Corp")), + ] + assert fetch_keymen.await_args_list == [ + call(conn, "post-1", organization_aliases=()), + call(conn, "post-1", organization_aliases=aliases), + ] + assert forest[0]["entity_name"] == "Demo Corp" + + +def test_affiliate_forest_loads_only_resolved_affiliation_ancestor_closure(monkeypatch) -> None: + """Hierarchy and alias reads both stay inside the touched organization closure.""" + group_id = UUID("00000000-0000-0000-0000-000000000001") + company_id = UUID("00000000-0000-0000-0000-000000000002") + unrelated_id = UUID("00000000-0000-0000-0000-000000000003") + observed_calls: list[tuple[str, tuple[object, ...]]] = [] + + class _Connection: + async def fetch(self, query: str, *args: object): + observed_calls.append((query, args)) + return [ + { + "corporate_entity_id": group_id, + "parent_entity_id": None, + "entity_name": "Demo Group", + "entity_level_code": "group", + }, + { + "corporate_entity_id": company_id, + "parent_entity_id": group_id, + "entity_name": "Demo Electronics Korea", + "entity_level_code": "company", + }, + ] + + fetch_aliases = AsyncMock(return_value=()) + fetch_keymen = AsyncMock( + return_value=[ + { + "person_id": "ada", + "person_name": "Ada West", + "person_side_code": "our_side", + "affiliations": [ + { + "organization_name": "Demo Electronics Korea", + "corporate_entity_id": str(company_id), + }, + { + "organization_name": "Unresolved Supplier", + "corporate_entity_id": None, + }, + ], + } + ] + ) + attach_labels = AsyncMock() + monkeypatch.setattr(ingestion, "fetch_corroborated_organization_aliases", fetch_aliases) + monkeypatch.setattr(ingestion, "fetch_post_keymen", fetch_keymen) + monkeypatch.setattr(ingestion, "_attach_lookup_labels", attach_labels) + + conn = _Connection() + forest = asyncio.run(ingestion.fetch_affiliate_forest(conn, "post-1")) + + assert fetch_aliases.await_args_list == [ + call(conn, organization_names=("Unresolved Supplier",)), + call( + conn, + organization_names=( + "Demo Electronics Korea", + "Demo Group", + "Unresolved Supplier", + ), + ), + ] + assert len(observed_calls) == 1 + query, args = observed_calls[0] + assert "with recursive affiliate_entity" in query.lower() + assert "corporate_entity_id = any($1::uuid[])" in query.lower() + assert args == ([company_id],) + assert unrelated_id not in args[0] + assert [node["entity_name"] for node in forest] == ["Demo Group", "Unresolved Supplier"] + attach_labels.assert_awaited_once() def test_voc_evidence_skips_unused_organization_aliases(monkeypatch) -> None: diff --git a/tests/test_affiliate_tree_alias_scope.py b/tests/test_affiliate_tree_alias_scope.py new file mode 100644 index 000000000..35ca41521 --- /dev/null +++ b/tests/test_affiliate_tree_alias_scope.py @@ -0,0 +1,74 @@ +"""Bounded alias behavior for the buyer-facing Affiliate Tree.""" + +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, Mock +from uuid import UUID + +import backend.app.affiliate_tree_ingestion as ingestion +from lineageweave.organization_alias import OrganizationNameAlias + + +def test_loaded_hierarchy_keeps_relevant_aliases_without_global_alias_scan(monkeypatch) -> None: + """Touched resolved entities retain aliases selected only by their names.""" + group_id = UUID("00000000-0000-0000-0000-000000000001") + company_id = UUID("00000000-0000-0000-0000-000000000002") + keymen = [ + { + "person_id": "ada", + "person_name": "Ada West", + "person_side_code": "our_side", + "affiliations": [ + { + "organization_name": "Demo Corp", + "corporate_entity_id": str(company_id), + } + ], + } + ] + relevant_aliases = ( + OrganizationNameAlias("DG", "Demo Group", str(group_id)), + OrganizationNameAlias("DC", "Demo Corp", str(company_id)), + ) + + class _Connection: + async def fetch(self, query: str, *_args: object): + assert "with recursive affiliate_entity" in query.lower() + return [ + { + "corporate_entity_id": group_id, + "parent_entity_id": None, + "entity_name": "Demo Group", + "entity_level_code": "group", + }, + { + "corporate_entity_id": company_id, + "parent_entity_id": group_id, + "entity_name": "Demo Corp", + "entity_level_code": "company", + }, + ] + + conn = _Connection() + fetch_keymen = AsyncMock(return_value=keymen) + fetch_aliases = AsyncMock(return_value=relevant_aliases) + attach_labels = AsyncMock() + attach_aliases = Mock() + monkeypatch.setattr(ingestion, "fetch_post_keymen", fetch_keymen) + monkeypatch.setattr(ingestion, "fetch_corroborated_organization_aliases", fetch_aliases) + monkeypatch.setattr(ingestion, "_attach_lookup_labels", attach_labels) + monkeypatch.setattr(ingestion, "attach_organization_aliases", attach_aliases) + + forest = asyncio.run(ingestion.fetch_affiliate_forest(conn, "post-1")) + + fetch_keymen.assert_awaited_once_with(conn, "post-1", organization_aliases=()) + fetch_aliases.assert_awaited_once_with( + conn, + organization_names=("Demo Corp", "Demo Group"), + ) + attach_aliases.assert_called_once_with( + forest, + relevant_aliases, + entity_id_key="entity_id", + ) diff --git a/tests/test_affiliate_tree_malformed_hierarchy.py b/tests/test_affiliate_tree_malformed_hierarchy.py new file mode 100644 index 000000000..9e4798465 --- /dev/null +++ b/tests/test_affiliate_tree_malformed_hierarchy.py @@ -0,0 +1,83 @@ +"""Malformed corporate hierarchy must remain visible instead of disappearing.""" + +from __future__ import annotations + +import pytest + +from lineageweave.affiliate_tree import AffiliationLeaf, CorporateEntityRow, build_affiliate_forest + + +def _leaf(entity_id: str, organization_name: str) -> AffiliationLeaf: + return AffiliationLeaf( + person_id="person-1", + person_name="Ada West", + person_side_code="our_side", + organization_name=organization_name, + corporate_entity_id=entity_id, + ) + + +def test_pure_parent_cycle_becomes_a_deterministic_disclosed_forest() -> None: + entities = ( + CorporateEntityRow("alpha-id", "beta-id", "Alpha Corp", "company"), + CorporateEntityRow("beta-id", "alpha-id", "Beta Corp", "company"), + ) + + forward = build_affiliate_forest(entities, (_leaf("beta-id", "Beta Corp"),)) + reversed_input = build_affiliate_forest(tuple(reversed(entities)), (_leaf("beta-id", "Beta Corp"),)) + + assert [node.to_dict() for node in forward] == [node.to_dict() for node in reversed_input] + assert len(forward) == 1 + assert forward[0].entity_id == "alpha-id" + assert forward[0].hierarchy_issue == "cycle_parent_ignored" + assert [child.entity_id for child in forward[0].children] == ["beta-id"] + assert forward[0].children[0].people[0].person_id == "person-1" + + +def test_self_parent_is_kept_as_a_root_and_disclosed() -> None: + forest = build_affiliate_forest( + (CorporateEntityRow("solo-id", "solo-id", "Solo Corp", "company"),), + (_leaf("solo-id", "Solo Corp"),), + ) + + assert len(forest) == 1 + assert forest[0].entity_id == "solo-id" + assert forest[0].hierarchy_issue == "self_parent_ignored" + assert forest[0].children == () + + +def test_missing_parent_is_kept_and_disclosed_without_inventing_an_edge() -> None: + forest = build_affiliate_forest( + (CorporateEntityRow("child-id", "missing-parent", "Visible Child", "company"),), + (_leaf("child-id", "Visible Child"),), + ) + + assert len(forest) == 1 + assert forest[0].entity_id == "child-id" + assert forest[0].hierarchy_issue == "parent_not_available" + assert forest[0].children == () + + +def test_duplicate_entity_identity_fails_closed_instead_of_using_input_order() -> None: + entities = ( + CorporateEntityRow("shared-id", None, "Original Corp", "company"), + CorporateEntityRow("shared-id", None, "Conflicting Corp", "company"), + ) + + with pytest.raises(ValueError, match="duplicate corporate entity id: shared-id"): + build_affiliate_forest(entities, (_leaf("shared-id", "Original Corp"),)) + + +def test_unavailable_entity_references_do_not_collapse_by_display_name() -> None: + forest = build_affiliate_forest( + (), + ( + _leaf("missing-alpha", "Shared Display Name"), + _leaf("missing-beta", "Shared Display Name"), + ), + ) + + assert [(node.entity_id, node.entity_name, node.resolved) for node in forest] == [ + ("missing-alpha", "Shared Display Name", False), + ("missing-beta", "Shared Display Name", False), + ] diff --git a/tests/test_organization_name_resolution_ingestion.py b/tests/test_organization_name_resolution_ingestion.py index 4f21121ac..1b8e4bc7f 100644 --- a/tests/test_organization_name_resolution_ingestion.py +++ b/tests/test_organization_name_resolution_ingestion.py @@ -149,3 +149,31 @@ def test_fetch_corroborated_aliases_keeps_catalog_ties_unbound() -> None: ) aliases = asyncio.run(ingestion.fetch_corroborated_organization_aliases(conn)) assert aliases == (OrganizationNameAlias("DC", "Demo Corp", None),) + + +class _BoundedAliasConnection: + def __init__(self) -> None: + self.query = "" + self.args: tuple[object, ...] = () + + async def fetch(self, query: str, *args: object): + self.query = query + self.args = args + return [] + + +def test_fetch_corroborated_aliases_bounds_rows_before_catalog_join() -> None: + conn = _BoundedAliasConnection() + + aliases = asyncio.run( + ingestion.fetch_corroborated_organization_aliases( + conn, + organization_names=(" Demo Co ", "Demo Corp", "Demo Co"), + ) + ) + + assert aliases == () + assert conn.args == (STATUS_CORROBORATED, ["Demo Co", "Demo Corp"]) + lowered = conn.query.lower() + assert "raw_organization_name = any($2::text[])" in lowered + assert "resolved_organization_name = any($2::text[])" in lowered