javadandCursor
b93ecf1691
Keep source, config, and deploy files. Leave runtime data, uploads, and logs out of git. Co-authored-by: Cursor <cursoragent@cursor.com>
78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
import pytest
|
|
|
|
from kamidso.repositories.store import RepositoryStore
|
|
from kamidso.settings.flags import load_kamidso_settings
|
|
from kamidso.workspace.service import WorkspaceService, hash_password, verify_password
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_identity_creates_admin():
|
|
store = RepositoryStore()
|
|
settings = load_kamidso_settings()
|
|
service = WorkspaceService(store, settings)
|
|
await service.bootstrap_identity()
|
|
user = await store.get_user_by_email(settings.bootstrap_admin_email)
|
|
assert user is not None
|
|
assert user["role"] == "developer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workspace_isolation_for_files():
|
|
store = RepositoryStore()
|
|
file_doc = await store.create_file(
|
|
{"workspace_id": "ws_a", "name": "a.xlsx", "storage_key": "k", "status": "ready"}
|
|
)
|
|
assert await store.get_file("ws_a", file_doc["_id"]) is not None
|
|
assert await store.get_file("ws_b", file_doc["_id"]) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_user_by_mobile_isolates_workspaces():
|
|
store = RepositoryStore()
|
|
settings = load_kamidso_settings()
|
|
service = WorkspaceService(store, settings)
|
|
first = await service.get_or_create_user_by_mobile("09120000001")
|
|
second = await service.get_or_create_user_by_mobile("09120000002")
|
|
again = await service.get_or_create_user_by_mobile("09120000001")
|
|
assert first.workspace_id != second.workspace_id
|
|
assert first.id == again.id
|
|
assert first.workspace_id == again.workspace_id
|
|
assert first.role == "admin"
|
|
assert first.mobile == "09120000001"
|
|
assert first.permissions["can_manage_datasources"] is True
|
|
|
|
|
|
def test_password_hash_roundtrip():
|
|
digest, salt = hash_password("secret")
|
|
assert verify_password("secret", digest, salt)
|
|
assert not verify_password("wrong", digest, salt)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_latest_run_prefers_newest_started_at():
|
|
from kamidso.contracts.run import KamidsoRun
|
|
|
|
store = RepositoryStore()
|
|
older = KamidsoRun(
|
|
id="run_old",
|
|
conversation_id="c1",
|
|
user_id="u1",
|
|
workspace_id="ws_a",
|
|
status="completed",
|
|
started_at="2026-01-01T00:00:00+00:00",
|
|
)
|
|
newer = KamidsoRun(
|
|
id="run_new",
|
|
conversation_id="c1",
|
|
user_id="u1",
|
|
workspace_id="ws_a",
|
|
status="running",
|
|
started_at="2026-01-02T00:00:00+00:00",
|
|
)
|
|
await store.create_run(older)
|
|
await store.create_run(newer)
|
|
latest = await store.get_latest_run("ws_a", "c1")
|
|
assert latest is not None
|
|
assert latest.id == "run_new"
|
|
assert await store.get_latest_run("ws_b", "c1") is None
|