Dedupe if-intf/config_interface collects and allow idle manual collect.

Remap legacy if_intf items, skip duplicate CLI+metric work, hide aux command duplicates in the workbook, use profile titles for sheet tabs, sticky table headers, and let scheduled tasks trigger collect-now when idle.

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_batch_workbook_api.py

@@ -68,6 +68,7 @@ class BatchWorkbookApiTests(unittest.TestCase):
         self.assertEqual(out["sheets"][0]["metric_id"], "arp")
         self.assertEqual(out["sheets"][0]["row_count"], 2)
         self.assertEqual(out["sheets"][0]["commands"][0]["raw_command"], "show arp | one-line")
+        self.assertTrue(out["sheets"][0].get("title"))
 
     def test_list_metric_rows_rejects_commands_sheet(self) -> None:
         db = MagicMock()

tests/test_collect_dedupe.py

@@ -0,0 +1,76 @@
+"""Collect work-list dedupe / if_intf remap / manual collect."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from netx_api.biz_state.collect_runner import (
+    _resolve_collect_profile,
+    dispatch_collect,
+    trigger_collect_now,
+)
+from netx_api.biz_state.profiles import get_profile, reload_profiles
+
+
+class CollectProfileResolveTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        reload_profiles()
+
+    def test_if_intf_remaps_to_config_interface(self) -> None:
+        p = _resolve_collect_profile("zte.if_intf")
+        self.assertIsNotNone(p)
+        assert p is not None
+        self.assertEqual(p.profile_id, "zte.config_interface")
+        self.assertEqual(p.metric_id, "config_interface")
+
+    def test_enabled_profile_passthrough(self) -> None:
+        p = _resolve_collect_profile("zte.arp")
+        self.assertIsNotNone(p)
+        assert p is not None
+        self.assertEqual(p.profile_id, "zte.arp")
+
+    def test_if_intf_profile_disabled(self) -> None:
+        raw = get_profile("zte.if_intf")
+        self.assertIsNotNone(raw)
+        assert raw is not None
+        self.assertFalse(raw.enabled)
+
+
+class ManualCollectTests(unittest.TestCase):
+    def test_trigger_collect_now_passes_manual(self) -> None:
+        task = MagicMock()
+        task.collect_running = False
+        db = MagicMock()
+        db.get.return_value = task
+        with (
+            patch("netx_api.biz_state.collect_runner.SessionLocal", return_value=db),
+            patch("netx_api.biz_state.collect_runner.dispatch_collect") as dc,
+        ):
+            out = trigger_collect_now("t1")
+        self.assertTrue(out["ok"])
+        dc.assert_called_once_with("t1", manual=True)
+
+    def test_dispatch_manual_allows_paused(self) -> None:
+        task = MagicMock()
+        task.collect_running = False
+        task.status = "paused"
+        task.id = "t1"
+        task.source = "managed"
+        task.ne_id = "n1"
+        task.ne_name = "NE"
+        task.vendor = "zte"
+        db = MagicMock()
+        db.get.return_value = task
+        # items query → empty so it returns early after setting error
+        q = MagicMock()
+        q.filter.return_value.order_by.return_value.all.return_value = []
+        db.query.return_value = q
+        with patch("netx_api.biz_state.collect_runner.SessionLocal", return_value=db):
+            dispatch_collect("t1", manual=True)
+        self.assertEqual(task.last_error, "no enabled task items")
+
+
+if __name__ == "__main__":
+    unittest.main()