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="王老师",
        geocode_consent=False,
    )
    await directory_db.submit_teacher_application(
        user_id=user_id,
        source="test",
        bot_id="primary",
    )
    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()

    await _approved_teacher(directory_db)
    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_consent_geocode_queue_and_cache(app_modules):
    directory_db = app_modules.load("wbb.utils.dbdirectory")
    geocoding = app_modules.load("wbb.services.directory_geocoding")
    await directory_db.upsert_directory_identity(
        user_id=300,
        username="student300",
        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["geocode_consent"] is None
    assert await directory_db.claim_geocode_job() is None

    await directory_db.set_geocode_consent(
        user_id=300,
        consent=True,
        source="test",
        actor_id=300,
        actor_name="李同学",
    )
    job = await directory_db.claim_geocode_job()
    assert job["longitude"] == pytest.approx(121.4737)
    region = geocoding.parse_nominatim_region(
        {
            "address": {
                "state": "上海市",
                "city": "上海市",
                "city_district": "黄浦区",
                "road": "测试道路",
            }
        }
    )
    assert region["label"] == "上海市 黄浦区"
    assert "road" not in region
    await directory_db.complete_geocode_job(job=job, region=region)

    completed = await directory_db.get_directory_location(300)
    assert completed["geocode_status"] == "completed"
    assert completed["region"]["district"] == "黄浦区"
    await directory_db.enqueue_geocode_job(300)
    assert await directory_db.claim_geocode_job() is None

    await directory_db.set_geocode_consent(
        user_id=300,
        consent=False,
        source="test",
        actor_id=300,
        actor_name="李同学",
    )
    disabled = await directory_db.get_directory_location(300)
    assert disabled["region"] is None
    assert disabled["geocode_status"] == "disabled"


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)

        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

        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 "更新位置" in student_labels
        assert "附近老师" in student_labels
        assert "申请成为老师" 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
    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()