from __future__ import annotations from datetime import timedelta from types import SimpleNamespace import pytest from aiohttp import CookieJar from aiohttp.test_utils import TestClient, TestServer from pyrogram.enums import ChatMemberStatus async def _approved_teacher(directory_db, *, user_id: int = 200): await directory_db.upsert_directory_identity( user_id=user_id, username=f"teacher{user_id}", first_name="王", last_name="技师", ) await directory_db.save_directory_location( user_id=user_id, longitude=116.4074, latitude=39.9042, source="test", actor_id=user_id, actor_name="王技师", ) _, created = await directory_db.submit_teacher_application( user_id=user_id, source="test", bot_id="primary", chat_id=-100200, chat_title="测试群", ) assert created is True await directory_db.decide_teacher_application( user_id=user_id, action="approve", actor_id="admin", actor_name="admin", source="test", ) await directory_db.set_teacher_state( user_id=user_id, action="list", actor_id=user_id, actor_name="王技师", source="test", ) return await directory_db.get_directory_profile(user_id) async def test_teacher_application_distance_listing_and_presence(app_modules): directory_db = app_modules.load("wbb.utils.dbdirectory") directory_service = app_modules.load("wbb.services.directory") await directory_db.ensure_directory_indexes() approved = await _approved_teacher(directory_db) assert approved["application_source_chat_id"] == -100200 assert approved["application_source_chat_title"] == "测试群" profile, applied = await directory_db.set_teacher_state( user_id=200, action="online", actor_id=200, actor_name="王技师", source="test", idempotency_key="presence:one", ) assert applied is True assert directory_db.teacher_is_online(profile) is True duplicate, applied = await directory_db.set_teacher_state( user_id=200, action="online", actor_id=200, actor_name="王技师", source="test", idempotency_key="presence:one", ) assert applied is False assert directory_db.teacher_is_online(duplicate) is True items, total = await directory_db.list_directory_teachers( longitude=116.4174, latitude=39.9042, page=1, page_size=10, ) assert total == 1 public = directory_service.public_teacher(items[0]) assert public["display_name"] == "王 技师" assert 800 < public["distance_meters"] < 1000 assert "longitude" not in public assert "latitude" not in public await app_modules.wbb.control_db.directory_profiles.update_one( {"user_id": 200}, { "$set": { "online_until": directory_db.utc_now() - timedelta(seconds=1), } }, ) assert await directory_db.expire_teacher_presence() == 1 expired = await directory_db.get_directory_profile(200) assert directory_db.teacher_is_online(expired) is False assert expired["listed"] is True async def test_location_only_keeps_coordinates_and_source(app_modules): directory_db = app_modules.load("wbb.utils.dbdirectory") await directory_db.upsert_directory_identity( user_id=300, username="member300", first_name="李", last_name="同学", ) location = await directory_db.save_directory_location( user_id=300, longitude=121.4737, latitude=31.2304, source="test", actor_id=300, actor_name="李同学", ) assert location["source"] == "test" assert location["longitude"] == pytest.approx(121.4737) assert location["latitude"] == pytest.approx(31.2304) assert "region" not in location assert "geocode_consent" not in location assert "geocode_status" not in location assert "coordinate_key" not in location await app_modules.wbb.control_db.directory_locations.update_one( {"user_id": 300}, { "$set": { "region": {"label": "旧区域"}, "geocode_consent": True, "geocode_status": "completed", "coordinate_key": "legacy", } }, ) updated = await directory_db.save_directory_location( user_id=300, longitude=121.48, latitude=31.24, source="test", actor_id=300, actor_name="李同学", ) assert "region" not in updated assert "geocode_consent" not in updated assert "geocode_status" not in updated assert "coordinate_key" not in updated async def test_clearing_legacy_coordinates_keeps_public_teacher_state(app_modules): directory_db = app_modules.load("wbb.utils.dbdirectory") await directory_db.upsert_directory_identity( user_id=350, username="teacher350", first_name="公开", last_name="技师", ) await directory_db.save_directory_location( user_id=350, longitude=121.48, latitude=31.24, source="test", actor_id=350, actor_name="公开技师", ) await app_modules.wbb.control_db.directory_profiles.update_one( {"user_id": 350}, { "$set": { "listed": True, "presence_status": "online", "online_until": directory_db.utc_now() + timedelta(hours=1), } }, ) assert await directory_db.clear_directory_location( user_id=350, actor_id=350, actor_name="公开技师", source="test", reason="不再保存精确坐标", ) profile = await directory_db.get_directory_profile(350) assert profile["listed"] is True assert profile["presence_status"] == "online" assert profile["online_until"] is not None async def test_cross_bot_membership_verification_uses_live_status(app_modules): directory_db = app_modules.load("wbb.utils.dbdirectory") directory_service = app_modules.load("wbb.services.directory") user = SimpleNamespace( id=400, username="member400", first_name="测试", last_name="成员", is_bot=False, ) app_modules.app.members[(-100123, 400)] = SimpleNamespace( status=ChatMemberStatus.ADMINISTRATOR, user=user, ) await directory_db.upsert_directory_membership( bot_id="primary", chat_id=-100123, user_id=400, status="member", active=True, ) member = await directory_service.verify_platform_membership(user_id=400) assert member["allowed"] is True assert member["chat_id"] == -100123 admin = await directory_service.verify_platform_membership( user_id=400, require_admin=True, ) assert admin["allowed"] is True app_modules.app.members[(-100123, 400)] = SimpleNamespace( status=ChatMemberStatus.LEFT, user=user, ) denied = await directory_service.verify_platform_membership(user_id=400) assert denied["allowed"] is False async def test_directory_admin_api_requires_role_and_manages_global_data(app_modules): admin_api = app_modules.load("wbb.admin.api") directory_db = app_modules.load("wbb.utils.dbdirectory") app_modules.wbb.BOT_PERMISSIONS = {"teacher_directory.manage"} application = admin_api.build_admin_application() await application["admin_api"].initialize() client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True)) await client.start_server() try: login = await client.post( "/api/admin/v1/auth/login", json={"username": "admin", "password": "qwe0.123456"}, ) login_data = (await login.json())["data"] changed = await client.put( "/api/admin/v1/auth/password", headers={"X-CSRF-Token": login_data["csrf_token"]}, json={ "current_password": "qwe0.123456", "new_password": "changed-pass-123", }, ) csrf = (await changed.json())["data"]["csrf_token"] headers = {"X-CSRF-Token": csrf} await directory_db.upsert_directory_identity( user_id=500, username="student500", first_name="周", last_name="同学", ) updated = await client.put( "/api/admin/v1/directory/locations/500", headers=headers, json={ "longitude": 113.2644, "latitude": 23.1291, "reason": "修正测试位置", "confirm": True, }, ) assert updated.status == 200 payload = (await updated.json())["data"] assert payload["longitude"] == pytest.approx(113.2644) assert payload["source"] == "web" assert "region" not in payload assert "geocode_status" not in payload listed = await client.get("/api/admin/v1/directory/locations") assert listed.status == 200 item = (await listed.json())["data"]["items"][0] assert item["latitude"] == pytest.approx(23.1291) assert "point" not in item assert "coordinate_key" not in item assert "region" not in item assert "geocode_consent" not in item settings = await client.get("/api/admin/v1/directory/settings") assert settings.status == 200 settings_data = (await settings.json())["data"] assert settings_data["nearby_radius_km"] == 20 assert "nominatim_enabled" not in settings_data unconfirmed = await client.delete( "/api/admin/v1/directory/locations/500", headers=headers, json={"reason": "测试"}, ) assert unconfirmed.status == 409 deleted = await client.delete( "/api/admin/v1/directory/locations/500", headers=headers, json={"reason": "删除测试位置", "confirm": True}, ) assert deleted.status == 200 assert (await deleted.json())["data"]["deleted"] is True app_modules.wbb.BOT_PERMISSIONS = set() denied = await client.get("/api/admin/v1/directory/settings") assert denied.status == 403 assert (await denied.json())["error"]["details"]["required_permission"] == ( "teacher_directory.manage" ) finally: await client.close() async def test_role_specific_bottom_menus_are_chinese(app_modules): module = app_modules.load("wbb.modules.teacher_directory") directory_service = app_modules.load("wbb.services.directory") try: student = module._directory_keyboard( {"application_status": "none"}, is_admin=False, ) student_labels = [ button.text for row in student.inline_keyboard for button in row ] assert "浏览技师" in student_labels assert "Telegram 定位" not in student_labels assert "手动选择位置" not in student_labels assert "清除位置" not in student_labels assert "申请成为技师" in student_labels assert "我的状态" not in student_labels assert "上线" not in student_labels assert "下线" not in student_labels assert "技师审批" not in student_labels teacher_admin = module._directory_keyboard( { "application_status": "approved", "listed": True, "online": True, }, is_admin=True, ) teacher_labels = [ button.text for row in teacher_admin.inline_keyboard for button in row ] assert "下榜" in teacher_labels assert "🔴 下线" in teacher_labels assert "技师审批" in teacher_labels group_student = module._group_directory_keyboard( {"application_status": "none"}, chat_id=-100123, is_admin=False, ) group_student_labels = [ button.text for row in group_student.inline_keyboard for button in row ] assert "浏览技师" in group_student_labels assert "申请成为技师" in group_student_labels assert "我的状态" not in group_student_labels teacher_rows = module._teacher_result_markup( [ { "user_id": index, "username": f"teacher{index}", "display_name": f"技师 {index}", "online": index % 2 == 0, } for index in range(1, 5) ], radius_key="all", page=1, total=4, ).inline_keyboard assert [len(row) for row in teacher_rows[:2]] == [3, 1] assert teacher_rows[0][0].text.startswith("🔴") assert teacher_rows[0][1].text.startswith("🟢") finally: await directory_service.stop_presence_sweeper() async def test_internal_membership_rpc_requires_loopback_token(app_modules): admin_api = app_modules.load("wbb.admin.api") app_modules.wbb.INTERNAL_TOKEN = "internal-test-token" application = admin_api.build_admin_application() await application["admin_api"].initialize() client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True)) await client.start_server() user = SimpleNamespace( id=600, username="admin600", first_name="目录", last_name="管理员", is_bot=False, ) app_modules.app.members[(-100600, 600)] = SimpleNamespace( status=ChatMemberStatus.ADMINISTRATOR, user=user, ) try: denied = await client.post( "/api/internal/v1/directory/verify-local", json={ "user_id": "600", "chat_ids": ["-100600"], "require_admin": True, }, ) assert denied.status == 403 allowed = await client.post( "/api/internal/v1/directory/verify-local", headers={"Authorization": "Bearer internal-test-token"}, json={ "user_id": "600", "chat_ids": ["-100600"], "require_admin": True, }, ) assert allowed.status == 200 data = (await allowed.json())["data"] assert data["allowed"] is True assert data["chat_id"] == "-100600" finally: await client.close()