javadandCursor
b93ecf1691
Keep source, config, and deploy files. Leave runtime data, uploads, and logs out of git. Co-authored-by: Cursor <cursoragent@cursor.com>
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
import pytest
|
|
|
|
from kamidso.contracts.user import CurrentUser
|
|
from kamidso.repositories.store import RepositoryStore
|
|
from kamidso.resources.files import FileService
|
|
from kamidso.settings.flags import load_kamidso_settings
|
|
from kamidso.storage.local import LocalStorage
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_and_list_file(tmp_path):
|
|
settings = load_kamidso_settings()
|
|
storage_path = str(tmp_path / "storage")
|
|
store = RepositoryStore()
|
|
storage = LocalStorage(storage_path)
|
|
service = FileService(store, storage, settings)
|
|
user = CurrentUser(
|
|
id="usr_test",
|
|
email="test@kamidso.test",
|
|
name="Test",
|
|
role="developer",
|
|
workspace_id="ws_test",
|
|
)
|
|
doc = await service.upload(user, "sample.csv", b"a,b\n1,2\n", mime_type="text/csv")
|
|
assert doc["_id"].startswith("file_")
|
|
assert doc["storage_key"] == f"uploads/{user.workspace_id}/{doc['_id']}/sample.csv"
|
|
files = await service.list_files(user)
|
|
assert len(files) == 1
|
|
preview = await service.preview(user, doc["_id"])
|
|
assert preview["type"] == "tabular"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_materialize_legacy_data_storage_key(tmp_path):
|
|
storage = LocalStorage(str(tmp_path / "data"))
|
|
legacy = (
|
|
tmp_path
|
|
/ "data"
|
|
/ "storage"
|
|
/ "ws_1"
|
|
/ "files"
|
|
/ "file_abc"
|
|
/ "Book1 (1).xlsx"
|
|
)
|
|
legacy.parent.mkdir(parents=True)
|
|
legacy.write_bytes(b"xlsx-bytes")
|
|
dest = tmp_path / "runtime" / "Book1 (1).xlsx"
|
|
await storage.materialize("ws_1/files/file_abc/Book1 (1).xlsx", str(dest))
|
|
assert dest.read_bytes() == b"xlsx-bytes"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_stores_title_and_kind(tmp_path):
|
|
settings = load_kamidso_settings()
|
|
store = RepositoryStore()
|
|
storage = LocalStorage(str(tmp_path / "storage"))
|
|
service = FileService(store, storage, settings)
|
|
user = CurrentUser(
|
|
id="usr_test",
|
|
email="test@kamidso.test",
|
|
name="Test",
|
|
role="developer",
|
|
workspace_id="ws_test",
|
|
)
|
|
doc = await service.upload(
|
|
user,
|
|
"Book1 (1).xlsx",
|
|
b"not-a-real-xlsx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
title="فروش ۱۴۰۳",
|
|
)
|
|
public = FileService.to_public(doc)
|
|
assert public["title"] == "فروش ۱۴۰۳"
|
|
assert public["kind"] == "excel"
|
|
assert public["original_name"] == "Book1 (1).xlsx"
|
|
|
|
renamed = await service.update_file(user, doc["_id"], title=" گزارش فروش ")
|
|
assert FileService.to_public(renamed)["title"] == "گزارش فروش"
|