data-collector-mcp/tests/test_auth.py

from __future__ import annotations

import time
import unittest
from unittest.mock import patch

from data_collector_mcp import auth


class AuthTests(unittest.TestCase):
    def test_load_projects_config_normalizes_project(self) -> None:
        with patch(
            "data_collector_mcp.auth.read_sys_config_value",
            return_value=[
                {
                    "project_key": " dev-01 ",
                    "project_name": " DEV ",
                    "base_url": "http://gateway.test/",
                    "data_collector_base_url": "http://collector.test/",
                    "username": "admin",
                    "password": "123456",
                }
            ],
        ):
            projects = auth.load_projects_config()

        self.assertEqual(
            projects,
            [
                {
                    "project_key": "dev-01",
                    "project_name": "DEV",
                    "base_url": "http://gateway.test",
                    "data_collector_base_url": "http://collector.test",
                    "username": "admin",
                    "password": "123456",
                    "enabled": True,
                }
            ],
        )

    def test_load_projects_config_requires_data_collector_base_url(self) -> None:
        with patch(
            "data_collector_mcp.auth.read_sys_config_value",
            return_value=[
                {
                    "project_key": "dev-01",
                    "base_url": "http://gateway.test",
                    "username": "admin",
                    "password": "123456",
                }
            ],
        ):
            with self.assertRaisesRegex(ValueError, "data_collector_base_url is required"):
                auth.load_projects_config()

    def test_resolve_project_token_uses_valid_cache(self) -> None:
        project = {"project_key": "dev-01"}
        with patch(
            "data_collector_mcp.auth.read_sys_config_value",
            return_value={"auth_token": "cached-token", "expire_at": int(time.time()) + 3600},
        ), patch("data_collector_mcp.auth.request_json") as request_json:
            token = auth.resolve_project_token(project)

        self.assertEqual(token, "cached-token")
        request_json.assert_not_called()

    def test_resolve_project_token_logs_in_when_cache_expired(self) -> None:
        project = {
            "project_key": "dev-01",
            "base_url": "http://gateway.test",
            "username": "admin",
            "password": "123456",
        }
        with patch(
            "data_collector_mcp.auth.read_sys_config_value",
            return_value={"auth_token": "old-token", "expire_at": int(time.time()) - 1},
        ), patch(
            "data_collector_mcp.auth.request_json",
            return_value={
                "errcode": 0,
                "token": "new-token",
                "token_expire_time": int(time.time()) + 3600,
            },
        ) as request_json, patch("data_collector_mcp.auth.write_sys_config_value") as write_value:
            token = auth.resolve_project_token(project)

        self.assertEqual(token, "new-token")
        request_json.assert_called_once_with(
            "POST",
            "http://gateway.test/api/ai/auth/password_login",
            json_payload={"username": "admin", "password": "123456"},
        )
        write_value.assert_called_once()
        self.assertEqual(write_value.call_args.args[0], "mcp_data_collector_token_dev-01")
        self.assertEqual(write_value.call_args.args[1]["auth_token"], "new-token")


if __name__ == "__main__":
    unittest.main()