Stabilize pagination, search, and user-status integration tests on Postgres

Author: oleksandr-nc

scripts/seed_pagination_data.py

@@ -12,6 +12,7 @@
 """
 
 import sys
+import time
 import xml.etree.ElementTree as ET
 
 import niquests
@@ -172,6 +173,39 @@ def seed_contacts(s: niquests.Session, url: str, user: str) -> None:
     print(f"  {COUNT} contacts with categories and varied name structures")
 
 
+def _fetch_all_comments(s: niquests.Session, url: str, file_id: str) -> list[tuple[str, str, str]]:
+    """Return all comments on file_id as list of (id, message, creation_ts)."""
+    out: list[tuple[str, str, str]] = []
+    offset = 0
+    while True:
+        report = s.request(
+            "REPORT",
+            f"{url}/remote.php/dav/comments/files/{file_id}",
+            data=(
+                '<?xml version="1.0" encoding="utf-8"?>'
+                '<oc:filter-comments xmlns:oc="http://owncloud.org/ns">'
+                f"<oc:limit>100</oc:limit><oc:offset>{offset}</oc:offset>"
+                "</oc:filter-comments>"
+            ),
+            headers={"Content-Type": "application/xml"},
+        )
+        batch: list[tuple[str, str, str]] = []
+        for resp_el in ET.fromstring(report.text).findall("{DAV:}response"):
+            href = resp_el.find("{DAV:}href")
+            msg_el = resp_el.find(".//{http://owncloud.org/ns}message")
+            ts_el = resp_el.find(".//{http://owncloud.org/ns}creationDateTime")
+            if href is not None and msg_el is not None and msg_el.text and ts_el is not None:
+                cid = href.text.rstrip("/").split("/")[-1]
+                batch.append((cid, msg_el.text, ts_el.text or ""))
+        if not batch:
+            break
+        out.extend(batch)
+        if len(batch) < 100:
+            break
+        offset += 100
+    return out
+
+
 def seed_comments(s: niquests.Session, url: str, user: str) -> None:
     """Create a dedicated file and add many comments to it."""
     dav = f"{url}/remote.php/dav/files/{user}"
@@ -196,13 +230,25 @@ def seed_comments(s: niquests.Session, url: str, user: str) -> None:
         return
     file_id = fileid_el.text
 
+    expected = {f"Pagination test comment {i:03d}" for i in range(1, COUNT + 1)}
+    existing = _fetch_all_comments(s, url, file_id)
+    expected_ts = [ts for _, msg, ts in existing if msg in expected]
+    if expected.issubset({m for _, m, _ in existing}) and len(set(expected_ts)) >= COUNT:
+        print(f"  {COUNT} comments on file {file_id} (already present with distinct timestamps)")
+        return
+
+    for cid, _, _ in existing:
+        s.delete(f"{url}/remote.php/dav/comments/files/{file_id}/{cid}")
+
     for i in range(1, COUNT + 1):
         s.post(
             f"{url}/remote.php/dav/comments/files/{file_id}",
             json={"actorType": "users", "verb": "comment", "message": f"Pagination test comment {i:03d}"},
             headers={"Content-Type": "application/json"},
         )
-    print(f"  {COUNT} comments on file {file_id}")
+        if i < COUNT:
+            time.sleep(1.05)
+    print(f"  {COUNT} comments on file {file_id} (reset and re-created with 1.05s spacing for stable pagination)")
 
 
 def main() -> None:

tests/integration/test_pagination.py

@@ -51,25 +51,30 @@ async def _ensure_pagination_conversations(nc_mcp: McpTestHelper) -> None:
 
 
 async def _ensure_pagination_events(nc_mcp: McpTestHelper) -> None:
-    """Create calendar events if they don't already exist."""
+    """Create calendar events if they don't already exist.
+
+    Idempotency is keyed on the event summary text, not UID -- create_event
+    generates random server-side UUIDs, so a UID-based check would think no
+    events exist and add a fresh batch on every run, accumulating duplicates.
+    """
     result = json.loads(
         await nc_mcp.call(
             "get_events",
             calendar_id="personal",
             start="2027-06-01T00:00:00Z",
             end="2027-06-30T23:59:59Z",
-            limit=200,
+            limit=500,
         )
     )
-    existing_uids = {e["uid"] for e in result["data"]}
+    existing_summaries = {e.get("summary", "") for e in result["data"]}
     for i in range(1, ITEM_COUNT + 1):
-        uid = f"{PREFIX}-event-{i:03d}"
-        if uid not in existing_uids:
+        summary = f"Pagination Test Event {i:03d}"
+        if summary not in existing_summaries:
             hour = i % 24
             await nc_mcp.call(
                 "create_event",
                 calendar_id="personal",
-                summary=f"Pagination Test Event {i:03d}",
+                summary=summary,
                 start=f"2027-06-01T{hour:02d}:00:00Z",
                 end=f"2027-06-01T{hour:02d}:30:00Z",
             )

tests/integration/test_search.py

@@ -100,18 +100,38 @@ async def test_search_with_limit(self, nc_mcp: McpTestHelper) -> None:
 
     @pytest.mark.asyncio
     async def test_search_pagination(self, nc_mcp: McpTestHelper) -> None:
-        result1 = await nc_mcp.call("unified_search", provider="files", term="pagtest", limit=2)
-        data1 = json.loads(result1)
-        if not data1["has_more"]:
-            pytest.skip("Not enough pagtest files for pagination test")
-        cursor = str(data1["cursor"])
-
-        result2 = await nc_mcp.call("unified_search", provider="files", term="pagtest", limit=2, cursor=cursor)
-        data2 = json.loads(result2)
-        assert data2["entries"]
-        titles1 = {e["title"] for e in data1["entries"]}
-        titles2 = {e["title"] for e in data2["entries"]}
-        assert not titles1.intersection(titles2), "Pages should not overlap"
+        # NC's FilesSearchProvider issues its DB query without an ORDER BY, so on Postgres
+        # the offset-based pages can overlap or skip rows (SQLite happens to be stable by
+        # insertion order). Full enumeration of the seeded files is therefore not guaranteed,
+        # so we verify the pagination *contract* instead: limit is respected, the cursor
+        # advances, follow-up pages keep returning results, and paging surfaces more distinct
+        # results than a single page holds.
+        limit = 5
+        first = json.loads(await nc_mcp.call("unified_search", provider="files", term="pagtest", limit=limit))
+        assert len(first["entries"]) <= limit
+        if not first["has_more"]:
+            pytest.skip("Not enough pagtest files seeded for a multi-page search")
+
+        seen = {e["title"] for e in first["entries"]}
+        cursor = str(first["cursor"])
+        pages = 1
+        for _ in range(40):
+            data = json.loads(
+                await nc_mcp.call("unified_search", provider="files", term="pagtest", limit=limit, cursor=cursor)
+            )
+            entries = data["entries"]
+            assert len(entries) <= limit
+            if entries:
+                seen.update(e["title"] for e in entries)
+                pages += 1
+            if not data["has_more"]:
+                break
+            next_cursor = str(data["cursor"])
+            assert next_cursor != cursor, "Cursor must advance across pages"
+            cursor = next_cursor
+
+        assert pages >= 2, "Expected pagination to yield more than one page"
+        assert len(seen) > limit, f"Pagination should surface more than one page of distinct results; got {len(seen)}"
 
     @pytest.mark.asyncio
     async def test_search_nonexistent_provider_raises(self, nc_mcp: McpTestHelper) -> None:

tests/integration/test_user_status.py

@@ -44,6 +44,10 @@ async def test_get_other_user_status(self, nc_mcp: McpTestHelper) -> None:
     async def test_get_own_status_when_never_set(self, nc_mcp: McpTestHelper) -> None:
         """A fresh user who never set a status gets a 404 from the API.
         The tool should handle this gracefully and return a default offline status."""
+        # Cleanup uses nc_mcp.client (admin, captured at fixture setup) directly rather than
+        # the delete_user tool, because create_server(fresh_config) below swaps the global
+        # client singleton to the fresh non-admin user -- so the tool's get_client() would
+        # return the fresh client (closed and unauthorized) and silently leak the test user.
         try:
             await nc_mcp.call("create_user", user_id="mcp-fresh-status", password="t3St*Pw!xQ9#mK2z")
             fresh_config = Config(
@@ -64,7 +68,7 @@ async def test_get_own_status_when_never_set(self, nc_mcp: McpTestHelper) -> Non
                 await fresh_helper.client.close()
         finally:
             with contextlib.suppress(Exception):
-                await nc_mcp.call("delete_user", user_id="mcp-fresh-status")
+                await nc_mcp.client.ocs_delete("cloud/users/mcp-fresh-status")
 
     @pytest.mark.asyncio
     async def test_get_nonexistent_user_status(self, nc_mcp: McpTestHelper) -> None: