File size: 4,325 Bytes
bf3cbf1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """Smoke tests for the SPINOR-RL CLI."""
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
import pytest
from spinor_os.engine import ExperimentationOS
from spinor_os.persistence import PersistenceManager
def run_cli(*args, fast_llm=True, env=None):
"""Run the CLI with the same interpreter and return (returncode, stdout, stderr)."""
merged = dict(__import__("os").environ)
merged["SPINOR_LOG_LEVEL"] = "WARNING"
if fast_llm:
# Point the default LLM provider at a non-existent endpoint so tests
# fall back to deterministic local allocation quickly.
merged["SPINOR_LLM_PROVIDER"] = "ollama"
merged["OLLAMA_BASE_URL"] = "http://127.0.0.1:1"
if env:
merged.update(env)
result = subprocess.run(
[sys.executable, "-m", "spinor_os"] + list(args),
capture_output=True,
text=True,
env=merged,
)
return result.returncode, result.stdout, result.stderr
def test_cli_assign_local():
rc, out, err = run_cli(
"assign",
"--local",
"--employee", "emp-cli-001",
"--role", "field_representative",
"--territory", "northeast",
)
assert rc == 0, f"CLI failed: {err}"
data = json.loads(out)
assert data["count"] == 1
assert data["source"] == "local_constrained_allocation"
assert data["fallback_status"] == "primary"
def test_cli_assign_remote_fallback():
rc, out, err = run_cli(
"assign",
"--employee", "emp-cli-002",
"--role", "field_representative",
"--territory", "northeast",
env={"ADVANTAGE_FOUNDRY_ASSIGN_URL": "http://127.0.0.1:1"},
)
assert rc == 0, f"CLI failed: {err}"
data = json.loads(out)
assert data["count"] == 1
assert data["source"] == "local_constrained_allocation"
assert data["fallback_status"] == "local_fallback"
assert data.get("provenance")
assert data.get("error")
def _create_experiment(db_path: str) -> str:
"""Create a real employee, hypothesis, and experiment for the import test."""
os = ExperimentationOS(persistence=PersistenceManager(db_path))
os.register_employee(
employee_id="emp-cli-001",
role="field-scientist",
territory="northeast",
)
hypothesis = os.propose_hypothesis(
statement="Imported outreach increases appointments.",
causal_claim="A targeted email causes a higher appointment rate.",
predicted_effect={
"metric": "appointment_rate",
"direction": "increase",
"magnitude": 0.1,
"unit": "percentage_point",
"timing": "7d",
},
employee_owner="emp-cli-001",
falsification_criteria=["No increase in appointment rate."],
customer_segment="enterprise",
territory="northeast",
modification="imported_email",
)
experiment = os.start_experiment(hypothesis.hypothesis_id)
return experiment.experiment_id
def test_cli_import_csv_with_experiment():
with tempfile.TemporaryDirectory() as tmp:
db_path = os.path.join(tmp, "test.db")
experiment_id = _create_experiment(db_path)
csv_path = Path(tmp) / "events.csv"
csv_path.write_text(
"experiment_id,actor_id,event_type,outcome_value,metric,execution_quality\n"
",emp-cli-001,OUTCOME_OBSERVED,0.18,appointment_rate,0.90\n"
)
rc, out, err = run_cli(
"import-csv",
str(csv_path),
"--experiment",
experiment_id,
env={"SPINOR_DB": db_path},
)
assert rc == 0, f"CLI failed: {err}"
data = json.loads(out)
assert data["rows_read"] == 1
assert data["rows_ingested"] == 1
assert data["errors"] == []
assert data["event_ids"]
def test_cli_import_csv_missing_file():
rc, out, err = run_cli("import-csv", "/tmp/does_not_exist_12345.csv")
assert rc != 0
def test_cli_health():
rc, out, err = run_cli("health")
assert rc == 0
data = json.loads(out)
assert data["status"] == "healthy"
def test_cli_ladder():
rc, out, err = run_cli("ladder")
assert rc == 0
data = json.loads(out)
assert data["central_design_law"]
assert data["maturity_ladder"]
|