74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from isphere_ai_bridge.draft import generate_drafts
|
|
from isphere_ai_bridge.local_db import ISphereLocalDbReader, open_readonly_connection
|
|
from isphere_ai_bridge.locator import discover_msglib_databases
|
|
from isphere_ai_bridge.mock_db import create_mock_msglib
|
|
|
|
|
|
class LocalDbPocTests(unittest.TestCase):
|
|
def test_mock_database_reads_normalized_messages(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
|
|
reader = ISphereLocalDbReader(db_path)
|
|
|
|
probe = reader.probe()
|
|
self.assertTrue(probe["can_open_readonly"])
|
|
self.assertIn("tblPersonMsg", probe["message_tables"])
|
|
self.assertIn("tblMsgGroupPersonMsg", probe["message_tables"])
|
|
|
|
messages = reader.read_messages(limit=10)
|
|
self.assertEqual(4, len(messages))
|
|
self.assertEqual("sys-001", messages[0].message_id)
|
|
self.assertEqual("system", messages[0].conversation_type)
|
|
self.assertEqual("grp-001", messages[1].message_id)
|
|
self.assertEqual("group", messages[1].conversation_type)
|
|
self.assertEqual("p2p-002", messages[2].message_id)
|
|
self.assertEqual("direct", messages[2].conversation_type)
|
|
self.assertIn("#tblPersonMsg:p2p-002", messages[2].raw_ref)
|
|
|
|
def test_connection_is_read_only(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
|
|
with open_readonly_connection(db_path) as con:
|
|
with self.assertRaises(sqlite3.OperationalError):
|
|
con.execute("create table should_not_write (id int)")
|
|
|
|
def test_discovers_expected_account_msglib_path(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
base = Path(tmp) / "isphere"
|
|
db_path = (
|
|
base
|
|
/ "connection.10083"
|
|
/ "27000000"
|
|
/ "Account"
|
|
/ "EMP001"
|
|
/ "MsgLib.db"
|
|
)
|
|
create_mock_msglib(db_path)
|
|
|
|
candidates = discover_msglib_databases(base)
|
|
self.assertEqual(1, len(candidates))
|
|
self.assertTrue(candidates[0].exists)
|
|
self.assertEqual("EMP001", candidates[0].account_id)
|
|
self.assertEqual(db_path, candidates[0].path)
|
|
|
|
def test_generates_draft_only_outputs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
|
|
messages = ISphereLocalDbReader(db_path).read_messages(limit=2)
|
|
drafts = generate_drafts(messages)
|
|
|
|
self.assertEqual(2, len(drafts))
|
|
self.assertEqual("draft-only", drafts[0].status)
|
|
self.assertIn("Draft only", drafts[0].draft_text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|