Implement S7 operator connect helpers for SaaS and channels.

Add dry-run-default connect dispatcher and per-integration scripts with gitignored local capability state, docs, and unit tests. Mutations require --apply and use nemohermes/openshell only.
2026-07-27 13:32:42 -07:00
parent 76df8b7ab6
commit d6b74c42f6
17 changed files with 2229 additions and 35 deletions
@@ -0,0 +1,276 @@
+"""Tests for S7 connect state management and capability report schema.
+
+Validates:
+  - Capability state JSON schema matches setup-education fixture format
+  - Status values are valid (connected | skipped | later | error | offline)
+  - State file operations (create, read, update)
+  - Exported capability report matches fixture schema
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import pathlib
+import tempfile
+
+import pytest
+
+# ── Paths ──────────────────────────────────────────────────────────────────
+_REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
+_FIXTURES_DIR = _REPO_ROOT / "data" / "fixtures" / "setup"
+
+# ── Valid status values ────────────────────────────────────────────────────
+VALID_STATUSES = {"connected", "skipped", "later", "error", "offline"}
+
+# ── Required capability entry fields ───────────────────────────────────────
+REQUIRED_ENTRY_FIELDS = {"area", "provider", "status", "details"}
+
+# ── Required top-level report fields ───────────────────────────────────────
+REQUIRED_REPORT_FIELDS = {"salon_name", "is_fixture", "capabilities"}
+
+
+# ── Fixture loading tests ──────────────────────────────────────────────────
+
+@pytest.mark.parametrize("fixture_name", [
+    "capability_matrix.json",
+    "capability_matrix_all_connected.json",
+    "capability_matrix_with_errors.json",
+])
+def test_fixture_schema(fixture_name: str):
+    """Each fixture file has valid schema."""
+    fixture_path = _FIXTURES_DIR / fixture_name
+    assert fixture_path.exists(), f"Fixture not found: {fixture_path}"
+
+    data = json.loads(fixture_path.read_text(encoding="utf-8"))
+
+    # Top-level fields
+    for field in REQUIRED_REPORT_FIELDS:
+        assert field in data, f"Missing required field: {field}"
+
+    # Capabilities array
+    assert isinstance(data["capabilities"], list)
+    assert len(data["capabilities"]) > 0
+
+    # Each capability entry
+    for i, entry in enumerate(data["capabilities"]):
+        for field in REQUIRED_ENTRY_FIELDS:
+            assert field in entry, f"Entry {i} missing field: {field}"
+        assert entry["status"] in VALID_STATUSES, (
+            f"Entry {i} invalid status: {entry['status']!r}"
+        )
+
+
+def test_fixture_is_fixture_flag():
+    """All fixtures have is_fixture: true."""
+    for fixture_path in _FIXTURES_DIR.glob("capability_matrix*.json"):
+        data = json.loads(fixture_path.read_text(encoding="utf-8"))
+        assert data.get("is_fixture") is True, (
+            f"{fixture_path.name} should have is_fixture: true"
+        )
+
+
+# ── Capability state schema tests ──────────────────────────────────────────
+
+def test_capability_state_schema():
+    """Validate capability state JSON schema."""
+    state = {
+        "generated_at": "2026-07-27T00:00:00+00:00",
+        "is_fixture": False,
+        "square": {
+            "status": "connected",
+            "details": "Square remote MCP registered",
+            "updated_at": "2026-07-27T00:00:00+00:00",
+        },
+        "whatsapp": {
+            "status": "connected",
+            "details": "WhatsApp channel active",
+            "updated_at": "2026-07-27T00:00:00+00:00",
+        },
+        "quickbooks": {
+            "status": "skipped",
+            "details": "Operator skipped",
+            "updated_at": "2026-07-27T00:00:00+00:00",
+        },
+        "vagaro": {
+            "status": "error",
+            "details": "API verification failed",
+            "updated_at": "2026-07-27T00:00:00+00:00",
+        },
+    }
+
+    # Validate each target
+    meta_keys = {"generated_at", "is_fixture"}
+    for key, value in state.items():
+        if key in meta_keys:
+            continue
+        assert "status" in value, f"Target {key} missing status"
+        assert value["status"] in VALID_STATUSES, (
+            f"Target {key} invalid status: {value['status']!r}"
+        )
+        assert "details" in value, f"Target {key} missing details"
+        assert "updated_at" in value, f"Target {key} missing updated_at"
+
+
+def test_capability_state_json_serializable():
+    """State must be JSON-serializable."""
+    state = {
+        "square": {"status": "connected", "details": "OK", "updated_at": "2026-01-01T00:00:00Z"},
+        "whatsapp": {"status": "skipped", "details": "Skipped", "updated_at": "2026-01-01T00:00:00Z"},
+    }
+    # Should not raise
+    json.dumps(state)
+
+
+# ── Status value tests ─────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("status", VALID_STATUSES)
+def test_valid_status_values(status: str):
+    """All valid status values are recognized."""
+    assert status in VALID_STATUSES
+
+
+@pytest.mark.parametrize("invalid_status", ["pending", "unknown", "active", ""])
+def test_invalid_status_values(invalid_status: str):
+    """Invalid status values are not in the valid set."""
+    assert invalid_status not in VALID_STATUSES
+
+
+# ── Area mapping tests ─────────────────────────────────────────────────────
+
+AREA_MAP = {
+    "name": "identity",
+    "profile": "profile",
+    "whatsapp": "channels",
+    "email": "channels",
+    "telegram": "channels",
+    "square": "scheduling",
+    "quickbooks": "books",
+    "vagaro": "scheduling",
+}
+
+
+@pytest.mark.parametrize("target,expected_area", AREA_MAP.items())
+def test_area_mapping(target: str, expected_area: str):
+    """Each target maps to the correct area."""
+    assert AREA_MAP[target] == expected_area
+
+
+# ── .local directory tests ─────────────────────────────────────────────────
+
+def test_local_dir_gitignored():
+    """.local/ must be in .gitignore."""
+    gitignore = _REPO_ROOT / ".gitignore"
+    content = gitignore.read_text(encoding="utf-8")
+    assert ".local/" in content, ".local/ should be in .gitignore"
+
+
+def test_local_dir_not_tracked():
+    """.local/ should not be in git."""
+    import subprocess
+    result = subprocess.run(
+        ["git", "ls-files", ".local/"],
+        capture_output=True, text=True, cwd=str(_REPO_ROOT),
+    )
+    assert result.stdout.strip() == "", ".local/ should not be tracked by git"
+
+
+# ── Owner-safe keyword tests ───────────────────────────────────────────────
+
+FORBIDDEN_OWNER_KEYWORDS = [
+    "terminal", "docker", "nano", "shell", "bash", "sudo",
+    "apt-get", "yum", "dnf", "pip install", "npm install",
+]
+
+
+def test_connect_scripts_no_owner_keywords_in_help():
+    """Connect script help output must not contain forbidden keywords as instructions.
+
+    Safety guarantees like 'Owner never receives terminal instructions' are allowed
+    because they describe what the owner does NOT receive, not instructions to follow.
+    """
+    import subprocess
+
+    script = _REPO_ROOT / "scripts" / "connect.sh"
+    result = subprocess.run(
+        ["bash", str(script), "--help"],
+        capture_output=True, text=True, cwd=str(_REPO_ROOT),
+    )
+    output = result.stdout.lower()
+
+    for keyword in FORBIDDEN_OWNER_KEYWORDS:
+        # Allow "shell" in the context of "OpenShell" (platform CLI name)
+        if keyword == "shell":
+            import re
+            matches = re.findall(r'(?<!open)shell', output)
+            assert len(matches) == 0, (
+                f"Help output contains forbidden keyword: {keyword}"
+            )
+            continue
+
+        # Allow keywords that appear in safety guarantee contexts
+        # (e.g., "Owner never receives terminal/Docker/nano instructions")
+        # These are guarantees about what the owner does NOT receive.
+        safety_patterns = [
+            "never receives",
+            "never gets",
+            "owner never",
+            "no owner",
+            "owner-safe",
+        ]
+        is_safety_context = any(
+            pattern in output and keyword in output
+            for pattern in safety_patterns
+        )
+        if is_safety_context:
+            # Verify the keyword appears in a safety guarantee, not an instruction
+            import re
+            # Check that the keyword is NOT preceded by instruction-like verbs
+            instruction_patterns = [
+                r'run\s+.*' + keyword,
+                r'execute\s+.*' + keyword,
+                r'open\s+.*' + keyword,
+                r'type\s+.*' + keyword,
+                r'install\s+.*' + keyword,
+            ]
+            has_instruction = any(
+                re.search(pattern, output) for pattern in instruction_patterns
+            )
+            assert not has_instruction, (
+                f"Help output contains instruction with forbidden keyword: {keyword}"
+            )
+            continue
+
+        assert keyword not in output, (
+            f"Help output contains forbidden keyword: {keyword}"
+        )
+
+
+def test_capability_report_no_owner_keywords():
+    """Capability report text must not contain forbidden keywords."""
+    # Import the capability report formatter
+    sys_path = str(_REPO_ROOT / "skills" / "_lib")
+    import sys
+    if sys_path not in sys.path:
+        sys.path.insert(0, sys_path)
+
+    from lumina_skills.setup.capability_report import (
+        CapabilityEntry,
+        ConnectionStatus,
+        build_capability_report,
+        format_capability_report_text,
+    )
+
+    entries = [
+        CapabilityEntry("channels", "whatsapp", ConnectionStatus.CONNECTED),
+        CapabilityEntry("channels", "email", ConnectionStatus.SKIPPED),
+        CapabilityEntry("scheduling", "vagaro", ConnectionStatus.OFFLINE),
+        CapabilityEntry("books", "quickbooks_online", ConnectionStatus.ERROR),
+    ]
+    report = build_capability_report(entries, "Test Salon", is_fixture=True)
+    text = format_capability_report_text(report).lower()
+
+    for keyword in FORBIDDEN_OWNER_KEYWORDS:
+        assert keyword not in text, (
+            f"Capability report contains forbidden keyword: {keyword}"
+        )