13b56fd1b0
LibreOffice Calc extension for Silo PLM integration. Uses shared silo-client package (submodule) for API communication. Changes from monorepo version: - SiloClient class removed from client.py, replaced with CalcSiloSettings adapter + factory function wrapping silo_client.SiloClient - silo_calc_component.py adds silo-client to sys.path - Makefile build-oxt copies silo_client into .oxt for self-contained packaging - All other modules unchanged
346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""Basic tests for silo_calc modules (no UNO dependency)."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
# Add pythonpath to sys.path so we can import without LibreOffice
|
|
_pkg_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
_pypath = os.path.join(_pkg_dir, "pythonpath")
|
|
if _pypath not in sys.path:
|
|
sys.path.insert(0, _pypath)
|
|
|
|
from silo_calc import project_files, sync_engine
|
|
from silo_calc import settings as _settings
|
|
from silo_calc import sheet_format as sf
|
|
|
|
|
|
class TestSheetFormat(unittest.TestCase):
|
|
def test_bom_header_counts(self):
|
|
self.assertEqual(len(sf.BOM_VISIBLE_HEADERS), 11)
|
|
self.assertEqual(len(sf.BOM_PROPERTY_HEADERS), 13)
|
|
self.assertEqual(len(sf.BOM_SYNC_HEADERS), 4)
|
|
self.assertEqual(sf.BOM_TOTAL_COLS, 28)
|
|
|
|
def test_column_indices(self):
|
|
self.assertEqual(sf.COL_ITEM, 0)
|
|
self.assertEqual(sf.COL_PN, 3)
|
|
self.assertEqual(sf.COL_UNIT_COST, 6)
|
|
self.assertEqual(sf.COL_QTY, 7)
|
|
self.assertEqual(sf.COL_EXT_COST, 8)
|
|
|
|
def test_detect_sheet_type_bom(self):
|
|
headers = [
|
|
"Item",
|
|
"Level",
|
|
"Source",
|
|
"PN",
|
|
"Description",
|
|
"Seller Description",
|
|
"Unit Cost",
|
|
"QTY",
|
|
"Ext Cost",
|
|
]
|
|
self.assertEqual(sf.detect_sheet_type(headers), "bom")
|
|
|
|
def test_detect_sheet_type_items(self):
|
|
headers = ["PN", "Description", "Type", "Source"]
|
|
self.assertEqual(sf.detect_sheet_type(headers), "items")
|
|
|
|
def test_detect_sheet_type_unknown(self):
|
|
self.assertIsNone(sf.detect_sheet_type([]))
|
|
self.assertIsNone(sf.detect_sheet_type(["Foo", "Bar"]))
|
|
|
|
def test_col_letter(self):
|
|
self.assertEqual(sf.col_letter(0), "A")
|
|
self.assertEqual(sf.col_letter(25), "Z")
|
|
self.assertEqual(sf.col_letter(26), "AA")
|
|
self.assertEqual(sf.col_letter(27), "AB")
|
|
|
|
def test_property_key_map_bidirectional(self):
|
|
for header, key in sf.PROPERTY_KEY_MAP.items():
|
|
self.assertEqual(sf.DB_FIELD_TO_HEADER[key], header)
|
|
|
|
|
|
class TestSyncEngine(unittest.TestCase):
|
|
def _make_row(self, pn="F01-0001", desc="Test", cost="10.00", qty="2"):
|
|
"""Create a minimal BOM row with enough columns."""
|
|
cells = [""] * sf.BOM_TOTAL_COLS
|
|
cells[sf.COL_PN] = pn
|
|
cells[sf.COL_DESCRIPTION] = desc
|
|
cells[sf.COL_UNIT_COST] = cost
|
|
cells[sf.COL_QTY] = qty
|
|
return cells
|
|
|
|
def test_compute_row_hash_deterministic(self):
|
|
row = self._make_row()
|
|
h1 = sync_engine.compute_row_hash(row)
|
|
h2 = sync_engine.compute_row_hash(row)
|
|
self.assertEqual(h1, h2)
|
|
self.assertEqual(len(h1), 64) # SHA-256 hex
|
|
|
|
def test_compute_row_hash_changes(self):
|
|
row1 = self._make_row(cost="10.00")
|
|
row2 = self._make_row(cost="20.00")
|
|
self.assertNotEqual(
|
|
sync_engine.compute_row_hash(row1),
|
|
sync_engine.compute_row_hash(row2),
|
|
)
|
|
|
|
def test_classify_row_new(self):
|
|
row = self._make_row()
|
|
# No stored hash -> new
|
|
self.assertEqual(sync_engine.classify_row(row), sync_engine.STATUS_NEW)
|
|
|
|
def test_classify_row_synced(self):
|
|
row = self._make_row()
|
|
# Set stored hash to current hash
|
|
row[sf.COL_ROW_HASH] = sync_engine.compute_row_hash(row)
|
|
row[sf.COL_ROW_STATUS] = "synced"
|
|
self.assertEqual(sync_engine.classify_row(row), sync_engine.STATUS_SYNCED)
|
|
|
|
def test_classify_row_modified(self):
|
|
row = self._make_row()
|
|
row[sf.COL_ROW_HASH] = sync_engine.compute_row_hash(row)
|
|
# Now change a cell
|
|
row[sf.COL_UNIT_COST] = "99.99"
|
|
self.assertEqual(sync_engine.classify_row(row), sync_engine.STATUS_MODIFIED)
|
|
|
|
def test_classify_rows_skips_header(self):
|
|
header = list(sf.BOM_ALL_HEADERS)
|
|
row1 = self._make_row()
|
|
all_rows = [header, row1]
|
|
classified = sync_engine.classify_rows(all_rows)
|
|
# Header row (index 0) should be skipped
|
|
self.assertEqual(len(classified), 1)
|
|
self.assertEqual(classified[0][0], 1) # row index
|
|
|
|
def test_update_row_sync_state(self):
|
|
row = self._make_row()
|
|
updated = sync_engine.update_row_sync_state(
|
|
row, "synced", updated_at="2025-01-01T00:00:00Z", parent_pn="A01-0003"
|
|
)
|
|
self.assertEqual(updated[sf.COL_ROW_STATUS], "synced")
|
|
self.assertEqual(updated[sf.COL_UPDATED_AT], "2025-01-01T00:00:00Z")
|
|
self.assertEqual(updated[sf.COL_PARENT_PN], "A01-0003")
|
|
# Hash should be set
|
|
self.assertEqual(len(updated[sf.COL_ROW_HASH]), 64)
|
|
|
|
def test_build_push_diff(self):
|
|
row_new = self._make_row(pn="NEW-0001")
|
|
row_synced = self._make_row(pn="F01-0001")
|
|
row_synced[sf.COL_ROW_HASH] = sync_engine.compute_row_hash(row_synced)
|
|
|
|
row_modified = self._make_row(pn="F01-0002")
|
|
row_modified[sf.COL_ROW_HASH] = sync_engine.compute_row_hash(row_modified)
|
|
row_modified[sf.COL_UNIT_COST] = "999.99" # change after hash
|
|
|
|
classified = [
|
|
(1, sync_engine.STATUS_NEW, row_new),
|
|
(2, sync_engine.STATUS_SYNCED, row_synced),
|
|
(3, sync_engine.STATUS_MODIFIED, row_modified),
|
|
]
|
|
diff = sync_engine.build_push_diff(classified)
|
|
self.assertEqual(len(diff["new"]), 1)
|
|
self.assertEqual(len(diff["modified"]), 1)
|
|
self.assertEqual(diff["unchanged"], 1)
|
|
self.assertEqual(len(diff["conflicts"]), 0)
|
|
|
|
def test_conflict_detection(self):
|
|
row = self._make_row(pn="F01-0001")
|
|
row[sf.COL_ROW_HASH] = sync_engine.compute_row_hash(row)
|
|
row[sf.COL_UPDATED_AT] = "2025-01-01T00:00:00Z"
|
|
row[sf.COL_UNIT_COST] = "changed" # local modification
|
|
|
|
classified = [(1, sync_engine.STATUS_MODIFIED, row)]
|
|
diff = sync_engine.build_push_diff(
|
|
classified,
|
|
server_timestamps={
|
|
"F01-0001": "2025-06-01T00:00:00Z"
|
|
}, # server changed too
|
|
)
|
|
self.assertEqual(len(diff["conflicts"]), 1)
|
|
self.assertEqual(len(diff["modified"]), 0)
|
|
|
|
|
|
class TestSettings(unittest.TestCase):
|
|
def test_load_defaults(self):
|
|
# Use a temp dir so we don't touch real settings
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
cfg = _settings.load()
|
|
self.assertEqual(cfg["ssl_verify"], True)
|
|
self.assertEqual(cfg["default_schema"], "kindred-rd")
|
|
|
|
def test_save_and_load(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("api_url", "https://silo.test/api")
|
|
cfg = _settings.load()
|
|
self.assertEqual(cfg["api_url"], "https://silo.test/api")
|
|
|
|
def test_save_auth_and_clear(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.save_auth("testuser", "editor", "local", "silo_abc123")
|
|
cfg = _settings.load()
|
|
self.assertEqual(cfg["auth_username"], "testuser")
|
|
self.assertEqual(cfg["api_token"], "silo_abc123")
|
|
|
|
_settings.clear_auth()
|
|
cfg = _settings.load()
|
|
self.assertEqual(cfg["api_token"], "")
|
|
self.assertEqual(cfg["auth_username"], "")
|
|
|
|
|
|
class TestProjectFiles(unittest.TestCase):
|
|
def test_get_project_sheet_path(self):
|
|
path = project_files.get_project_sheet_path("3DX10")
|
|
self.assertTrue(str(path).endswith("sheets/3DX10/3DX10.ods"))
|
|
|
|
def test_save_and_read(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("projects_dir", tmp)
|
|
|
|
test_data = b"PK\x03\x04fake-ods-content"
|
|
path = project_files.save_project_sheet("TEST", test_data)
|
|
self.assertTrue(path.is_file())
|
|
self.assertEqual(path.name, "TEST.ods")
|
|
|
|
read_back = project_files.read_project_sheet("TEST")
|
|
self.assertEqual(read_back, test_data)
|
|
|
|
def test_list_project_sheets(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("projects_dir", tmp)
|
|
|
|
# Create two project dirs
|
|
for code in ("AAA", "BBB"):
|
|
d = Path(tmp) / "sheets" / code
|
|
d.mkdir(parents=True)
|
|
(d / f"{code}.ods").write_bytes(b"fake")
|
|
|
|
sheets = project_files.list_project_sheets()
|
|
codes = [s[0] for s in sheets]
|
|
self.assertIn("AAA", codes)
|
|
self.assertIn("BBB", codes)
|
|
|
|
|
|
class TestAIClient(unittest.TestCase):
|
|
"""Test ai_client helpers that don't require network or UNO."""
|
|
|
|
def test_default_constants(self):
|
|
from silo_calc import ai_client
|
|
|
|
self.assertTrue(ai_client.OPENROUTER_API_URL.startswith("https://"))
|
|
self.assertTrue(len(ai_client.DEFAULT_MODEL) > 0)
|
|
self.assertTrue(len(ai_client.DEFAULT_INSTRUCTIONS) > 0)
|
|
|
|
def test_is_configured_false_by_default(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
old = os.environ.pop("OPENROUTER_API_KEY", None)
|
|
try:
|
|
self.assertFalse(ai_client.is_configured())
|
|
finally:
|
|
if old is not None:
|
|
os.environ["OPENROUTER_API_KEY"] = old
|
|
|
|
def test_is_configured_with_env_var(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
old = os.environ.get("OPENROUTER_API_KEY")
|
|
os.environ["OPENROUTER_API_KEY"] = "sk-test-key"
|
|
try:
|
|
self.assertTrue(ai_client.is_configured())
|
|
finally:
|
|
if old is not None:
|
|
os.environ["OPENROUTER_API_KEY"] = old
|
|
else:
|
|
os.environ.pop("OPENROUTER_API_KEY", None)
|
|
|
|
def test_is_configured_with_settings(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("openrouter_api_key", "sk-test-key")
|
|
old = os.environ.pop("OPENROUTER_API_KEY", None)
|
|
try:
|
|
self.assertTrue(ai_client.is_configured())
|
|
finally:
|
|
if old is not None:
|
|
os.environ["OPENROUTER_API_KEY"] = old
|
|
|
|
def test_chat_completion_missing_key_raises(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
old = os.environ.pop("OPENROUTER_API_KEY", None)
|
|
try:
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
ai_client.chat_completion([{"role": "user", "content": "test"}])
|
|
self.assertIn("not configured", str(ctx.exception))
|
|
finally:
|
|
if old is not None:
|
|
os.environ["OPENROUTER_API_KEY"] = old
|
|
|
|
def test_get_model_default(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
self.assertEqual(ai_client._get_model(), ai_client.DEFAULT_MODEL)
|
|
|
|
def test_get_model_from_settings(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("openrouter_model", "anthropic/claude-3-haiku")
|
|
self.assertEqual(ai_client._get_model(), "anthropic/claude-3-haiku")
|
|
|
|
def test_get_instructions_default(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
self.assertEqual(
|
|
ai_client._get_instructions(), ai_client.DEFAULT_INSTRUCTIONS
|
|
)
|
|
|
|
def test_get_instructions_from_settings(self):
|
|
from silo_calc import ai_client
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_settings._SETTINGS_DIR = Path(tmp)
|
|
_settings._SETTINGS_FILE = Path(tmp) / "test-settings.json"
|
|
_settings.put("openrouter_instructions", "Custom instructions")
|
|
self.assertEqual(ai_client._get_instructions(), "Custom instructions")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|