Release v2.0.25 force-release lease row action
All checks were successful
技能自动化发布 / release (push) Successful in 6s

This commit is contained in:
2026-09-12 16:18:36 +08:00
parent 3f1cde1f99
commit 3ce0cc40a1
12 changed files with 251 additions and 10 deletions

View File

@@ -144,6 +144,67 @@ class TestLeaseRelease:
assert result["success"] is False
assert result["error"]["code"] == "ERROR:LEASE_NOT_FOUND"
def test_force_release_by_id(self, clean_db):
"""force-release by lease row id frees the account for pick-web."""
from db.accounts_repo import acquire_lease, find_account_ids, release_lease_by_id
from service.account_service import cmd_account_add_web, cmd_lease_force_release
import io, sys
cmd_account_add_web(
"maersk", "force-release@maersk.com", "email", "FR", "t1", "production", "booking",
None, None, None, None,
)
ids = find_account_ids("maersk")
acc_id = ids[0]
lease = acquire_lease(
account_id=acc_id,
lease_token="f1e2d3c4b5a697887766554433221100",
holder="stuck-holder",
purpose="download",
ttl_sec=900,
)
assert find_account_ids("maersk") == []
released = release_lease_by_id(int(lease["id"]))
assert released is not None
assert released["status"] == "released"
assert acc_id in find_account_ids("maersk")
# Re-acquire and exercise CLI command output shape
lease2 = acquire_lease(
account_id=acc_id,
lease_token="a0b1c2d3e4f5061728394a5b6c7d8e9f",
holder="stuck-2",
purpose="download",
ttl_sec=900,
)
buf = io.StringIO()
old = sys.stdout
sys.stdout = buf
try:
cmd_lease_force_release(int(lease2["id"]))
finally:
sys.stdout = old
line = buf.getvalue().strip().splitlines()[-1]
result = json.loads(line)
assert result.get("success") is True
assert (result.get("data") or {}).get("status") == "released"
def test_force_release_inactive_fails(self, clean_db):
from service.account_service import cmd_lease_force_release
import io, sys
captured = io.StringIO()
sys.stdout = captured
try:
cmd_lease_force_release(999999)
finally:
sys.stdout = sys.__stdout__
result = json.loads(captured.getvalue())
assert result["success"] is False
assert result["error"]["code"] == "ERROR:LEASE_NOT_FOUND"
class TestLeaseCleanup:
"""lease cleanup / expired lease handling."""

View File

@@ -78,7 +78,22 @@ class TestRowOpenActions(unittest.TestCase):
self.assertEqual(copy.get("placements"), ["row"])
self.assertEqual(copy.get("interaction"), "direct")
self.assertFalse(copy.get("replaces"))
steps = copy.get("steps") or []
def test_lease_force_release_row_action(self) -> None:
action = _by_id(_load_actions(), "lease-force-release")
self.assertEqual(action.get("label"), "强制释放租约")
self.assertEqual(action.get("placements"), ["row"])
self.assertEqual(action.get("interaction"), "confirm")
self.assertEqual(action.get("executionProfile"), "sync")
bind = action.get("bind") or {}
self.assertEqual(bind.get("tables"), ["account_leases"])
self.assertEqual(bind.get("inputMapping"), {"leaseId": "$row.$pk"})
self.assertEqual(bind.get("visibleWhen"), {"status": "active"})
self.assertEqual(
(action.get("entrypoint") or {}).get("args"),
["force-release", "--id", "{{leaseId}}"],
)
steps = action.get("steps") or []
self.assertTrue(any(s.get("do") == "reloadTable" for s in steps))