| """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: |
| |
| |
| 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"] |
|
|
