kamidso-back/tests/kamidso/test_spreadsheet_skill.py

import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
SKILLS = ROOT / "skills"


def _run(script: Path, args: dict) -> dict:
    proc = subprocess.run(
        [sys.executable, str(script), json.dumps(args)],
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
        cwd=str(script.parent),
        env={**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1"},
    )
    assert proc.returncode == 0, proc.stderr or proc.stdout
    payload = json.loads(proc.stdout)
    text = next(
        chunk["content"]
        for chunk in payload["chunks"]
        if chunk.get("output_type") == "text"
    )
    return json.loads(text)


def test_sales_intelligence_layout_and_run(tmp_path):
    skill_dir = SKILLS / "sales-intelligence"
    assert (skill_dir / "SKILL.md").is_file()
    sample = tmp_path / "sales.csv"
    sample.write_text(
        "date,customer,product,sales_amount\n2024-01-01,A,P1,100\n2024-02-01,B,P1,150\n",
        encoding="utf-8",
    )
    body = _run(skill_dir / "scripts" / "sales_intelligence.py", {"input_file": str(sample)})
    assert body["ok"] is True
    assert body["skill_id"] == "sales-intelligence"
    assert body["data"]["total_sales"] == 250


def test_skill_manager_resolves_sales_intelligence():
    from kamidso.engine.core.agent.skill.manage import SkillManager

    path = Path(SkillManager(None)._get_skill_path("sales-intelligence"))
    if not path.is_dir():
        path = SKILLS / "sales-intelligence"
    assert path.name == "sales-intelligence"
    assert (path / "SKILL.md").is_file()
    assert (path / "manifest.json").is_file()


@pytest.mark.parametrize(
    "skill_id,filename,rows,extra",
    [
        (
            "profitability-cost-intelligence",
            "profitability_cost_intelligence.py",
            "date,revenue,cogs\n2024-01-01,100,40\n2024-02-01,80,50\n",
            {},
        ),
        (
            "receivables-cash-control",
            "receivables_cash_control.py",
            "customer,due_date,invoice_amount,outstanding\nA,2024-01-01,100,80\nB,2024-06-01,50,50\n",
            {},
        ),
        (
            "inventory-procurement-intelligence",
            "inventory_procurement_intelligence.py",
            "item,quantity,unit_cost\nX,10,5\nY,2,20\n",
            {},
        ),
        (
            "financial-anomaly-investigator",
            "financial_anomaly_investigator.py",
            "date,amount,vendor\n2024-01-01,10,A\n2024-01-02,12,A\n2024-01-03,11,A\n2024-01-04,900,A\n",
            {"enable_isolation_forest": False},
        ),
        (
            "iran-payroll-compliance",
            "iran_payroll_compliance.py",
            "employee,gross_pay\nAli,100\nReza,120\n",
            {},
        ),
        (
            "iran-tax-statutory-compliance",
            "iran_tax_statutory_compliance.py",
            "account,debit,credit\n1001,50,0\n2001,0,50\n",
            {},
        ),
        (
            "forecast-scenario-decision-intelligence",
            "forecast_scenario_decision.py",
            "date,value\n"
            + "\n".join(f"2024-{month:02d}-01,{10 + month}" for month in range(1, 13))
            + "\n",
            {"mode": "forecast", "horizon": 2, "frequency": "MS", "min_train": 6},
        ),
    ],
)
def test_production_skill_scripts(tmp_path, skill_id, filename, rows, extra):
    sample = tmp_path / "data.csv"
    sample.write_text(rows, encoding="utf-8")
    args = {"input_file": str(sample), **extra}
    body = _run(SKILLS / skill_id / "scripts" / filename, args)
    assert body["ok"] is True
    assert body.get("skill_id") == skill_id


def test_financial_reconciliation_two_files(tmp_path):
    left = tmp_path / "bank.csv"
    right = tmp_path / "ledger.csv"
    left.write_text("date,amount,reference\n2024-01-01,100,R1\n", encoding="utf-8")
    right.write_text("date,amount,reference\n2024-01-01,100,R1\n", encoding="utf-8")
    body = _run(
        SKILLS / "financial-reconciliation" / "scripts" / "financial_reconciliation.py",
        {"left_file": str(left), "right_file": str(right)},
    )
    assert body["ok"] is True


def test_root_cause_from_evidence():
    body = _run(
        SKILLS / "root-cause-investigator" / "scripts" / "root_cause_investigator.py",
        {
            "target": {"metric": "profit", "event": "decline"},
            "evidence": [
                {
                    "claim": "Purchase cost increased",
                    "contribution": -2.1,
                    "confidence": 0.9,
                    "evidence_type": "DRIVER",
                }
            ],
        },
    )
    assert body["ok"] is True
    assert body["skill_id"] == "root-cause-investigator"