247 lines
8.2 KiB
Python
247 lines
8.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""自定义平台：动态解析、platform ensure、add-web / pick-web / browser URL。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
|
||
import pytest
|
||
|
||
|
||
JIANGCHANG_URL = "https://jc2009.com/admin/index.html"
|
||
|
||
|
||
def _ensure_jiangchang(clean_db) -> None:
|
||
from service.account_service import cmd_platform_ensure
|
||
|
||
rc = cmd_platform_ensure(
|
||
key="jiangchang",
|
||
display_name="匠厂后台",
|
||
domain="generic",
|
||
url=JIANGCHANG_URL,
|
||
auth_strategy=None,
|
||
aliases=json.dumps(["jc", "匠厂后台"], ensure_ascii=False),
|
||
capabilities=None,
|
||
)
|
||
assert rc == 0
|
||
|
||
|
||
class TestBuiltinPlatformsUnchanged:
|
||
def test_douyin_alias(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
result = resolve_platform_key_dynamic("抖音", conn=clean_db)
|
||
assert result.ok
|
||
assert result.platform_key == "douyin"
|
||
|
||
def test_toutiao_key(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
result = resolve_platform_key_dynamic("toutiao", conn=clean_db)
|
||
assert result.ok
|
||
assert result.platform_key == "toutiao"
|
||
|
||
|
||
class TestDynamicPlatformResolution:
|
||
def test_unregistered_valid_key(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
result = resolve_platform_key_dynamic("unknown_custom", conn=clean_db)
|
||
assert not result.ok
|
||
assert result.error_code == "PLATFORM_NOT_REGISTERED"
|
||
assert "platform ensure" in (result.message or "")
|
||
|
||
def test_invalid_platform_key_format(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 == "INVALID_PLATFORM_KEY"
|
||
|
||
def test_empty_platform(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
result = resolve_platform_key_dynamic("", conn=clean_db)
|
||
assert result.error_code == "INVALID_PLATFORM"
|
||
|
||
def test_registered_jiangchang(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
_ensure_jiangchang(clean_db)
|
||
result = resolve_platform_key_dynamic("jiangchang", conn=clean_db)
|
||
assert result.ok
|
||
assert result.platform_key == "jiangchang"
|
||
|
||
def test_registered_alias_jc(self, clean_db):
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
_ensure_jiangchang(clean_db)
|
||
result = resolve_platform_key_dynamic("jc", conn=clean_db)
|
||
assert result.ok
|
||
assert result.platform_key == "jiangchang"
|
||
|
||
|
||
class TestPlatformEnsure:
|
||
def test_ensure_returns_json(self, clean_db, capsys):
|
||
from service.account_service import cmd_platform_ensure
|
||
|
||
rc = cmd_platform_ensure(
|
||
key="jiangchang",
|
||
display_name="匠厂后台",
|
||
url=JIANGCHANG_URL,
|
||
)
|
||
assert rc == 0
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["ok"] is True
|
||
assert payload["platform_key"] == "jiangchang"
|
||
assert payload["action"] in ("created", "updated")
|
||
|
||
|
||
class TestCustomPlatformAccounts:
|
||
def test_add_web_after_ensure(self, clean_db):
|
||
from service.account_service import cmd_account_add_web
|
||
from db.accounts_repo import find_accounts
|
||
|
||
_ensure_jiangchang(clean_db)
|
||
cmd_account_add_web(
|
||
platform_input="jiangchang",
|
||
login_id="manual",
|
||
login_id_type="manual",
|
||
label="匠厂后台 手工登录",
|
||
tenant_id=None,
|
||
environment="production",
|
||
role="admin",
|
||
provider_code=None,
|
||
url=None,
|
||
extra_json_str=None,
|
||
profile_dir_override=None,
|
||
)
|
||
accs = find_accounts(platform_key="jiangchang")
|
||
assert len(accs) == 1
|
||
assert accs[0]["platform_key"] == "jiangchang"
|
||
assert accs[0]["url"] == JIANGCHANG_URL
|
||
assert accs[0]["profile_dir"]
|
||
|
||
def test_pick_web_after_ensure(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_add_web, cmd_account_pick_web
|
||
|
||
_ensure_jiangchang(clean_db)
|
||
cmd_account_add_web(
|
||
platform_input="jiangchang",
|
||
login_id="manual",
|
||
login_id_type="manual",
|
||
label="匠厂后台",
|
||
tenant_id=None,
|
||
environment="production",
|
||
role="admin",
|
||
provider_code=None,
|
||
url=None,
|
||
extra_json_str=None,
|
||
profile_dir_override=None,
|
||
)
|
||
capsys.readouterr()
|
||
cmd_account_pick_web(
|
||
platform_input="jiangchang",
|
||
do_lease=True,
|
||
holder="test-holder",
|
||
ttl_sec=300,
|
||
)
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["platform_key"] == "jiangchang"
|
||
assert payload["profile_dir"]
|
||
assert payload.get("lease_token")
|
||
|
||
def test_pick_web_unregistered_returns_structured_error(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_pick_web
|
||
|
||
cmd_account_pick_web(platform_input="unknown_custom")
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["success"] is False
|
||
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
|
||
assert "platform ensure" in payload["error"]["message"]
|
||
|
||
def test_add_web_unregistered_returns_structured_error(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_add_web
|
||
|
||
cmd_account_add_web(
|
||
platform_input="unknown_custom",
|
||
login_id="manual",
|
||
login_id_type="manual",
|
||
label="x",
|
||
tenant_id=None,
|
||
environment="production",
|
||
role="admin",
|
||
provider_code=None,
|
||
url=None,
|
||
extra_json_str=None,
|
||
profile_dir_override=None,
|
||
)
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["success"] is False
|
||
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
|
||
|
||
|
||
class TestBrowserServiceUrlResolution:
|
||
def test_custom_platform_default_url_from_db(self, clean_db):
|
||
from db.accounts_repo import insert_account, upsert_platform
|
||
from service.browser_service import resolve_account_browser_url
|
||
import time
|
||
|
||
now = int(time.time())
|
||
upsert_platform(
|
||
clean_db,
|
||
key="jiangchang",
|
||
display_name="匠厂后台",
|
||
default_url=JIANGCHANG_URL,
|
||
)
|
||
aid = insert_account(
|
||
conn=clean_db,
|
||
platform_key="jiangchang",
|
||
account_label="test",
|
||
login_id=None,
|
||
login_id_type="manual",
|
||
tenant_id=None,
|
||
environment="production",
|
||
role="admin",
|
||
provider_code="jiangchang",
|
||
profile_dir="/tmp/test-profile",
|
||
url="",
|
||
extra_json="{}",
|
||
now=now,
|
||
auth_strategy="qr_code_manual",
|
||
)
|
||
clean_db.commit()
|
||
account = {
|
||
"id": aid,
|
||
"platform_key": "jiangchang",
|
||
"url": "",
|
||
}
|
||
assert resolve_account_browser_url(account, conn=clean_db) == JIANGCHANG_URL
|
||
|
||
def test_account_url_overrides_platform_default(self, clean_db):
|
||
from service.browser_service import resolve_account_browser_url
|
||
|
||
account = {
|
||
"platform_key": "jiangchang",
|
||
"url": "https://custom.example.com/admin",
|
||
}
|
||
assert (
|
||
resolve_account_browser_url(account, conn=clean_db)
|
||
== "https://custom.example.com/admin"
|
||
)
|
||
|
||
|
||
class TestPlatformKeyFormatValidation:
|
||
def test_ensure_rejects_invalid_key(self, clean_db, capsys):
|
||
from service.account_service import cmd_platform_ensure
|
||
|
||
rc = cmd_platform_ensure(key="江厂 空格", display_name="bad")
|
||
assert rc == 1
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_KEY"
|