opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-20 09:38:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use sqlite3 session in test_workspaces_members (#38775)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-14 16:15:11 +09:00 committed by GitHub
parent e21071cbb8
commit 995de4994d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 213 additions and 215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

428
api/tests/unit_tests/controllers/openapi/test_workspaces_members.py
View File

@ -19,19 +19,21 @@ from __future__ import annotations
import builtins
import json
import sys
import uuid
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import MagicMock, Mock
from unittest.mock import Mock, patch
import pytest
from flask import Flask
from flask.views import MethodView
from pydantic import ValidationError
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from werkzeug.exceptions import BadRequest, NotFound, UnprocessableEntity
from controllers.openapi import bp as openapi_bp
from controllers.openapi import workspaces as workspaces_module
from controllers.openapi._errors import MemberLicenseExceeded, MemberLimitExceeded
from controllers.openapi._models import MemberInvitePayload, MemberRoleUpdatePayload
from controllers.openapi.auth.data import AuthData
@ -41,7 +43,10 @@ from controllers.openapi.workspaces import (
WorkspaceSwitchApi,
)
from libs.oauth_bearer import AuthContext, Scope, SubjectType, TokenType, reset_auth_ctx, set_auth_ctx
from models.account import AccountStatus, TenantAccountRole
from models import Account, Tenant, TenantAccountJoin
from models.account import AccountStatus, TenantAccountRole, TenantStatus
from models.base import TypeBase
from services.account_service import TenantService as RealTenantService
from services.errors.account import (
AccountAlreadyInTenantError,
AccountNotLinkTenantError,
@ -86,6 +91,22 @@ def openapi_app() -> Flask:
return app
@pytest.fixture
def database_session(sqlite_engine: Engine):
models = (Account, Tenant, TenantAccountJoin)
tables = [model.metadata.tables[model.__tablename__] for model in models]
TypeBase.metadata.create_all(sqlite_engine, tables=tables)
database_session = scoped_session(sessionmaker(bind=sqlite_engine, expire_on_commit=False))
try:
with (
patch.object(workspaces_module.db, "session", database_session),
patch("models.account.db", SimpleNamespace(engine=sqlite_engine)),
):
yield database_session()
finally:
database_session.remove()
def _rule(app: Flask, path: str):
return next(r for r in app.url_map.iter_rules() if r.rule == path)
@ -118,62 +139,57 @@ def _auth_data(account_id: uuid.UUID) -> AuthData:
)
def _account(account_id: str = "acct-1", email: str = "u@example.com") -> SimpleNamespace:
return SimpleNamespace(
id=account_id,
name="User",
email=email,
status=AccountStatus.ACTIVE,
avatar=None,
)
def _account(account_id: str = "acct-1", email: str = "u@example.com") -> Account:
account = Account(name="User", email=email, status=AccountStatus.ACTIVE)
account.id = account_id
return account
def _tenant(tenant_id: str = "ws-1") -> SimpleNamespace:
return SimpleNamespace(
id=tenant_id,
name="WS",
status="normal",
created_at=datetime(2026, 5, 18, tzinfo=UTC),
)
def _tenant(tenant_id: str = "ws-1", *, status: TenantStatus = TenantStatus.NORMAL) -> Tenant:
tenant = Tenant(name="WS", status=status)
tenant.id = tenant_id
tenant.created_at = datetime(2026, 5, 18)
return tenant
def _persist_workspace(
session: Session,
workspace_id: str,
memberships: list[tuple[str, str, TenantAccountRole, bool]],
*,
status: TenantStatus = TenantStatus.NORMAL,
) -> tuple[Tenant, list[Account]]:
tenant = _tenant(workspace_id, status=status)
accounts: list[Account] = []
session.add(tenant)
for account_id, email, role, current in memberships:
account = _account(account_id=account_id, email=email)
membership = TenantAccountJoin(
tenant_id=tenant.id,
account_id=account.id,
current=current,
role=role,
)
accounts.append(account)
session.add_all([account, membership])
session.commit()
return tenant, accounts
def _tenant_service(**overrides) -> SimpleNamespace:
"""TenantService double for the workspaces module.
Read getters (`get_tenant_by_id`, `find_workspace_for_account`) delegate
to the session they're handed, so tests keep driving entity loads through
``mock_db.session.get`` / ``.execute`` and their existing side_effect
ordering — the SQL those methods run is covered in test_account_service.py.
Domain mutators default to no-op Mocks; override per test as needed.
"""
"""Retain domain mutator doubles while delegating reads to the real service."""
methods: dict = {
"switch_tenant": Mock(),
"get_tenant_members": Mock(return_value=[]),
"switch_tenant": RealTenantService.switch_tenant,
"get_tenant_members": RealTenantService.get_tenant_members,
"remove_member_from_tenant": Mock(),
"update_member_role": Mock(),
"get_tenant_by_id": lambda tenant_id, *, session: session.get(None, tenant_id),
"find_workspace_for_account": lambda account_id, workspace_id, *, session: session.execute(None).first(),
"get_tenant_by_id": RealTenantService.get_tenant_by_id,
"find_workspace_for_account": RealTenantService.find_workspace_for_account,
}
methods.update(overrides)
return SimpleNamespace(**methods)
def _account_service(**overrides) -> SimpleNamespace:
"""AccountService double; ``get_account_by_id`` delegates to the injected
session (see :func:`_tenant_service`)."""
methods: dict = {
"get_account_by_id": lambda account_id, *, session: session.get(None, account_id),
}
methods.update(overrides)
return SimpleNamespace(**methods)
def _db_mock() -> MagicMock:
mock_db = MagicMock()
mock_db.session.return_value = mock_db.session
return mock_db
# ---------------------------------------------------------------------------
# Route registration
# ---------------------------------------------------------------------------
@ -268,9 +284,7 @@ def test_update_role_rejects_invalid_body_with_422(app: Flask, bypass_pipeline):
# ---------------------------------------------------------------------------
def test_switch_returns_workspace_detail_with_current_true(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch
):
def test_switch_returns_workspace_detail_with_current_true(app: Flask, bypass_pipeline, database_session: Session):
"""Happy path: switch service is called, then the workspace+membership
row is re-queried so the returned `current` reflects post-commit state.
"""
@ -278,18 +292,11 @@ def test_switch_returns_workspace_detail_with_current_true(
acct_id = uuid.uuid4()
api = WorkspaceSwitchApi()
mock_db = _db_mock()
mock_db.session.get.return_value = _account(account_id=str(acct_id))
membership = SimpleNamespace(role=TenantAccountRole.OWNER, current=True)
mock_db.session.execute.return_value.first.return_value = (_tenant(ws_id), membership)
switch_mock = Mock()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"TenantService",
_tenant_service(switch_tenant=switch_mock),
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, False)],
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}:switch", method="POST"):
_seed(_auth_ctx(account_id=acct_id))
@ -298,11 +305,18 @@ def test_switch_returns_workspace_detail_with_current_true(
assert status == 200
assert body["id"] == ws_id
assert body["current"] is True
assert switch_mock.called
membership = database_session.scalar(
select(TenantAccountJoin).where(
TenantAccountJoin.tenant_id == ws_id,
TenantAccountJoin.account_id == str(acct_id),
)
)
assert membership is not None
assert membership.current is True
def test_switch_404s_when_service_raises_account_not_link_tenant(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
"""If switch_tenant raises (e.g. Tenant.status != NORMAL), the body
surfaces as NotFound, not 500."""
@ -310,15 +324,17 @@ def test_switch_404s_when_service_raises_account_not_link_tenant(
acct_id = uuid.uuid4()
api = WorkspaceSwitchApi()
mock_db = _db_mock()
mock_db.session.get.return_value = _account(account_id=str(acct_id))
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, False)],
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"TenantService",
_tenant_service(switch_tenant=Mock(side_effect=AccountNotLinkTenantError("…"))),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}:switch", method="POST"):
_seed(_auth_ctx(account_id=acct_id))
@ -331,29 +347,18 @@ def test_switch_404s_when_service_raises_account_not_link_tenant(
# ---------------------------------------------------------------------------
def test_members_list_returns_normalized_rows(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_members_list_returns_normalized_rows(app: Flask, bypass_pipeline, database_session: Session):
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
member_id = str(uuid.uuid4())
api = WorkspaceMembersApi()
member = SimpleNamespace(
id="m-1",
name="Mia",
email="mia@example.com",
status=AccountStatus.ACTIVE,
avatar=None,
role=TenantAccountRole.ADMIN,
_, members = _persist_workspace(
database_session,
ws_id,
[(member_id, "mia@example.com", TenantAccountRole.ADMIN, False)],
)
mock_db = _db_mock()
mock_db.session.get.return_value = _tenant(ws_id)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"TenantService",
_tenant_service(get_tenant_members=Mock(return_value=[member])),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
members[0].name = "Mia"
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}/members"):
_seed(_auth_ctx(account_id=acct_id))
@ -369,33 +374,15 @@ def test_members_list_returns_normalized_rows(app: Flask, bypass_pipeline, monke
assert body["data"][0]["status"] == "active"
def test_members_list_paginates_with_query_params(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_members_list_paginates_with_query_params(app: Flask, bypass_pipeline, database_session: Session):
"""`?page=2&limit=2` slices service output and reports total/has_more."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
members = [
SimpleNamespace(
id=f"m-{i}",
name=f"User {i}",
email=f"u{i}@example.com",
status=AccountStatus.ACTIVE,
avatar=None,
role=TenantAccountRole.NORMAL,
)
for i in range(5)
]
mock_db = _db_mock()
mock_db.session.get.return_value = _tenant(ws_id)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"TenantService",
_tenant_service(get_tenant_members=Mock(return_value=members)),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
member_ids = [str(uuid.uuid4()) for _ in range(5)]
memberships = [(member_ids[i], f"u{i}@example.com", TenantAccountRole.NORMAL, False) for i in range(5)]
_persist_workspace(database_session, ws_id, memberships)
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}/members?page=2&limit=2"):
_seed(_auth_ctx(account_id=acct_id))
@ -406,19 +393,15 @@ def test_members_list_paginates_with_query_params(app: Flask, bypass_pipeline, m
assert body["limit"] == 2
assert body["total"] == 5
assert body["has_more"] is True
assert [d["id"] for d in body["data"]] == ["m-2", "m-3"]
assert [d["id"] for d in body["data"]] == member_ids[2:4]
def test_members_list_rejects_unknown_query_param(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_members_list_rejects_unknown_query_param(app: Flask, bypass_pipeline):
"""Strict (`extra='forbid'`) — typos like `?pg=2` surface as 422 (unified via @accepts)."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
mock_db = _db_mock()
mock_db.session.get.return_value = _tenant(ws_id)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}/members?pg=2"):
_seed(_auth_ctx(account_id=acct_id))
with pytest.raises(UnprocessableEntity):
@ -431,29 +414,26 @@ def test_members_list_rejects_unknown_query_param(app: Flask, bypass_pipeline, m
def test_invite_happy_path_returns_invite_url_and_member_id(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
invited = _account(account_id="new-1", email="new@example.com")
mock_db = _db_mock()
# session.get is called twice: once for inviter Account, once for Tenant
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
invited_id = str(uuid.uuid4())
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
database_session.add(_account(account_id=invited_id, email="new@example.com"))
database_session.commit()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(invite_new_member=Mock(return_value="tok-123")),
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"AccountService",
_account_service(get_account_by_email_with_case_fallback=Mock(return_value=invited)),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members",
@ -468,7 +448,7 @@ def test_invite_happy_path_returns_invite_url_and_member_id(
assert body["result"] == "success"
assert body["email"] == "new@example.com"
assert body["role"] == "normal"
assert body["member_id"] == "new-1"
assert body["member_id"] == invited_id
assert "token=tok-123" in body["invite_url"]
assert "email=new%40example.com" in body["invite_url"]
assert body["tenant_id"] == ws_id
@ -514,24 +494,28 @@ def _invite_request(app, ws_id: str, acct_id: uuid.UUID):
)
def test_invite_blocked_by_saas_members_cap(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_invite_blocked_by_saas_members_cap(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
"""SaaS billing plan member cap → MemberLimitExceeded (403)."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
invite_mock = Mock()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(invite_new_member=invite_mock),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"FeatureService",
SimpleNamespace(
get_features=Mock(
@ -548,7 +532,9 @@ def test_invite_blocked_by_saas_members_cap(app: Flask, bypass_pipeline, monkeyp
invite_mock.assert_not_called()
def test_invite_blocked_by_ee_workspace_members_license(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_invite_blocked_by_ee_workspace_members_license(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
"""EE License workspace_members cap → MemberLicenseExceeded (403).
Note: billing.enabled is False (EE without SaaS billing); only the
@ -558,18 +544,20 @@ def test_invite_blocked_by_ee_workspace_members_license(app: Flask, bypass_pipel
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
invite_mock = Mock()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(invite_new_member=invite_mock),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"FeatureService",
SimpleNamespace(
get_features=Mock(
@ -590,30 +578,31 @@ def test_invite_blocked_by_ee_workspace_members_license(app: Flask, bypass_pipel
invite_mock.assert_not_called()
def test_invite_ce_passes_when_both_caps_disabled(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_invite_ce_passes_when_both_caps_disabled(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
"""CE deployment (no billing, no license) → quota gate is a no-op,
invite proceeds normally."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
invited = _account(account_id="new-1", email="new@example.com")
mock_db = _db_mock()
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
invited_id = str(uuid.uuid4())
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
database_session.add(_account(account_id=invited_id, email="new@example.com"))
database_session.commit()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(invite_new_member=Mock(return_value="tok-ce")),
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"AccountService",
_account_service(get_account_by_email_with_case_fallback=Mock(return_value=invited)),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"FeatureService",
SimpleNamespace(get_features=Mock(return_value=_features())), # all defaults
)
@ -626,20 +615,24 @@ def test_invite_ce_passes_when_both_caps_disabled(app: Flask, bypass_pipeline, m
assert body["email"] == "new@example.com"
def test_invite_400_when_already_in_tenant(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_invite_400_when_already_in_tenant(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(invite_new_member=Mock(side_effect=AccountAlreadyInTenantError("already in tenant"))),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members",
@ -657,25 +650,28 @@ def test_invite_400_when_already_in_tenant(app: Flask, bypass_pipeline, monkeypa
# ---------------------------------------------------------------------------
def test_delete_member_happy_path(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_delete_member_happy_path(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
ws_id, member_id = str(uuid.uuid4()), str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMemberApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [
_account(account_id=str(acct_id)), # operator
_tenant(ws_id), # tenant
_account(account_id=member_id), # target member
]
_persist_workspace(
database_session,
ws_id,
[
(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True),
(member_id, "member@example.com", TenantAccountRole.NORMAL, False),
],
)
remove_mock = Mock()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"TenantService",
_tenant_service(remove_member_from_tenant=remove_mock),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members/{member_id}",
@ -699,24 +695,27 @@ def test_delete_member_happy_path(app: Flask, bypass_pipeline, monkeypatch: pyte
(MemberNotInTenantError("not in tenant"), NotFound),
],
)
def test_delete_member_exception_mapping(app: Flask, bypass_pipeline, monkeypatch, exc, expected):
def test_delete_member_exception_mapping(
app: Flask, bypass_pipeline, monkeypatch, exc, expected, database_session: Session
):
ws_id, member_id = str(uuid.uuid4()), str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMemberApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [
_account(account_id=str(acct_id)),
_tenant(ws_id),
_account(account_id=member_id),
]
_persist_workspace(
database_session,
ws_id,
[
(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True),
(member_id, "member@example.com", TenantAccountRole.NORMAL, False),
],
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"TenantService",
_tenant_service(remove_member_from_tenant=Mock(side_effect=exc)),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members/{member_id}",
@ -732,18 +731,16 @@ def test_delete_member_exception_mapping(app: Flask, bypass_pipeline, monkeypatc
)
def test_delete_member_404_when_member_missing(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_delete_member_404_when_member_missing(app: Flask, bypass_pipeline, database_session: Session):
ws_id, member_id = str(uuid.uuid4()), str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMemberApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [
_account(account_id=str(acct_id)),
_tenant(ws_id),
None, # member not found
]
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members/{member_id}",
@ -764,25 +761,28 @@ def test_delete_member_404_when_member_missing(app: Flask, bypass_pipeline, monk
# ---------------------------------------------------------------------------
def test_update_role_happy_path(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_update_role_happy_path(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
ws_id, member_id = str(uuid.uuid4()), str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMemberApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [
_account(account_id=str(acct_id)),
_tenant(ws_id),
_account(account_id=member_id),
]
_persist_workspace(
database_session,
ws_id,
[
(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True),
(member_id, "member@example.com", TenantAccountRole.NORMAL, False),
],
)
update_mock = Mock()
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"TenantService",
_tenant_service(update_member_role=update_mock),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members/{member_id}",
@ -810,24 +810,27 @@ def test_update_role_happy_path(app: Flask, bypass_pipeline, monkeypatch: pytest
(MemberNotInTenantError("not in tenant"), NotFound),
],
)
def test_update_role_exception_mapping(app: Flask, bypass_pipeline, monkeypatch, exc, expected):
def test_update_role_exception_mapping(
app: Flask, bypass_pipeline, monkeypatch, exc, expected, database_session: Session
):
ws_id, member_id = str(uuid.uuid4()), str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMemberApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [
_account(account_id=str(acct_id)),
_tenant(ws_id),
_account(account_id=member_id),
]
_persist_workspace(
database_session,
ws_id,
[
(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True),
(member_id, "member@example.com", TenantAccountRole.NORMAL, False),
],
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"TenantService",
_tenant_service(update_member_role=Mock(side_effect=exc)),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members/{member_id}",
@ -850,22 +853,13 @@ def test_update_role_exception_mapping(app: Flask, bypass_pipeline, monkeypatch,
# ---------------------------------------------------------------------------
def test_load_tenant_rejects_archived_workspace(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_load_tenant_rejects_archived_workspace(app: Flask, bypass_pipeline, database_session: Session):
"""Member management against an archived workspace → 404."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
archived = SimpleNamespace(id=ws_id, name="WS", status="archive", created_at=datetime(2026, 5, 18, tzinfo=UTC))
mock_db = _db_mock()
mock_db.session.get.return_value = archived
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
"TenantService",
_tenant_service(),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
_persist_workspace(database_session, ws_id, [], status=TenantStatus.ARCHIVE)
with app.test_request_context(f"/openapi/v1/workspaces/{ws_id}/members"):
_seed(_auth_ctx(account_id=acct_id))
@ -878,23 +872,27 @@ def test_load_tenant_rejects_archived_workspace(app: Flask, bypass_pipeline, mon
# ---------------------------------------------------------------------------
def test_invite_400_when_register_error(app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch):
def test_invite_400_when_register_error(
app: Flask, bypass_pipeline, monkeypatch: pytest.MonkeyPatch, database_session: Session
):
"""AccountRegisterError (frozen email, workspace creation blocked) → 400."""
ws_id = str(uuid.uuid4())
acct_id = uuid.uuid4()
api = WorkspaceMembersApi()
mock_db = _db_mock()
mock_db.session.get.side_effect = [_account(account_id=str(acct_id)), _tenant(ws_id)]
_persist_workspace(
database_session,
ws_id,
[(str(acct_id), "caller@example.com", TenantAccountRole.OWNER, True)],
)
monkeypatch.setattr(
sys.modules["controllers.openapi.workspaces"],
workspaces_module,
"RegisterService",
SimpleNamespace(
invite_new_member=Mock(side_effect=AccountRegisterError("Workspace is not allowed to create.")),
),
)
monkeypatch.setattr(sys.modules["controllers.openapi.workspaces"], "db", mock_db)
with app.test_request_context(
f"/openapi/v1/workspaces/{ws_id}/members",