abyssVpn/services/api/accounts/tests.py

import json
from unittest.mock import patch

from django.test import TestCase, override_settings

from accounts.models import User
from accounts.services.feishu_oauth import upsert_user_from_feishu
from vpn.models import Device, VpnLog


class FeishuUserInfoTests(TestCase):
    def test_upsert_user_prefers_chinese_display_name(self):
        user = upsert_user_from_feishu(
            {
                "open_id": "ou_name",
                "email": "chenhuan@example.com",
                "name": {"zh_cn": "陈欢", "en_us": "Chen Huan"},
                "en_name": "Chen Huan",
            }
        )

        self.assertEqual(user.name, "陈欢")
        self.assertEqual(user.email, "chenhuan@example.com")

    def test_upsert_user_reads_nested_user_payload(self):
        user = upsert_user_from_feishu(
            {
                "user": {
                    "open_id": "ou_nested",
                    "display_name": "陈欢",
                    "email": "chenhuan@example.com",
                }
            }
        )

        self.assertEqual(user.feishu_open_id, "ou_nested")
        self.assertEqual(user.name, "陈欢")


@override_settings(FEISHU_EVENT_VERIFICATION_TOKEN="verify-token")
class FeishuEventCallbackTests(TestCase):
    def setUp(self):
        self.user = User.objects.create(feishu_open_id="ou_123", email="u@example.com", name="User")
        self.device = Device.objects.create(
            user=self.user,
            device_name="pc-01",
            device_fingerprint="fingerprint-01",
            netmaker_node_id="host-01",
        )

    def test_url_verification_returns_challenge(self):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps({"token": "verify-token", "challenge": "challenge-value"}),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["challenge"], "challenge-value")

    @override_settings(FEISHU_EVENT_VERIFICATION_TOKEN="", DEBUG=False)
    def test_url_verification_can_pass_before_token_is_configured(self):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps({"challenge": "challenge-value"}),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["challenge"], "challenge-value")

    @override_settings(FEISHU_EVENT_VERIFICATION_TOKEN="", DEBUG=False)
    def test_real_events_require_token_in_production(self):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps(
                {
                    "schema": "2.0",
                    "header": {"event_type": "contact.user.deleted_v3"},
                    "event": {"object": {"open_id": "ou_123"}},
                }
            ),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 403)

    @patch("vpn.services.offboarding.NetmakerClient")
    def test_deleted_user_event_disables_user_and_revokes_vpn(self, client_cls):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps(
                {
                    "schema": "2.0",
                    "header": {
                        "event_type": "contact.user.deleted_v3",
                        "event_id": "event-01",
                        "token": "verify-token",
                    },
                    "event": {"object": {"open_id": "ou_123"}},
                }
            ),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 200)
        self.user.refresh_from_db()
        self.device.refresh_from_db()
        self.assertEqual(self.user.status, User.STATUS_DISABLED)
        self.assertEqual(self.device.status, Device.STATUS_REVOKED)
        client_cls.return_value.delete_node.assert_called_once_with("host-01")
        self.assertTrue(VpnLog.objects.filter(user=self.user, action="user_disabled").exists())

    @patch("vpn.services.offboarding.NetmakerClient")
    def test_frozen_user_update_disables_user_and_revokes_vpn(self, client_cls):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps(
                {
                    "schema": "2.0",
                    "header": {
                        "event_type": "contact.user.updated_v3",
                        "event_id": "event-02",
                        "token": "verify-token",
                    },
                    "event": {
                        "object": {
                            "open_id": "ou_123",
                            "status": {"is_frozen": True},
                        }
                    },
                }
            ),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 200)
        self.user.refresh_from_db()
        self.device.refresh_from_db()
        self.assertEqual(self.user.status, User.STATUS_DISABLED)
        self.assertEqual(self.device.status, Device.STATUS_REVOKED)
        client_cls.return_value.delete_node.assert_called_once_with("host-01")

    def test_invalid_event_token_is_rejected(self):
        response = self.client.post(
            "/auth/feishu/event-callback",
            data=json.dumps({"token": "wrong", "challenge": "challenge-value"}),
            content_type="application/json",
        )

        self.assertEqual(response.status_code, 403)