TelegramBotAdmin/tests/test_directory.py

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_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.keyboard for button in row
        ]
        assert "Telegram 定位" in student_labels
        assert "手动选择位置" in student_labels
        assert "附近技师" 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.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 "我的状态" 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()