399 lines
14 KiB
Python
399 lines
14 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-display-name" 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"
|
||
|
||
|
||
JIANGCHANG_META_KWARGS = dict(
|
||
display_name="匠厂后台",
|
||
url=JIANGCHANG_URL,
|
||
domain="jc2009.com",
|
||
auth_strategy="qr_code_manual",
|
||
aliases=["jiangchang", "匠厂", "匠厂后台", "jc2009"],
|
||
)
|
||
|
||
|
||
class TestAutoRegisterFromCallerMeta:
|
||
def _meta(self):
|
||
from service.account_service import PlatformCallerMeta
|
||
|
||
return PlatformCallerMeta(**JIANGCHANG_META_KWARGS)
|
||
|
||
def test_pick_web_existing_platform_unchanged(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="h1")
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["platform_key"] == "jiangchang"
|
||
assert payload.get("profile_dir")
|
||
|
||
def test_pick_web_unregistered_without_meta(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_pick_web
|
||
|
||
cmd_account_pick_web(platform_input="auto_reg_plat")
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["success"] is False
|
||
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
|
||
assert "--platform-display-name" in payload["error"]["message"]
|
||
|
||
def test_pick_web_unregistered_with_meta_auto_registers(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
|
||
from util.platforms import resolve_platform_key_dynamic
|
||
|
||
meta = PlatformCallerMeta(
|
||
display_name="自动注册测试",
|
||
url="https://example.com/admin",
|
||
domain="example.com",
|
||
auth_strategy="qr_code_manual",
|
||
aliases=["auto_reg_plat"],
|
||
)
|
||
cmd_account_pick_web(platform_input="auto_reg_plat", platform_meta=meta)
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["success"] is False
|
||
assert payload["error"]["code"] == "ERROR:NO_ACCOUNT"
|
||
|
||
result = resolve_platform_key_dynamic("auto_reg_plat", conn=clean_db)
|
||
assert result.ok
|
||
assert result.platform_key == "auto_reg_plat"
|
||
|
||
def test_add_web_unregistered_with_meta_auto_registers(self, clean_db):
|
||
from service.account_service import cmd_account_add_web, PlatformCallerMeta
|
||
from db.accounts_repo import find_accounts
|
||
|
||
meta = PlatformCallerMeta(
|
||
display_name="匠厂后台",
|
||
url=JIANGCHANG_URL,
|
||
domain="jc2009.com",
|
||
auth_strategy="qr_code_manual",
|
||
aliases=["jiangchang"],
|
||
)
|
||
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,
|
||
platform_meta=meta,
|
||
)
|
||
accs = find_accounts(platform_key="jiangchang")
|
||
assert len(accs) == 1
|
||
assert accs[0]["url"] == JIANGCHANG_URL
|
||
|
||
def test_invalid_platform_key_no_register(self, clean_db, capsys):
|
||
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
|
||
|
||
meta = PlatformCallerMeta(display_name="坏", url="https://x.com")
|
||
cmd_account_pick_web(platform_input="bad key!", platform_meta=meta)
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_KEY"
|
||
row = clean_db.execute(
|
||
"SELECT COUNT(*) FROM platforms WHERE platform_key LIKE '%bad%'"
|
||
).fetchone()[0]
|
||
assert row == 0
|
||
|
||
def test_disabled_platform_not_re_enabled(self, clean_db, capsys):
|
||
from db.accounts_repo import upsert_platform
|
||
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
|
||
|
||
upsert_platform(
|
||
clean_db,
|
||
key="jiangchang",
|
||
display_name="匠厂后台",
|
||
default_url=JIANGCHANG_URL,
|
||
)
|
||
clean_db.execute("UPDATE platforms SET enabled = 0 WHERE platform_key = 'jiangchang'")
|
||
clean_db.commit()
|
||
|
||
meta = PlatformCallerMeta(display_name="匠厂后台", url=JIANGCHANG_URL)
|
||
cmd_account_pick_web(platform_input="jiangchang", platform_meta=meta)
|
||
out = capsys.readouterr().out.strip()
|
||
payload = json.loads(out)
|
||
assert payload["error"]["code"] == "ERROR:PLATFORM_DISABLED"
|
||
|
||
enabled = clean_db.execute(
|
||
"SELECT enabled FROM platforms WHERE platform_key = 'jiangchang'"
|
||
).fetchone()[0]
|
||
assert enabled == 0
|
||
|
||
def test_auto_register_idempotent(self, clean_db):
|
||
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
|
||
|
||
meta = PlatformCallerMeta(
|
||
display_name="幂等测试",
|
||
url="https://example.com/admin",
|
||
aliases=["idem_plat"],
|
||
)
|
||
cmd_account_pick_web(platform_input="idem_plat", platform_meta=meta)
|
||
cmd_account_pick_web(platform_input="idem_plat", platform_meta=meta)
|
||
count = clean_db.execute(
|
||
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'idem_plat'"
|
||
).fetchone()[0]
|
||
assert count == 1
|