57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from app.services.attachments import attachment_record
|
|
from app.services.storage import WorkspaceStore
|
|
from app.settings import ProviderConfig, ProviderModel, Settings
|
|
|
|
|
|
class ConversationAttachmentStorageTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary_directory = tempfile.TemporaryDirectory()
|
|
temporary_root = Path(self.temporary_directory.name)
|
|
backend_root = Path(__file__).resolve().parents[1]
|
|
provider = ProviderConfig("test", "Test", "https://example.invalid/v1", "test-key", (ProviderModel("test-model"),))
|
|
self.settings = Settings(
|
|
task_root=temporary_root / "tasks",
|
|
conversation_root=temporary_root / "conversations",
|
|
library_root=backend_root / "cdsl_library",
|
|
engine_root=backend_root / "engine" / "cdsl_engine",
|
|
llm_base_url=provider.base_url,
|
|
llm_api_key=provider.api_key,
|
|
llm_model="test-model",
|
|
llm_timeout_s=1,
|
|
default_provider_id="test",
|
|
providers=(provider,),
|
|
)
|
|
self.store = WorkspaceStore(self.settings)
|
|
self.conversation_id = "conv_000000000001"
|
|
self.store.ensure_conversation(self.conversation_id)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary_directory.cleanup()
|
|
|
|
def test_upload_is_owned_by_conversation_without_creating_a_task(self) -> None:
|
|
relative_path, target = self.store.write_conversation_upload(self.conversation_id, "reference.png", b"reference-bytes")
|
|
attachment = attachment_record(self.conversation_id, "reference.png", "image/png", relative_path, b"reference-bytes", "image")
|
|
conversation = self.store.add_conversation_attachment(self.conversation_id, attachment)
|
|
|
|
self.assertEqual(target, (self.settings.conversation_root / self.conversation_id / relative_path).resolve())
|
|
self.assertTrue(target.is_file())
|
|
self.assertEqual(conversation["attachments"], [attachment])
|
|
self.assertEqual(list(self.settings.task_root.glob("cad_*")), [])
|
|
|
|
def test_attachment_cannot_be_linked_to_another_conversation(self) -> None:
|
|
relative_path, _ = self.store.write_conversation_upload(self.conversation_id, "reference.png", b"reference-bytes")
|
|
attachment = attachment_record("conv_000000000002", "reference.png", "image/png", relative_path, b"reference-bytes", "image")
|
|
|
|
with self.assertRaisesRegex(ValueError, "does not belong"):
|
|
self.store.add_conversation_attachment(self.conversation_id, attachment)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|