chendelian
3f1cde1f99
All checks were successful
技能自动化发布 / release (push) Successful in 6s
Co-authored-by: Cursor <cursoragent@cursor.com>
134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""resolve_platform_key_dynamic 与 JSON 解析边界。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
|
|
class TestResolvePlatformKeyDynamicParsing:
|
|
def test_builtin_still_works_without_db_row(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("马士基", conn=clean_db)
|
|
assert result.platform_key == "maersk"
|
|
|
|
def test_disabled_platform_returns_disabled(self, clean_db):
|
|
from db.accounts_repo import upsert_platform
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
upsert_platform(
|
|
clean_db,
|
|
key="custom_off",
|
|
display_name="Off",
|
|
default_url="",
|
|
)
|
|
clean_db.execute(
|
|
"UPDATE platforms SET enabled = 0 WHERE platform_key = ?",
|
|
("custom_off",),
|
|
)
|
|
clean_db.commit()
|
|
result = resolve_platform_key_dynamic("custom_off", conn=clean_db)
|
|
assert result.error_code == "PLATFORM_DISABLED"
|
|
|
|
def test_get_platform_spec_from_db(self, clean_db):
|
|
from db.accounts_repo import upsert_platform
|
|
from util.platforms import get_platform_spec
|
|
|
|
upsert_platform(
|
|
clean_db,
|
|
key="jiangchang",
|
|
display_name="匠厂后台",
|
|
default_url="https://jc2009.com/admin/index.html",
|
|
)
|
|
spec = get_platform_spec("jiangchang", conn=clean_db)
|
|
assert spec["display_name"] == "匠厂后台"
|
|
assert spec["default_url"] == "https://jc2009.com/admin/index.html"
|
|
|
|
def test_malformed_aliases_json_does_not_crash(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
import time
|
|
|
|
now = int(time.time())
|
|
clean_db.execute(
|
|
"""
|
|
INSERT INTO platforms (
|
|
platform_key, display_name, domain, provider_code,
|
|
default_url, aliases_json, capabilities_json,
|
|
enabled, default_auth_strategy, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, 1, NULL, ?, ?)
|
|
""",
|
|
(
|
|
"bad_alias",
|
|
"Bad",
|
|
"generic",
|
|
"bad_alias",
|
|
"",
|
|
"not-json",
|
|
"{}",
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
clean_db.commit()
|
|
result = resolve_platform_key_dynamic("bad_alias", conn=clean_db)
|
|
assert result.platform_key == "bad_alias"
|
|
|
|
|
|
class TestPlatformKeyFormatAndAmbiguity:
|
|
def test_format_rejects_mixed_case(self):
|
|
from util.platforms import is_valid_platform_key_format
|
|
|
|
assert is_valid_platform_key_format("haokan") is True
|
|
assert is_valid_platform_key_format("HaoKan") is False
|
|
assert is_valid_platform_key_format("hao_kan") is True
|
|
|
|
def test_deepseek_display_resolves_to_key(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("DeepSeek", conn=clean_db)
|
|
assert result.ok
|
|
assert result.platform_key == "deepseek"
|
|
|
|
def test_douyin_alias_still_unique(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("抖音", conn=clean_db)
|
|
assert result.platform_key == "douyin"
|
|
|
|
def test_wechat_key_wins(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("wechat", conn=clean_db)
|
|
assert result.ok
|
|
assert result.platform_key == "wechat"
|
|
|
|
def test_gongzhonghao_is_ambiguous(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("公众号", conn=clean_db)
|
|
assert not result.ok
|
|
assert result.error_code == "AMBIGUOUS_PLATFORM"
|
|
assert "wechat" in (result.message or "")
|
|
assert "wechat_mp" in (result.message or "")
|
|
|
|
def test_unregistered_lookalike_key(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("HaoKan", conn=clean_db)
|
|
assert result.error_code == "INVALID_PLATFORM_KEY"
|
|
assert "platform ensure" in (result.message or "")
|
|
|
|
def test_unregistered_valid_key_not_inserted(self, clean_db):
|
|
from util.platforms import resolve_platform_key_dynamic
|
|
|
|
result = resolve_platform_key_dynamic("hao_kan", conn=clean_db)
|
|
assert result.error_code == "PLATFORM_NOT_REGISTERED"
|
|
assert "已拒绝写入" in (result.message or "")
|
|
count = clean_db.execute(
|
|
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'hao_kan'"
|
|
).fetchone()[0]
|
|
assert count == 0
|
|
|