fix: avoid login tenant context import shadowing

Author: zgqgit

src/backend/bisheng/user/domain/services/user.py

@@ -387,8 +387,6 @@ async def _reject_login_if_user_has_no_usable_access(
         # qualifies for login. Without bypass, every DAO below trips
         # NoTenantContextError because do_orm_execute can't infer a tenant for
         # tenant-aware tables (userrole/department/usergroup/roleaccess).
-        from bisheng.core.context.tenant import bypass_tenant_filter
-
         with bypass_tenant_filter():
             if role_ids is None:
                 roles = await UserRoleDao.aget_user_roles(db_user.user_id)
@@ -479,7 +477,10 @@ async def user_login(cls, request: Request, user: UserLogin, auth_jwt: AuthJwt =
                 NoTenantsAvailableError,
                 TenantDisabledError,
             )
-            from bisheng.core.context.tenant import DEFAULT_TENANT_ID, bypass_tenant_filter
+            from bisheng.database.models.department import (
+                UserDepartmentDao,
+            )
+            from bisheng.database.models.tenant import TenantDao, UserTenantDao
 
             # Reject login when the user's primary department mounts to a
             # disabled child tenant. TenantResolver intentionally walks past
@@ -488,11 +489,6 @@ async def user_login(cls, request: Request, user: UserLogin, auth_jwt: AuthJwt =
             # treats it as a member-level freeze. Without this guard a member
             # of a disabled tenant would get fallback-routed to Root and log
             # in normally, which contradicts the operator's intent.
-            from bisheng.database.models.department import (
-                UserDepartmentDao,
-            )
-            from bisheng.database.models.tenant import TenantDao, UserTenantDao
-
             with bypass_tenant_filter():
                 primary_dept = await UserDepartmentDao.aget_user_primary_department(
                     db_user.user_id,
@@ -571,7 +567,6 @@ async def user_login(cls, request: Request, user: UserLogin, auth_jwt: AuthJwt =
         # DB.status is authoritative; Redis blacklist is a defensive cross-check.
         if settings.multi_tenant.enabled and tenant_id and tenant_id > 0:
             from bisheng.common.errcode.tenant import TenantDisabledError
-            from bisheng.core.context.tenant import bypass_tenant_filter
             from bisheng.database.models.tenant import TenantDao
 
             with bypass_tenant_filter():
@@ -656,7 +651,6 @@ async def user_login(cls, request: Request, user: UserLogin, auth_jwt: AuthJwt =
             extra_fields["tenant_id"] = tenant_id
             tenant_info = next((t for t in (tenants_list or []) if t.get("tenant_id") == tenant_id), None)
             if not tenant_info and settings.multi_tenant.enabled:
-                from bisheng.core.context.tenant import bypass_tenant_filter
                 from bisheng.database.models.tenant import TenantDao
 
                 with bypass_tenant_filter():

src/backend/test/user/test_user_access_request_dedup.py

@@ -105,3 +105,65 @@ async def test_login_access_guard_reuses_resolved_roles_and_department_flag():
         role_ids=[2],
         is_department_admin=False,
     )
+
+
+async def test_user_login_enters_tenant_bypass_before_multi_tenant_resolution():
+    db_user = SimpleNamespace(user_id=12, delete=0)
+    rejected_response = object()
+    login_request = SimpleNamespace(user_name="alice", password="encrypted")
+    captcha_setting = AsyncMock(return_value=False)
+
+    with (
+        patch.object(
+            user_service_module,
+            "settings",
+            SimpleNamespace(aget_from_db=captcha_setting),
+        ),
+        patch.object(
+            user_service_module.UserDao,
+            "aget_login_candidates_by_account",
+            new_callable=AsyncMock,
+            return_value=[db_user],
+        ),
+        patch.object(
+            user_service_module.UserService,
+            "decrypt_md5_password",
+            return_value="password",
+        ),
+        patch.object(
+            user_service_module.UserService,
+            "judge_user_password",
+            new_callable=AsyncMock,
+        ),
+        patch.object(
+            user_service_module.UserService,
+            "clear_error_password_key",
+            new_callable=AsyncMock,
+        ),
+        patch.object(
+            user_service_module.UserRoleDao,
+            "aget_user_roles",
+            new_callable=AsyncMock,
+            return_value=[],
+        ),
+        patch.object(
+            user_service_module.DepartmentDao,
+            "aget_user_admin_departments",
+            new_callable=AsyncMock,
+            return_value=[],
+        ),
+        patch.object(
+            user_service_module.UserService,
+            "_reject_login_if_user_has_no_usable_access",
+            new_callable=AsyncMock,
+            return_value=rejected_response,
+        ),
+    ):
+        result = await user_service_module.UserService.user_login(
+            SimpleNamespace(),
+            login_request,
+            auth_jwt=SimpleNamespace(),
+        )
+
+    assert result is rejected_response
+    captcha_setting.assert_awaited_once_with("use_captcha")