Release v1.0.56: Credential & Browser Profile Manager

tests/test_lease.py

@@ -0,0 +1,242 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for lease acquire / conflict / release / cleanup.
+"""
+import json
+import time
+import pytest
+
+
+class TestLeaseAcquire:
+    """Lease acquisition on pick-web."""
+
+    def _add_account(self, platform, login, label, tenant, env, role):
+        from service.account_service import cmd_account_add_web
+        cmd_account_add_web(
+            platform, login, "email", label, tenant, env, role,
+            None, None, None, None,
+        )
+
+    def test_pick_web_lease_returns_token(self, clean_db):
+        """pick-web --lease returns lease_token in result."""
+        self._add_account("maersk", "test@maersk.com", "Test", "t1", "production", "booking")
+
+        from service.account_service import cmd_account_pick_web
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_account_pick_web(
+                "maersk",
+                environment=None,
+                tenant_id=None,
+                role=None,
+                do_lease=True,
+                ttl_sec=900,
+                holder="test-holder",
+                purpose="rpa-run",
+            )
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue())
+        assert "lease_token" in result
+        assert len(result["lease_token"]) == 32
+        assert "lease_expires_at" in result
+        assert result["lease_token"] not in ("", "none")
+
+    def test_active_lease_blocks_pick(self, clean_db):
+        """When the only matching account is leased, pick-web finds no eligible row."""
+        from db.accounts_repo import acquire_lease, find_account_ids
+
+        self._add_account("cma_cgm", "test@cma.com", "CMA", "t1", "production", "booking")
+        ids = find_account_ids("cma_cgm")
+        assert len(ids) == 1
+        acc_id = ids[0]
+
+        acquire_lease(
+            account_id=acc_id,
+            lease_token="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
+            holder="first-holder",
+            purpose="first-task",
+            ttl_sec=900,
+        )
+
+        assert find_account_ids("cma_cgm") == []
+
+        from service.account_service import cmd_account_pick_web
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_account_pick_web(
+                "cma_cgm",
+                environment=None,
+                tenant_id=None,
+                role=None,
+                do_lease=True,
+                ttl_sec=900,
+                holder="second-holder",
+                purpose="second-task",
+            )
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue())
+        assert result["success"] is False
+        assert result["error"]["code"] == "ERROR:NO_ACCOUNT"
+
+    def test_release_then_pick(self, clean_db):
+        """After release, the same account can be picked again."""
+        from db.accounts_repo import acquire_lease, release_lease, find_account_ids
+
+        self._add_account("evergreen", "ship@evergreen.com", "EVG", "t1", "production", "booking")
+        ids = find_account_ids("evergreen")
+        acc_id = ids[0]
+
+        token = "z9y8x7w6v5u4t3s2r1q0p9o8n7m6l5k4"
+        acquire_lease(acc_id, token, "holder", "purpose", 900)
+        ok = release_lease(token)
+        assert ok, "release should succeed"
+
+        from service.account_service import cmd_account_pick_web
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_account_pick_web(
+                "evergreen",
+                environment=None,
+                tenant_id=None,
+                role=None,
+                do_lease=True,
+                ttl_sec=900,
+                holder="new-holder",
+                purpose="new-task",
+            )
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue().strip().splitlines()[-1])
+        assert result["id"] == acc_id
+        assert "lease_token" in result
+
+
+class TestLeaseRelease:
+    """lease release command."""
+
+    def test_release_unknown_token(self, clean_db):
+        """release of non-existent token returns LEASE_NOT_FOUND."""
+        from service.account_service import cmd_lease_release
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_lease_release("nonexistent_token_123456789012")
+        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."""
+
+    def test_expired_lease_allows_pick(self, clean_db):
+        """After cleanup marks expired, account can be picked again."""
+        from db.accounts_repo import acquire_lease, cleanup_expired_leases, find_account_ids
+
+        self._add_account_("cargo_wise", "cw@cargowise.com", "CW", "t1", "production", "booking")
+        ids = find_account_ids("cargo_wise")
+        acc_id = ids[0]
+
+        # Acquire lease with very short TTL (1 second)
+        acquire_lease(acc_id, "shortlived1234567890123456789012", "holder", "short", 1)
+        # Wait for it to expire
+        time.sleep(1.5)
+        count = cleanup_expired_leases()
+        assert count >= 1, "at least one lease should be cleaned"
+
+        # Now pick-web should succeed
+        from service.account_service import cmd_account_pick_web
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_account_pick_web(
+                "cargo_wise",
+                environment=None,
+                tenant_id=None,
+                role=None,
+                do_lease=True,
+                ttl_sec=900,
+                holder="after-cleanup",
+                purpose="fresh-task",
+            )
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue().strip().splitlines()[-1])
+        assert result["id"] == acc_id
+        assert "lease_token" in result
+
+    def _add_account_(self, platform, login, label, tenant, env, role):
+        from service.account_service import cmd_account_add_web
+        cmd_account_add_web(
+            platform, login, "email", label, tenant, env, role,
+            None, None, None, None,
+        )
+
+
+class TestLeaseList:
+    """lease list command."""
+
+    def test_list_no_leases(self, clean_db):
+        """lease list --active returns empty when no leases."""
+        from service.account_service import cmd_lease_list
+        import io, sys
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_lease_list(active_only=True)
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue())
+        assert result["success"] is True
+        assert result["data"]["leases"] == []
+
+    def test_list_with_active_lease(self, clean_db):
+        """lease list --active shows active lease after acquisition."""
+        from db.accounts_repo import acquire_lease, find_account_ids
+        from service.account_service import cmd_account_add_web, cmd_lease_list
+        import io, sys
+
+        cmd_account_add_web(
+            "sinotrans", "sinotest@sinotrans.com", "email", "Sino", "t1", "production", "booking",
+            None, None, None, None,
+        )
+        ids = find_account_ids("sinotrans")
+        acquire_lease(ids[0], "listtest1234567890123456789012345", "holder", "purpose", 900)
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            cmd_lease_list(active_only=True)
+        finally:
+            sys.stdout = sys.__stdout__
+
+        result = json.loads(captured.getvalue())
+        assert result["success"] is True
+        leases = result["data"]["leases"]
+        assert len(leases) == 1
+        assert leases[0]["account_id"] == ids[0]
+        assert leases[0]["status"] == "active"