144 lines
4.2 KiB
Python
144 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for account ensure-web (pick-or-create)."""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
|
|
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()
|
|
|
|
|
|
class TestEnsureWeb:
|
|
def test_ensure_creates_when_empty(self, clean_db):
|
|
from service.account_service import cmd_account_ensure_web
|
|
|
|
raw = _capture(
|
|
cmd_account_ensure_web,
|
|
"maersk",
|
|
label="马士基默认",
|
|
)
|
|
result = json.loads(raw)
|
|
assert result.get("success") is not False
|
|
assert result["platform_key"] == "maersk"
|
|
assert result.get("profile_dir")
|
|
assert result.get("account_ensured") is True
|
|
assert result.get("id")
|
|
|
|
def test_ensure_reuses_existing_no_duplicate(self, clean_db):
|
|
from service.account_service import cmd_account_add_web, cmd_account_ensure_web
|
|
from db.accounts_repo import find_accounts
|
|
|
|
_capture(
|
|
cmd_account_add_web,
|
|
"maersk",
|
|
"a@test.com",
|
|
"email",
|
|
"已有",
|
|
None,
|
|
"production",
|
|
"default",
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
)
|
|
before = find_accounts(platform_key="maersk", status="active", limit=50)
|
|
assert len(before) == 1
|
|
|
|
raw = _capture(cmd_account_ensure_web, "maersk")
|
|
result = json.loads(raw)
|
|
assert result.get("account_ensured") is not True
|
|
assert result["id"] == before[0]["id"]
|
|
|
|
after = find_accounts(platform_key="maersk", status="active", limit=50)
|
|
assert len(after) == 1
|
|
|
|
def test_ensure_lease_busy_does_not_create(self, clean_db):
|
|
from db.accounts_repo import acquire_lease, find_account_ids, find_accounts
|
|
from service.account_service import cmd_account_add_web, cmd_account_ensure_web
|
|
|
|
_capture(
|
|
cmd_account_add_web,
|
|
"cma_cgm",
|
|
"b@cma.com",
|
|
"email",
|
|
"CMA",
|
|
None,
|
|
"production",
|
|
"default",
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
)
|
|
ids = find_account_ids("cma_cgm")
|
|
assert len(ids) == 1
|
|
acquire_lease(
|
|
account_id=ids[0],
|
|
lease_token="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
|
|
holder="other",
|
|
purpose="busy",
|
|
ttl_sec=900,
|
|
)
|
|
assert find_account_ids("cma_cgm") == []
|
|
|
|
raw = _capture(cmd_account_ensure_web, "cma_cgm", do_lease=True, holder="me")
|
|
err = json.loads(raw)
|
|
assert err["success"] is False
|
|
assert err["error"]["code"] == "ERROR:LEASE_CONFLICT"
|
|
|
|
still = find_accounts(platform_key="cma_cgm", status="active", limit=50)
|
|
assert len(still) == 1
|
|
|
|
def test_pick_web_ensure_flag_cli(self, clean_db, fake_env):
|
|
"""CLI account pick-web --ensure creates when empty."""
|
|
import os
|
|
import subprocess
|
|
|
|
scripts = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts"
|
|
)
|
|
env = {
|
|
**os.environ,
|
|
"PYTHONPATH": scripts,
|
|
"JIANGCHANG_DATA_ROOT": fake_env["data_root"],
|
|
"JIANGCHANG_USER_ID": fake_env["user_id"],
|
|
}
|
|
r = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"cli.app",
|
|
"account",
|
|
"pick-web",
|
|
"--platform",
|
|
"maersk",
|
|
"--ensure",
|
|
"--label",
|
|
"cli-default",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
cwd=scripts,
|
|
)
|
|
assert r.returncode == 0, r.stderr + r.stdout
|
|
# last non-empty JSON line
|
|
lines = [ln for ln in r.stdout.splitlines() if ln.strip()]
|
|
result = json.loads(lines[-1])
|
|
assert result.get("platform_key") == "maersk"
|
|
assert result.get("account_ensured") is True
|
|
assert result.get("profile_dir")
|