Xiaoxia AI
57 lines
2.4 KiB
Python
57 lines
2.4 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from packages.adapters.sqlalchemy_impl.models import Base
|
|
from packages.adapters.sqlalchemy_impl.user_repository import SQLAlchemyUserRepository
|
|
from packages.adapters.sqlalchemy_impl.workspace_member_repository import SQLAlchemyWorkspaceMemberRepository
|
|
from packages.adapters.sqlalchemy_impl.workspace_repository import SQLAlchemyWorkspaceRepository
|
|
from packages.application.workspace.create_workspace_use_case import CreateWorkspaceRequest, CreateWorkspaceUseCase
|
|
from packages.application.workspace.list_workspaces_use_case import ListWorkspacesRequest, ListWorkspacesUseCase
|
|
from packages.domain.auth import password_hasher
|
|
from packages.domain.entities import User
|
|
|
|
|
|
def test_workspace_create_and_list_use_cases_with_sqlalchemy_repositories():
|
|
session = _sqlite_session()
|
|
user_repository = SQLAlchemyUserRepository(session)
|
|
workspace_repository = SQLAlchemyWorkspaceRepository(session)
|
|
workspace_member_repository = SQLAlchemyWorkspaceMemberRepository(session)
|
|
user_repository.save(
|
|
User(
|
|
id="user-1",
|
|
email="user@example.com",
|
|
username="user",
|
|
display_name="User",
|
|
password_hash=password_hasher.hash_password("Password1"),
|
|
)
|
|
)
|
|
|
|
create_response, create_error = CreateWorkspaceUseCase(
|
|
workspace_repository=workspace_repository,
|
|
workspace_member_repository=workspace_member_repository,
|
|
user_repository=user_repository,
|
|
).execute(CreateWorkspaceRequest(name="Main Workspace", owner_user_id="user-1", subscription_plan="free"))
|
|
|
|
assert create_error is None
|
|
assert create_response is not None
|
|
assert create_response.name == "Main Workspace"
|
|
|
|
list_response, list_error = ListWorkspacesUseCase(
|
|
workspace_repository=workspace_repository,
|
|
workspace_member_repository=workspace_member_repository,
|
|
).execute(ListWorkspacesRequest(user_id="user-1"))
|
|
|
|
assert list_error is None
|
|
assert list_response is not None
|
|
assert len(list_response.workspaces) == 1
|
|
assert list_response.workspaces[0].workspace_id == create_response.workspace_id
|
|
assert list_response.workspaces[0].user_role == "owner"
|
|
assert list_response.workspaces[0].member_count == 1
|
|
|
|
|
|
def _sqlite_session():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
session_factory = sessionmaker(bind=engine)
|
|
return session_factory()
|