Bot & Automation
jorgasisten
/root/hermes-projects/jorgasisten
tests/test_operations.py
text
from types import SimpleNamespace
from typing import Any
from jorgasisten.schemas import AssistantDecision, SheetRow, SheetSchema
from jorgasisten.services.operations import OperationExecutor
class FakeSheets:
def __init__(self) -> None:
self.schemas = {"Stok": SheetSchema(title="Stok", headers=["Produk", "Qty", "Status"])}
self.rows = {
"Stok": [
SheetRow(sheet_name="Stok", row_number=2, values={"Produk": "Oli MPX", "Qty": "5", "Status": "active"})
]
}
def get_catalog(self) -> list[SheetSchema]:
return list(self.schemas.values())
def get_schema(self, sheet_name: str) -> SheetSchema:
return self.schemas[sheet_name]
def find_rows(self, sheet_name: str, filters: dict[str, Any], limit: int | None = None) -> list[SheetRow]:
rows = self.rows.get(sheet_name, [])
if not filters:
return rows[: limit or 10]
return [row for row in rows if filters.get("Produk", "").lower() in row.values["Produk"].lower()]
def read_rows(self, sheet_name: str) -> list[SheetRow]:
return self.rows.get(sheet_name, [])
def append_row(self, sheet_name: str, data: dict[str, Any]) -> SheetRow:
row = SheetRow(sheet_name=sheet_name, row_number=3, values=data)
self.rows.setdefault(sheet_name, []).append(row)
return row
def update_row(self, sheet_name: str, row_number: int, data: dict[str, Any]) -> SheetRow:
self.rows[sheet_name][0].values.update(data)
return self.rows[sheet_name][0]
def soft_delete_row(self, sheet_name: str, row_number: int) -> bool:
self.rows[sheet_name][0].values["Status"] = "deleted"
return True
def delete_row(self, sheet_name: str, row_number: int) -> None:
self.rows[sheet_name] = []
def create_sheet(self, sheet_name: str, headers: list[str]) -> SheetSchema:
self.schemas[sheet_name] = SheetSchema(title=sheet_name, headers=headers)
self.rows[sheet_name] = []
return self.schemas[sheet_name]
def update_headers(self, sheet_name: str, headers: list[str]) -> SheetSchema:
self.schemas[sheet_name] = SheetSchema(title=sheet_name, headers=headers)
return self.schemas[sheet_name]
def rename_sheet(self, sheet_name: str, new_sheet_name: str) -> SheetSchema:
schema = self.schemas.pop(sheet_name)
self.schemas[new_sheet_name] = SheetSchema(title=new_sheet_name, headers=schema.headers)
self.rows[new_sheet_name] = self.rows.pop(sheet_name, [])
return self.schemas[new_sheet_name]
def delete_sheet_full(self, sheet_name: str) -> None:
self.schemas.pop(sheet_name)
self.rows.pop(sheet_name, None)
def test_executor_blocks_write_for_viewer() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
executor = OperationExecutor(FakeSheets(), settings=settings)
decision = AssistantDecision(
intent="input_data",
sheet_name="Stok",
operation="create",
confidence=95,
data={"Produk": "Oli MPX", "Qty": "10"},
)
result = executor.execute(decision, actor_role="viewer")
assert result.ok is False
assert "belum boleh input atau update" in result.user_message.lower()
def test_executor_previews_create_before_saving() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="input_data",
sheet_name="Stok",
operation="create",
confidence=95,
data={"Produk": "Oli Baru", "Qty": "10", "Status": "active"},
)
preview = executor.execute(decision, actor_role="operator")
assert preview.ok is False
assert preview.requires_confirmation is True
assert "DATA TERBACA" in preview.user_message
assert len(fake.rows["Stok"]) == 1
saved = executor.execute(decision, actor_role="operator", confirmed=True)
assert saved.ok is True
assert len(fake.rows["Stok"]) == 2
assert fake.rows["Stok"][-1].values["Produk"] == "Oli Baru"
def test_executor_autofills_no_and_blocks_duplicate_identity() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
fake.schemas["Log"] = SheetSchema(title="Log", headers=["No", "Produk", "Status"])
fake.rows["Log"] = [
SheetRow(
sheet_name="Log",
row_number=2,
values={"No": "1", "Produk": "Oli MPX", "Status": "Selesai"},
)
]
executor = OperationExecutor(fake, settings=settings)
preview = executor.execute(
AssistantDecision(
intent="input_data",
sheet_name="Log",
operation="create",
confidence=95,
data={"Produk": "Ban Luar", "Status": "Belum"},
),
actor_role="operator",
)
assert "- No: 2" in preview.user_message
duplicate = executor.execute(
AssistantDecision(
intent="input_data",
sheet_name="Log",
operation="create",
confidence=95,
data={"No": "1", "Produk": "Oli MPX", "Status": "Selesai"},
),
actor_role="operator",
)
assert duplicate.ok is False
assert "sudah ada di row 2" in duplicate.user_message
def test_executor_autofills_missing_no_gap_before_appending() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
fake.schemas["Log"] = SheetSchema(title="Log", headers=["No", "Produk", "Status"])
fake.rows["Log"] = [
SheetRow(
sheet_name="Log",
row_number=2,
values={"No": "1", "Produk": "Oli MPX", "Status": "Selesai"},
),
SheetRow(
sheet_name="Log",
row_number=4,
values={"No": "3", "Produk": "Ban Luar", "Status": "Belum"},
),
]
executor = OperationExecutor(fake, settings=settings)
preview = executor.execute(
AssistantDecision(
intent="input_data",
sheet_name="Log",
operation="create",
confidence=95,
data={"Produk": "Kampas Rem", "Status": "Belum"},
),
actor_role="operator",
)
assert "- No: 2" in preview.user_message
def test_executor_blocks_invalid_dropdown_value() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
fake.schemas["Log"] = SheetSchema(
title="Log",
headers=["Produk", "Status"],
field_options={"Status": ["Belum", "Selesai"]},
)
fake.rows["Log"] = []
executor = OperationExecutor(fake, settings=settings)
result = executor.execute(
AssistantDecision(
intent="input_data",
sheet_name="Log",
operation="create",
confidence=95,
data={"Produk": "Ban Luar", "Status": "Draft"},
),
actor_role="operator",
)
assert result.ok is False
assert "gak valid" in result.user_message
def test_executor_updates_single_matching_row() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="update_data",
sheet_name="Stok",
operation="update",
confidence=90,
filters={"Produk": "Oli"},
data={"Qty": "8"},
)
result = executor.execute(decision, actor_role="operator")
assert result.ok is True
assert fake.rows["Stok"][0].values["Qty"] == "8"
assert "Bot_Log" in fake.schemas
assert fake.rows["Bot_Log"][-1].values["Aksi"] == "update"
def test_executor_returns_general_chat_without_sheet_target() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
executor = OperationExecutor(FakeSheets(), settings=settings)
decision = AssistantDecision(
intent="chat_umum",
sheet_name="",
operation="chat",
confidence=95,
user_reply="bisaa pren, gw bantu jawab pertanyaan umum juga.",
)
result = executor.execute(decision, actor_role="guest")
assert result.ok is True
assert "pertanyaan umum" in result.user_message
def test_executor_creates_sheet_with_headers() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="struktur_sheet",
sheet_name="Booking",
operation="create_sheet",
confidence=92,
data={"headers": ["Nama", "WA", "Motor"]},
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is True
assert fake.schemas["Booking"].headers == ["Nama", "WA", "Motor"]
def test_executor_updates_sheet_headers() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="struktur_sheet",
sheet_name="Stok",
operation="update_headers",
confidence=92,
data={"headers": "Nama Produk, Harga, Stok"},
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is False
assert result.requires_confirmation is True
assert fake.schemas["Stok"].headers == ["Produk", "Qty", "Status"]
result = executor.execute(decision, actor_role="admin", confirmed=True)
assert result.ok is True
assert fake.schemas["Stok"].headers == ["Nama Produk", "Harga", "Stok"]
def test_executor_renames_sheet() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="struktur_sheet",
sheet_name="Stok",
operation="rename_sheet",
confidence=92,
data={"new_sheet_name": "Inventory"},
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is False
assert result.requires_confirmation is True
assert "Stok" in fake.schemas
result = executor.execute(decision, actor_role="admin", confirmed=True)
assert result.ok is True
assert "Inventory" in fake.schemas
assert "Stok" not in fake.schemas
def test_executor_deletes_full_sheet_only_with_confirmation() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
fake.create_sheet("Booking", ["Nama"])
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="struktur_sheet",
sheet_name="Booking",
operation="delete_sheet",
confidence=92,
data={"confirm_delete": True},
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is False
assert result.requires_confirmation is True
assert "Booking" in fake.schemas
result = executor.execute(decision, actor_role="admin", confirmed=True)
assert result.ok is True
assert "Booking" not in fake.schemas
def test_executor_blocks_full_sheet_delete_without_confirmation() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="struktur_sheet",
sheet_name="Stok",
operation="delete_sheet",
confidence=92,
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is False
assert "permanen" in result.user_message
def test_executor_deletes_row_after_confirmation() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="hapus_data",
sheet_name="Stok",
operation="delete",
confidence=92,
filters={"Produk": "Oli"},
)
result = executor.execute(decision, actor_role="admin")
assert result.ok is False
assert result.requires_confirmation is True
assert fake.rows["Stok"][0].values["Status"] == "active"
result = executor.execute(decision, actor_role="admin", confirmed=True)
assert result.ok is True
assert fake.rows["Stok"][0].values["Status"] == "deleted"
assert fake.rows["Bot_Log"][-1].values["Aksi"] == "delete"
def test_executor_blocks_delete_for_operator_role() -> None:
settings = SimpleNamespace(
require_admin_for_writes=True,
min_write_confidence=70,
min_delete_confidence=80,
safe_delete_strategy="soft",
max_result_rows=10,
)
fake = FakeSheets()
executor = OperationExecutor(fake, settings=settings)
decision = AssistantDecision(
intent="hapus_data",
sheet_name="Stok",
operation="delete",
confidence=92,
filters={"Produk": "Oli"},
)
result = executor.execute(decision, actor_role="operator")
assert result.ok is False
assert "khusus admin" in result.user_message