Ty
d6b74c42f6
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.
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""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}"
|
|
)
|