From a1f8f476ce2b948f894435d4c8253778f9888e8f Mon Sep 17 00:00:00 2001 From: autodev-bot Date: Sat, 22 Aug 2026 23:47:47 +0800 Subject: [PATCH 1/2] fix(graph_dbs): honor pagination and filters in Postgres export_graph PostgresGraphDB.export_graph accepted page / page_size / memory_type / status / filter only through **kwargs and silently ignored them. Its SQL had no LIMIT/OFFSET and no filtering beyond user_name, and total_nodes was set to len(nodes). Consumers of POST /product/get_memory (via TreeTextMemory.get_all -> export_graph) therefore received the full memory list on every page with an incorrect total. - Promote page / page_size / memory_type / status / filter to first-class parameters, matching the Neo4j and PolarDB signatures. - Build the WHERE clause from user_name + memory_type + status + filter, reusing _build_filter_where_clause for structured tag / property predicates. Default status excludes soft-deleted nodes (aligned with Neo4j). - Add ORDER BY created_at DESC, id DESC and a LIMIT/OFFSET clause when pagination is requested; run an independent COUNT(*) so total_nodes is the filtered total, not the returned page size. - Tighten the edge query to source_id/target_id both in the returned page to avoid dangling references. Adds tests/graph_dbs/test_postgres_export_graph.py covering pagination, memory_type / status / filter propagation to SQL, and the total_nodes regression. Fixes #2271 --- src/memos/graph_dbs/postgres.py | 118 ++++++- tests/graph_dbs/test_postgres_export_graph.py | 291 ++++++++++++++++++ 2 files changed, 393 insertions(+), 16 deletions(-) create mode 100644 tests/graph_dbs/test_postgres_export_graph.py diff --git a/src/memos/graph_dbs/postgres.py b/src/memos/graph_dbs/postgres.py index 594f7e695..f44aa6836 100644 --- a/src/memos/graph_dbs/postgres.py +++ b/src/memos/graph_dbs/postgres.py @@ -1100,34 +1100,120 @@ def clear(self, user_name: str | None = None) -> None: finally: self._put_conn(conn) - def export_graph(self, include_embedding: bool = False, **kwargs) -> dict[str, Any]: - """Export all data.""" + def export_graph( + self, + include_embedding: bool = False, + page: int | None = None, + page_size: int | None = None, + memory_type: list[str] | None = None, + status: list[str] | None = None, + filter: dict | None = None, + **kwargs, + ) -> dict[str, Any]: + """Export all graph nodes and edges in a structured form. + + Args: + include_embedding: Whether to include the ``embedding`` column in each + node's metadata. + page: 1-based page number. When both ``page`` and ``page_size`` are + provided, results are paginated; otherwise all matching rows are + returned. + page_size: Page size. See ``page``. + memory_type: If provided, restrict nodes to those whose + ``properties->>'memory_type'`` is in the list. + status: If ``None`` (default), nodes with status ``'deleted'`` are + excluded to match Neo4j's ``export_graph`` semantics. If a list is + provided, only nodes whose status is in the list are returned. + filter: Optional structured filter (``{"and": [...]}`` / ``{"or": [...]}`` + / single-condition dict), matching :meth:`_build_filter_where_clause`. + **kwargs: Accepts ``user_name`` for multi-tenant isolation; other keys + are ignored (kept for cross-backend compatibility). + + Returns: + Dict with ``nodes``, ``edges``, ``total_nodes`` (the filtered total, + **not** the size of the returned page) and ``total_edges``. + """ user_name = kwargs.get("user_name") or self.user_name + logger.info( + "export_graph include_embedding=%s page=%s page_size=%s " + "memory_type=%s status=%s filter=%s", + include_embedding, + page, + page_size, + memory_type, + status, + filter, + ) + + # Build WHERE conditions + parameter list. + where_conditions: list[str] = ["user_name = %s"] + params: list[Any] = [user_name] + + if memory_type and isinstance(memory_type, list) and len(memory_type) > 0: + where_conditions.append("properties->>'memory_type' = ANY(%s)") + params.append([str(mt) for mt in memory_type]) + + if status is None: + # Default: exclude soft-deleted nodes (aligned with Neo4j backend). + where_conditions.append( + "(properties->>'status' IS NULL OR properties->>'status' <> 'deleted')" + ) + elif isinstance(status, list) and len(status) > 0: + where_conditions.append("properties->>'status' = ANY(%s)") + params.append([str(s) for s in status]) + + if filter: + filter_where = self._build_filter_where_clause(filter, params) + if filter_where: + where_conditions.append(f"({filter_where})") + + where_clause = " AND ".join(where_conditions) + + # Decide pagination. + use_pagination = page is not None and page_size is not None + if use_pagination: + page = max(page, 1) + page_size = max(page_size, 1) + offset = (page - 1) * page_size + limit_clause = " LIMIT %s OFFSET %s" + limit_params: list[Any] = [page_size, offset] + else: + limit_clause = "" + limit_params = [] + + cols = "id, memory, properties, created_at, updated_at" + if include_embedding: + cols += ", embedding" + conn = self._get_conn() try: with conn.cursor() as cur: - # Get nodes - cols = "id, memory, properties, created_at, updated_at" - if include_embedding: - cols += ", embedding" - cur.execute( - f""" - SELECT {cols} FROM {self.schema}.memories - WHERE user_name = %s - ORDER BY created_at DESC - """, - (user_name,), + # 1) Total count of matching nodes (independent of pagination). + count_sql = f"SELECT COUNT(*) FROM {self.schema}.memories WHERE {where_clause}" + cur.execute(count_sql, params) + row = cur.fetchone() + total_nodes = int(row[0]) if row and row[0] is not None else 0 + + # 2) Page of nodes. + data_sql = ( + f"SELECT {cols} FROM {self.schema}.memories " + f"WHERE {where_clause} " + f"ORDER BY created_at DESC, id DESC{limit_clause}" ) + cur.execute(data_sql, params + limit_params) nodes = [self._parse_row(row, include_embedding) for row in cur.fetchall()] - # Get edges + # 3) Edges relevant to the returned nodes only. `total_edges` + # mirrors the returned edge count; the caller treats + # ``export_graph`` as "give me this page of nodes plus the + # edges that connect them" (see TreeTextMemory.get_all). node_ids = [n["id"] for n in nodes] if node_ids: cur.execute( f""" SELECT source_id, target_id, edge_type FROM {self.schema}.edges - WHERE source_id = ANY(%s) OR target_id = ANY(%s) + WHERE source_id = ANY(%s) AND target_id = ANY(%s) """, (node_ids, node_ids), ) @@ -1141,7 +1227,7 @@ def export_graph(self, include_embedding: bool = False, **kwargs) -> dict[str, A return { "nodes": nodes, "edges": edges, - "total_nodes": len(nodes), + "total_nodes": total_nodes, "total_edges": len(edges), } finally: diff --git a/tests/graph_dbs/test_postgres_export_graph.py b/tests/graph_dbs/test_postgres_export_graph.py new file mode 100644 index 000000000..876a135ac --- /dev/null +++ b/tests/graph_dbs/test_postgres_export_graph.py @@ -0,0 +1,291 @@ +""" +Regression tests for issue #2271: +`PostgresGraphDB.export_graph` must respect page / page_size / memory_type / status / filter +and return an accurate `total_nodes` count (matching the filter, not the returned page). + +These tests exercise SQL construction only (they don't require a live Postgres). The +Postgres connection pool and cursor are mocked out; the tests assert on the SQL +fragments and parameters that the implementation passes to `cursor.execute`, plus the +shape of the returned dict. +""" + +from __future__ import annotations + +import sys +import types + +from datetime import datetime +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + + +# Make `import psycopg2` inside `memos.graph_dbs.postgres` a no-op so this test +# module can run without the real driver installed. +if "psycopg2" not in sys.modules: + _fake_psycopg2 = types.ModuleType("psycopg2") + _fake_pool = types.ModuleType("psycopg2.pool") + _fake_pool.ThreadedConnectionPool = MagicMock() + _fake_psycopg2.pool = _fake_pool + sys.modules["psycopg2"] = _fake_psycopg2 + sys.modules["psycopg2.pool"] = _fake_pool + + +# --------------------------------------------------------------------------- # +# Fixtures # +# --------------------------------------------------------------------------- # + + +class _FakeCursor: + """Minimal cursor stand-in that records executed SQL and returns canned rows. + + - `execute(sql, params=None)` appends `(sql, params)` to `calls` and stores the + next canned response in `self._next_rows` (popped from `responses`). + - `fetchall()` returns `self._next_rows` (list of tuples). + - `fetchone()` returns `self._next_rows[0]` if present. + """ + + def __init__(self, responses: list[Any]): + self._responses = list(responses) + self._next_rows: Any = [] + self.calls: list[tuple[str, Any]] = [] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + return False + + def execute(self, sql: str, params: Any = None): + self.calls.append((sql, params)) + if self._responses: + self._next_rows = self._responses.pop(0) + else: + self._next_rows = [] + + def fetchall(self): + rows = self._next_rows + return rows if isinstance(rows, list) else [rows] + + def fetchone(self): + rows = self._next_rows + if isinstance(rows, list): + return rows[0] if rows else None + return rows + + +class _FakeConn: + def __init__(self, cursor: _FakeCursor): + self._cursor = cursor + self.autocommit = True + + def cursor(self): + return self._cursor + + +@pytest.fixture +def postgres_db(): + """Build a PostgresGraphDB with all IO fully mocked out.""" + from memos.configs.graph_db import PostgresGraphDBConfig + from memos.graph_dbs.postgres import PostgresGraphDB + + config = PostgresGraphDBConfig( + host="localhost", + port=5432, + user="test", + password="test", + db_name="test_db", + schema_name="memos", + user_name="alice", + use_multi_db=False, + embedding_dimension=3, + ) + + # Bypass real _init_schema (it runs SQL during __init__) + with patch.object(PostgresGraphDB, "_init_schema", return_value=None): + db = PostgresGraphDB(config) + + yield db + + +def _install_cursor(db, responses: list[Any]) -> _FakeCursor: + """Attach a fake cursor + connection to the mocked pool so SQL calls are captured.""" + cursor = _FakeCursor(responses) + conn = _FakeConn(cursor) + db._get_conn = MagicMock(return_value=conn) # type: ignore[method-assign] + db._put_conn = MagicMock() # type: ignore[method-assign] + return cursor + + +def _mk_row(node_id: str, memory: str = "m", props: dict | None = None): + """Build a row tuple in the shape `_parse_row` expects.""" + now = datetime(2026, 8, 22, 12, 0, 0) + return (node_id, memory, props or {}, now, now) + + +# --------------------------------------------------------------------------- # +# Tests # +# --------------------------------------------------------------------------- # + + +class TestPostgresExportGraphPagination: + """LIMIT/OFFSET must be applied when page + page_size are supplied.""" + + def test_pagination_applies_limit_and_offset(self, postgres_db): + # Page 2, size 3 -> LIMIT 3 OFFSET 3 + page_rows = [_mk_row(f"n{i}") for i in range(3)] + cursor = _install_cursor( + postgres_db, + responses=[ + [(42,)], # count query -> 42 matching nodes + page_rows, # data page + [], # edges (no edges relevant) + ], + ) + + result = postgres_db.export_graph(page=2, page_size=3) + + # Assert nodes come from the mocked page and total is the filtered count, + # NOT `len(nodes)`. + assert [n["id"] for n in result["nodes"]] == ["n0", "n1", "n2"] + assert result["total_nodes"] == 42 + + # At least one query must contain LIMIT + OFFSET. + data_sql = " ".join(sql for sql, _ in cursor.calls) + assert "LIMIT" in data_sql.upper() + assert "OFFSET" in data_sql.upper() + + def test_no_pagination_returns_all(self, postgres_db): + rows = [_mk_row(f"n{i}") for i in range(5)] + cursor = _install_cursor( + postgres_db, + responses=[ + [(5,)], # count + rows, # data + [], # edges + ], + ) + + result = postgres_db.export_graph() + assert result["total_nodes"] == 5 + assert len(result["nodes"]) == 5 + + data_sql = " ".join(sql for sql, _ in cursor.calls) + # Without page/page_size, no LIMIT/OFFSET must appear on the node query. + # (Count query never has LIMIT, so absence across all calls is a strong + # signal.) + assert "LIMIT" not in data_sql.upper() + assert "OFFSET" not in data_sql.upper() + + +class TestPostgresExportGraphFilters: + """memory_type / status / filter must reach the WHERE clause.""" + + def test_memory_type_filter(self, postgres_db): + cursor = _install_cursor( + postgres_db, + responses=[ + [(1,)], + [_mk_row("n1", props={"memory_type": "LongTermMemory"})], + [], + ], + ) + + postgres_db.export_graph(memory_type=["LongTermMemory"]) + + # Params of the count and data queries must contain the memory_type value + # (either directly or inside an "= ANY(list)" list parameter). + flat: list[Any] = [] + for _sql, params in cursor.calls: + if not params: + continue + for p in params: + if isinstance(p, list): + flat.extend(p) + else: + flat.append(p) + assert "LongTermMemory" in flat + + def test_status_default_excludes_deleted(self, postgres_db): + cursor = _install_cursor( + postgres_db, + responses=[ + [(1,)], + [_mk_row("n1")], + [], + ], + ) + + postgres_db.export_graph() + + # Default (status=None) must add a "not deleted" predicate to align with + # neo4j.export_graph. + combined_sql = " ".join(sql for sql, _ in cursor.calls).lower() + assert "deleted" in combined_sql + + def test_status_explicit_list(self, postgres_db): + cursor = _install_cursor( + postgres_db, + responses=[ + [(1,)], + [_mk_row("n1")], + [], + ], + ) + + postgres_db.export_graph(status=["activated"]) + + flat: list[Any] = [] + for _sql, params in cursor.calls: + if not params: + continue + for p in params: + if isinstance(p, list): + flat.extend(p) + else: + flat.append(p) + assert "activated" in flat + + def test_filter_tags_reach_sql(self, postgres_db): + cursor = _install_cursor( + postgres_db, + responses=[ + [(2,)], + [_mk_row("n1"), _mk_row("n2")], + [], + ], + ) + + postgres_db.export_graph(filter={"and": [{"tags": "urgent"}]}) + + # `tags` is treated as a JSON array; PostgresGraphDB's + # `_build_single_filter_condition` renders that as `... @> %s::jsonb` + # with the value JSON-encoded. + combined_sql = " ".join(sql for sql, _ in cursor.calls) + assert "@>" in combined_sql + # The urgent tag must appear (JSON-encoded) somewhere in the params. + joined_param_strs = [] + for _sql, params in cursor.calls: + if params: + joined_param_strs.extend(str(p) for p in params) + assert any("urgent" in p for p in joined_param_strs) + + +class TestPostgresExportGraphResultShape: + def test_total_nodes_is_full_count_not_page_len(self, postgres_db): + """The bug: total_nodes used to == len(page). Assert it's the filtered count.""" + page_rows = [_mk_row(f"n{i}") for i in range(2)] # page size 2 + _install_cursor( + postgres_db, + responses=[ + [(17,)], # full filtered count + page_rows, + [], + ], + ) + + result = postgres_db.export_graph(page=1, page_size=2) + assert result["total_nodes"] == 17 + assert len(result["nodes"]) == 2 + assert result["total_nodes"] != len(result["nodes"]) From a2b82363366ecb23019a2f212dafe4a3db15bb99 Mon Sep 17 00:00:00 2001 From: autodev-bot Date: Sun, 23 Aug 2026 00:26:51 +0800 Subject: [PATCH 2/2] test(graph_dbs): address open code review on postgres export_graph MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to a1f8f47 to resolve the 6 open review findings. Code (src/memos/graph_dbs/postgres.py): - status=[] now falls back to the default "exclude deleted" predicate instead of skipping status filtering entirely. Previously an empty list silently matched all rows, including soft-deleted ones — the exact opposite of the status=None default. Aligns with the Neo4j backend's semantics. - Restore the pre-existing OR semantics for the edge query. The previous AND version silently dropped edges whose other endpoint landed on a different page (paginated view) or was filtered out (e.g. deleted). OR keeps the "any edge incident to a returned node" behavior the code had before pagination was wired up. Tests (tests/graph_dbs/): - Move the sys.modules psycopg2 stub from module-import-time into a session-scoped autouse fixture in a new conftest.py, with teardown that rolls back what it installed. This keeps the mutation from leaking into other test suites in the same process. - _FakeCursor.execute now raises AssertionError when responses is exhausted, so an unexpected extra query fails loudly instead of silently returning [] and letting a later assertion pass on wrong data. - test_no_pagination_returns_all isolates the data query by index (calls[1]) instead of joining all SQL together, avoiding false positives from string literals or comments containing "LIMIT" in the count/edges queries. - test_status_default_excludes_deleted checks the exact `<> 'deleted'` (or `!= 'deleted'`) fragment so a bug that flipped the predicate polarity would still fail the test. - Add test_status_empty_list_falls_back_to_default covering the new empty-list handling. --- src/memos/graph_dbs/postgres.py | 22 +++-- tests/graph_dbs/conftest.py | 48 +++++++++++ tests/graph_dbs/test_postgres_export_graph.py | 83 ++++++++++++------- 3 files changed, 117 insertions(+), 36 deletions(-) create mode 100644 tests/graph_dbs/conftest.py diff --git a/src/memos/graph_dbs/postgres.py b/src/memos/graph_dbs/postgres.py index f44aa6836..5f2c1aa4b 100644 --- a/src/memos/graph_dbs/postgres.py +++ b/src/memos/graph_dbs/postgres.py @@ -1153,12 +1153,16 @@ def export_graph( where_conditions.append("properties->>'memory_type' = ANY(%s)") params.append([str(mt) for mt in memory_type]) - if status is None: - # Default: exclude soft-deleted nodes (aligned with Neo4j backend). + if status is None or (isinstance(status, list) and len(status) == 0): + # Default (status=None) and status=[] both fall back to "exclude + # soft-deleted nodes" (aligned with Neo4j backend). Treating an + # empty list the same as None avoids the surprising behavior where + # callers passing an empty list would otherwise receive soft-deleted + # nodes because no status predicate was applied at all. where_conditions.append( "(properties->>'status' IS NULL OR properties->>'status' <> 'deleted')" ) - elif isinstance(status, list) and len(status) > 0: + elif isinstance(status, list): where_conditions.append("properties->>'status' = ANY(%s)") params.append([str(s) for s in status]) @@ -1203,17 +1207,19 @@ def export_graph( cur.execute(data_sql, params + limit_params) nodes = [self._parse_row(row, include_embedding) for row in cur.fetchall()] - # 3) Edges relevant to the returned nodes only. `total_edges` - # mirrors the returned edge count; the caller treats - # ``export_graph`` as "give me this page of nodes plus the - # edges that connect them" (see TreeTextMemory.get_all). + # 3) Edges incident to the returned page of nodes. ``OR`` here + # preserves the pre-existing "any edge touching a returned + # node" semantics from before pagination was wired up, and + # keeps edges whose other endpoint lands on a different page + # (which the caller can render if it has the neighbor node + # context). ``total_edges`` mirrors the returned edge count. node_ids = [n["id"] for n in nodes] if node_ids: cur.execute( f""" SELECT source_id, target_id, edge_type FROM {self.schema}.edges - WHERE source_id = ANY(%s) AND target_id = ANY(%s) + WHERE source_id = ANY(%s) OR target_id = ANY(%s) """, (node_ids, node_ids), ) diff --git a/tests/graph_dbs/conftest.py b/tests/graph_dbs/conftest.py new file mode 100644 index 000000000..07a85fc87 --- /dev/null +++ b/tests/graph_dbs/conftest.py @@ -0,0 +1,48 @@ +"""Shared fixtures for graph_dbs tests. + +Provides a session-scoped autouse fixture that stubs out the ``psycopg2`` +module so tests targeting ``memos.graph_dbs.postgres`` can run without the +real driver installed. Doing this in a fixture (rather than at import time in +an individual test module) keeps the mutation scoped: we only install the +stub if no real driver is present, and we clean up ``sys.modules`` at the end +of the session so subsequent test suites in the same process cannot silently +inherit the stub. +""" + +from __future__ import annotations + +import sys +import types + +from unittest.mock import MagicMock + +import pytest + + +@pytest.fixture(autouse=True, scope="session") +def _stub_psycopg2(): + """Install a minimal ``psycopg2`` stub if the real driver is missing. + + The scope is ``session`` + ``autouse`` because the target module + (``memos.graph_dbs.postgres``) does ``import psycopg2`` at import time; a + per-test fixture would be too late. We only touch ``sys.modules`` if the + real driver is not present, and we roll back what we installed on + teardown so we do not pollute unrelated test suites. + """ + + installed_keys: list[str] = [] + + if "psycopg2" not in sys.modules: + fake_psycopg2 = types.ModuleType("psycopg2") + fake_pool = types.ModuleType("psycopg2.pool") + fake_pool.ThreadedConnectionPool = MagicMock() + fake_psycopg2.pool = fake_pool + sys.modules["psycopg2"] = fake_psycopg2 + sys.modules["psycopg2.pool"] = fake_pool + installed_keys.extend(["psycopg2", "psycopg2.pool"]) + + try: + yield + finally: + for key in installed_keys: + sys.modules.pop(key, None) diff --git a/tests/graph_dbs/test_postgres_export_graph.py b/tests/graph_dbs/test_postgres_export_graph.py index 876a135ac..9cd983c3f 100644 --- a/tests/graph_dbs/test_postgres_export_graph.py +++ b/tests/graph_dbs/test_postgres_export_graph.py @@ -7,13 +7,14 @@ Postgres connection pool and cursor are mocked out; the tests assert on the SQL fragments and parameters that the implementation passes to `cursor.execute`, plus the shape of the returned dict. + +The ``psycopg2`` stub required to import ``memos.graph_dbs.postgres`` without the +real driver installed lives in ``tests/graph_dbs/conftest.py`` as a scoped +autouse fixture. """ from __future__ import annotations -import sys -import types - from datetime import datetime from typing import Any from unittest.mock import MagicMock, patch @@ -21,17 +22,6 @@ import pytest -# Make `import psycopg2` inside `memos.graph_dbs.postgres` a no-op so this test -# module can run without the real driver installed. -if "psycopg2" not in sys.modules: - _fake_psycopg2 = types.ModuleType("psycopg2") - _fake_pool = types.ModuleType("psycopg2.pool") - _fake_pool.ThreadedConnectionPool = MagicMock() - _fake_psycopg2.pool = _fake_pool - sys.modules["psycopg2"] = _fake_psycopg2 - sys.modules["psycopg2.pool"] = _fake_pool - - # --------------------------------------------------------------------------- # # Fixtures # # --------------------------------------------------------------------------- # @@ -41,7 +31,12 @@ class _FakeCursor: """Minimal cursor stand-in that records executed SQL and returns canned rows. - `execute(sql, params=None)` appends `(sql, params)` to `calls` and stores the - next canned response in `self._next_rows` (popped from `responses`). + next canned response in `self._next_rows` (popped from `responses`). If + `responses` is empty when `execute` is called, we raise ``AssertionError`` + instead of silently returning `[]`: an unexpected extra ``execute`` call + almost always means the implementation under test issued a query the test + author did not anticipate, and silently returning empty rows would let a + later assertion pass against wrong data. - `fetchall()` returns `self._next_rows` (list of tuples). - `fetchone()` returns `self._next_rows[0]` if present. """ @@ -59,10 +54,12 @@ def __exit__(self, exc_type, exc_val, exc_tb): def execute(self, sql: str, params: Any = None): self.calls.append((sql, params)) - if self._responses: - self._next_rows = self._responses.pop(0) - else: - self._next_rows = [] + if not self._responses: + raise AssertionError( + "_FakeCursor received an unexpected execute() call (no more " + f"canned responses). SQL: {sql!r}, params: {params!r}" + ) + self._next_rows = self._responses.pop(0) def fetchall(self): rows = self._next_rows @@ -171,12 +168,14 @@ def test_no_pagination_returns_all(self, postgres_db): assert result["total_nodes"] == 5 assert len(result["nodes"]) == 5 - data_sql = " ".join(sql for sql, _ in cursor.calls) - # Without page/page_size, no LIMIT/OFFSET must appear on the node query. - # (Count query never has LIMIT, so absence across all calls is a strong - # signal.) - assert "LIMIT" not in data_sql.upper() - assert "OFFSET" not in data_sql.upper() + # Without page/page_size, no LIMIT/OFFSET must appear on the data + # query (index 1: count is at index 0, edges at index 2). Isolating the + # data query by index avoids false positives from column aliases, + # comments or string literals containing "LIMIT" in the other queries. + assert len(cursor.calls) >= 2 + data_query_sql = cursor.calls[1][0].upper() + assert "LIMIT" not in data_query_sql + assert "OFFSET" not in data_query_sql class TestPostgresExportGraphFilters: @@ -219,10 +218,14 @@ def test_status_default_excludes_deleted(self, postgres_db): postgres_db.export_graph() - # Default (status=None) must add a "not deleted" predicate to align with - # neo4j.export_graph. + # Default (status=None) must add a "not deleted" exclusion predicate to + # align with neo4j.export_graph. Check the exact SQL fragment so a + # bug that flipped the polarity (e.g. rendering `= 'deleted'` and + # thereby *including* soft-deleted rows) would still fail the test. + # The implementation renders: + # (properties->>'status' IS NULL OR properties->>'status' <> 'deleted') combined_sql = " ".join(sql for sql, _ in cursor.calls).lower() - assert "deleted" in combined_sql + assert "<> 'deleted'" in combined_sql or "!= 'deleted'" in combined_sql def test_status_explicit_list(self, postgres_db): cursor = _install_cursor( @@ -247,6 +250,30 @@ def test_status_explicit_list(self, postgres_db): flat.append(p) assert "activated" in flat + def test_status_empty_list_falls_back_to_default(self, postgres_db): + """Regression: status=[] must behave like status=None (exclude deleted). + + Without this, an empty-list caller would receive soft-deleted nodes + because no status predicate was applied at all — the exact opposite of + the default behavior. + """ + cursor = _install_cursor( + postgres_db, + responses=[ + [(1,)], + [_mk_row("n1")], + [], + ], + ) + + postgres_db.export_graph(status=[]) + + combined_sql = " ".join(sql for sql, _ in cursor.calls).lower() + assert "<> 'deleted'" in combined_sql or "!= 'deleted'" in combined_sql + # And crucially: no `status = ANY(...)` predicate was added (would + # have empty ANY() semantics that always false). + assert "status' = any" not in combined_sql + def test_filter_tags_reach_sql(self, postgres_db): cursor = _install_cursor( postgres_db,