afv-library/skills/investigating-agentforce-d360/scripts/tests/test_resolve_session_main.py

150 lines
4.8 KiB
Python
Raw Normal View History

@W-22707610 feat: add investigating-agentforce-d360 skill Data Cloud 360° view of a single Agentforce session — DC-only, zero Splunk dependency. Pulls 24 STDM + GenAI DMOs via the Data Cloud Query REST API, assembles a hierarchical session tree (Interaction → Step → Generation → GatewayRequest), and renders a human-readable markdown summary with transcript + per-turn topic/action invocations + LLM generations + tool calls + audit chain. Migrated as a standalone Apache-2.0 skill from an internal hub plugin — self-contained, no sibling-skill or plugin dependencies. What this skill answers: - "Trace session <uuid>" / "Summarize what happened in <0Mw…>" - "Find escalated sessions today on Messaging in <org>" - Session discovery by time / agent / channel / outcome / conversation text when the user has no session id What it does NOT answer (use a different surface): - Design-time architecture — use investigating-agentforce-architecture - Runtime planner availability — DC alone can't tell you which topic/action was eligible for the classifier on a given turn Skill layout: - 8 Python pipeline modules (fetch_dc, assemble_dc, render_dc, discover_sessions, resolve_session, dc, storage, config) - 4 _shared helpers (paths, fs_guard, sql, __init__) with skill-scoped DATA_ROOT (~/.claude/data/investigating-agentforce-d360/) - 26 SQL templates under assets/dc/ - 27 test files (367 tests + 18 subtests, 100% passing) - 3 reference docs (artifacts.md, dc_dmo_fields.md, dc_pipeline_contract.md) - SKILL.md (sf-skills frontmatter, license: Apache-2.0, metadata.version: "1.0") - README.md (external-facing quick-start) - tools/grant_allowlist.py (idempotent first-run permission grant) - tools/archive_data_dir.sh (opt-in stop-hook tarballer) Quality gates: - pytest scripts/tests/: 367 passed + 18 subtests, 0 failures - npm run validate:skills: 62 of 62 skill(s) checked, 0 errors - Live end-to-end runs against 3 real Salesforce sessions exercising both the full-tree and STDM-lag gateway-direct render branches - 4 independent code-review rounds (correctness, security, markdown, architecture-critic) — all findings addressed Customer-data hygiene: no live tenant ids, no internal sprint markers, no hub/sibling-skill references. Synthetic fixtures look obviously synthetic (`019dface-…` UUIDs, `0MwTESTMSG…` MessagingSession ids, `00DTESTORG…` org ids, `MyAgent` placeholder agent name). Sibling skill: investigating-agentforce-architecture (PR #278) — same migration pattern, design-time metadata; complementary scope.
2026-05-28 18:57:52 +08:00
"""Tests for ``resolve_session.main`` CLI.
Argparse-driven exercises every branch:
- UUID pass-through
- --disk-only hit / miss
- live path with --org
- disk-first fallback when no --org
- not-found-no-org error
"""
from __future__ import annotations
import io
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import mock
from . import _bootstrap # noqa: F401 — sys.path setup
import resolve_session # type: ignore
from .fixtures.synthetic_session import IDS, write_to_disk # type: ignore
class _MainHarness:
"""Patch sys.argv + stdout/stderr; optionally plant the synthetic
fixture under DATA_ROOT for disk-resolution tests."""
def __init__(self, *, argv: list[str], plant_fixture: bool = False):
self.argv = argv
self.plant_fixture = plant_fixture
def __enter__(self):
self._tmp = TemporaryDirectory()
self._tmpdir = Path(self._tmp.name)
if self.plant_fixture:
write_to_disk(self._tmpdir)
self._patches = [
mock.patch.object(resolve_session.sys, "argv",
["resolve_session.py", *self.argv]),
mock.patch.object(resolve_session, "DATA_ROOT", self._tmpdir),
]
for p in self._patches:
p.start()
self.stdout = io.StringIO()
self.stderr = io.StringIO()
self._stdout_patch = mock.patch("sys.stdout", self.stdout)
self._stderr_patch = mock.patch("sys.stderr", self.stderr)
self._stdout_patch.start()
self._stderr_patch.start()
return self
def __exit__(self, *exc):
self._stderr_patch.stop()
self._stdout_patch.stop()
for p in self._patches:
p.stop()
self._tmp.cleanup()
# -----------------------------------------------------------------------------
# UUID pass-through
# -----------------------------------------------------------------------------
class UuidPassThroughTests(unittest.TestCase):
def test_uuid_input_prints_unchanged_returns_zero(self):
with _MainHarness(argv=["--id", IDS.SID]) as h:
rc = resolve_session.main()
self.assertEqual(rc, 0)
self.assertIn(IDS.SID, h.stdout.getvalue())
# -----------------------------------------------------------------------------
# --disk-only branch
# -----------------------------------------------------------------------------
class DiskOnlyBranchTests(unittest.TestCase):
def test_disk_only_hit_prints_uuid(self):
with _MainHarness(
argv=["--id", "0MwTESTMSG12345AAA", "--disk-only"],
plant_fixture=True,
) as h:
rc = resolve_session.main()
self.assertEqual(rc, 0)
self.assertIn(IDS.SID, h.stdout.getvalue())
def test_disk_only_miss_returns_one(self):
with _MainHarness(
argv=["--id", "0Mw000000000000", "--disk-only"],
plant_fixture=True,
) as h:
rc = resolve_session.main()
self.assertEqual(rc, 1)
self.assertIn("no local session dir matches", h.stderr.getvalue())
# -----------------------------------------------------------------------------
# Live path with --org
# -----------------------------------------------------------------------------
class LiveOrgPathTests(unittest.TestCase):
def test_live_path_returns_uuid_from_resolver(self):
with _MainHarness(
argv=["--id", "0MwTESTMSG12345AAA", "--org", "my-org"],
) as h:
with mock.patch.object(
resolve_session, "resolve", return_value="resolved-uuid",
) as r:
rc = resolve_session.main()
r.assert_called_once_with("0MwTESTMSG12345AAA", org="my-org")
self.assertEqual(rc, 0)
self.assertIn("resolved-uuid", h.stdout.getvalue())
# -----------------------------------------------------------------------------
# Disk-first convenience fallback
# -----------------------------------------------------------------------------
class DiskFirstFallbackTests(unittest.TestCase):
def test_disk_hit_when_no_org_supplied(self):
# No --org but disk resolution succeeds via fixture.
with _MainHarness(
argv=["--id", "0MwTESTMSG12345AAA"],
plant_fixture=True,
) as h:
rc = resolve_session.main()
self.assertEqual(rc, 0)
self.assertIn(IDS.SID, h.stdout.getvalue())
def test_disk_miss_no_org_returns_two_with_hint(self):
# No --org AND disk miss → exit 2 + helpful stderr.
with _MainHarness(
argv=["--id", "0Mw000000000000"],
) as h:
rc = resolve_session.main()
self.assertEqual(rc, 2)
self.assertIn("not found on disk", h.stderr.getvalue())
self.assertIn("--org <alias> required", h.stderr.getvalue())
if __name__ == "__main__":
unittest.main()