120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Human-friendly account resolve / pick-web --login-id / ensure by login_id."""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
|
|
|
|
def _capture(fn, *args, **kwargs):
|
|
buf = io.StringIO()
|
|
old = sys.stdout
|
|
sys.stdout = buf
|
|
try:
|
|
fn(*args, **kwargs)
|
|
finally:
|
|
sys.stdout = old
|
|
return buf.getvalue()
|
|
|
|
|
|
def _add_web(platform, login_id, label):
|
|
from service.account_service import cmd_account_add_web
|
|
|
|
payload = json.loads(
|
|
_capture(
|
|
cmd_account_add_web,
|
|
platform,
|
|
login_id,
|
|
"username",
|
|
label,
|
|
None,
|
|
"production",
|
|
"default",
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
)
|
|
)
|
|
if isinstance(payload, dict) and isinstance(payload.get("data"), dict):
|
|
return payload["data"]
|
|
return payload
|
|
|
|
|
|
class TestAccountResolve:
|
|
def test_resolve_by_login_id(self, clean_db):
|
|
from service.account_service import cmd_account_resolve
|
|
|
|
created = _add_web("maersk", "chendelian", "陈德联")
|
|
raw = _capture(
|
|
cmd_account_resolve,
|
|
platform_input="maersk",
|
|
login_id="ChenDelian",
|
|
)
|
|
acc = json.loads(raw)
|
|
assert acc.get("success") is not False
|
|
assert acc["id"] == created["id"]
|
|
assert acc["login_id"] == "chendelian"
|
|
|
|
def test_resolve_by_query_fallback_label(self, clean_db):
|
|
from service.account_service import cmd_account_resolve
|
|
|
|
created = _add_web("maersk", "u1", "我的马士基")
|
|
raw = _capture(
|
|
cmd_account_resolve,
|
|
platform_input="maersk",
|
|
query="我的马士基",
|
|
)
|
|
acc = json.loads(raw)
|
|
assert acc["id"] == created["id"]
|
|
|
|
def test_resolve_ambiguous(self, clean_db):
|
|
from service.account_service import cmd_account_resolve
|
|
|
|
_add_web("maersk", "same", "A")
|
|
_add_web("cma_cgm", "same", "B")
|
|
raw = _capture(cmd_account_resolve, login_id="same")
|
|
err = json.loads(raw)
|
|
assert err["success"] is False
|
|
assert err["error"]["code"] == "ERROR:AMBIGUOUS_ACCOUNT"
|
|
|
|
|
|
class TestPickByLoginId:
|
|
def test_pick_web_filters_login_id(self, clean_db):
|
|
from service.account_service import cmd_account_pick_web
|
|
|
|
a = _add_web("maersk", "alice", "A")
|
|
_add_web("maersk", "bob", "B")
|
|
raw = _capture(
|
|
cmd_account_pick_web,
|
|
"maersk",
|
|
login_id="bob",
|
|
)
|
|
picked = json.loads(raw)
|
|
assert picked["login_id"] == "bob"
|
|
assert picked["id"] != a["id"]
|
|
|
|
|
|
class TestEnsureByLoginId:
|
|
def test_ensure_creates_named_even_if_platform_has_others(self, clean_db):
|
|
from db.accounts_repo import find_accounts
|
|
from service.account_service import cmd_account_ensure_web
|
|
|
|
_add_web("maersk", "existing", "已有")
|
|
before = find_accounts(platform_key="maersk", status="active", limit=50)
|
|
assert len(before) == 1
|
|
|
|
raw = _capture(
|
|
cmd_account_ensure_web,
|
|
"maersk",
|
|
login_id="chendelian",
|
|
label="陈德联",
|
|
)
|
|
result = json.loads(raw)
|
|
assert result.get("account_ensured") is True
|
|
assert result["login_id"] == "chendelian"
|
|
|
|
after = find_accounts(platform_key="maersk", status="active", limit=50)
|
|
assert len(after) == 2
|