okooooooo/outlookEmailPlus
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: expand external pool coverage for guards and access control

Browse Source
This commit is contained in:
胡绍康 committed 2026-03-16 21:19:41 +08:00
1 parent aa01758ad9
commit 8b436a12f5
1 file changed
+174
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_external_pool.py
+174
View File
@@ -198,6 +198,46 @@ class ExternalPoolApiTests(unittest.TestCase):
self.assertFalse(data.get("success"))
self.assertEqual(data.get("code"), "CALLER_MISMATCH")
def test_external_pool_claim_complete_success(self):
client = self.app.test_client()
self._set_external_api_key("abc123")
with self.app.app_context():
from outlook_web.repositories import settings as settings_repo
settings_repo.set_setting("pool_external_enabled", "true")
self._insert_pool_account(provider="outlook")
claim_resp = client.post(
"/api/external/pool/claim-random",
headers=self._auth_headers(),
json={
"caller_id": "ext-worker-01",
"task_id": "task-ext-complete",
"provider": "outlook",
},
)
self.assertEqual(claim_resp.status_code, 200)
claim_data = claim_resp.get_json()["data"]
complete_resp = client.post(
"/api/external/pool/claim-complete",
headers=self._auth_headers(),
json={
"account_id": claim_data["account_id"],
"claim_token": claim_data["claim_token"],
"caller_id": "ext-worker-01",
"task_id": "task-ext-complete",
"result": "success",
"detail": "done",
},
)
self.assertEqual(complete_resp.status_code, 200)
data = complete_resp.get_json()
self.assertTrue(data.get("success"))
self.assertEqual(data.get("code"), "OK")
self.assertEqual(data.get("data", {}).get("pool_status"), "used")
def test_external_pool_stats_success(self):
client = self.app.test_client()
self._set_external_api_key("abc123")
@@ -267,6 +307,87 @@ class ExternalPoolApiTests(unittest.TestCase):
self.assertEqual(data.get("code"), "FEATURE_DISABLED")
self.assertEqual(data.get("data", {}).get("feature"), "pool_claim_random")
def test_external_pool_claim_release_disabled_in_public_mode(self):
client = self.app.test_client()
self._set_external_api_key("abc123")
with self.app.app_context():
from outlook_web.repositories import settings as settings_repo
settings_repo.set_setting("pool_external_enabled", "true")
self._insert_pool_account(provider="outlook")
claim_resp = client.post(
"/api/external/pool/claim-random",
headers=self._auth_headers(),
json={
"caller_id": "ext-worker-01",
"task_id": "task-ext-rel",
"provider": "outlook",
},
)
self.assertEqual(claim_resp.status_code, 200)
claim_data = claim_resp.get_json()["data"]
self._set_public_mode(True)
self._set_ip_whitelist(["127.0.0.1"])
self._set_disable_feature("external_api_disable_pool_claim_release", True)
resp = client.post(
"/api/external/pool/claim-release",
headers=self._auth_headers(),
json={
"account_id": claim_data["account_id"],
"claim_token": claim_data["claim_token"],
"caller_id": "ext-worker-01",
"task_id": "task-ext-rel",
},
)
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertEqual(data.get("code"), "FEATURE_DISABLED")
self.assertEqual(data.get("data", {}).get("feature"), "pool_claim_release")
def test_external_pool_claim_complete_disabled_in_public_mode(self):
client = self.app.test_client()
self._set_external_api_key("abc123")
with self.app.app_context():
from outlook_web.repositories import settings as settings_repo
settings_repo.set_setting("pool_external_enabled", "true")
self._insert_pool_account(provider="outlook")
claim_resp = client.post(
"/api/external/pool/claim-random",
headers=self._auth_headers(),
json={
"caller_id": "ext-worker-01",
"task_id": "task-ext-comp",
"provider": "outlook",
},
)
self.assertEqual(claim_resp.status_code, 200)
claim_data = claim_resp.get_json()["data"]
self._set_public_mode(True)
self._set_ip_whitelist(["127.0.0.1"])
self._set_disable_feature("external_api_disable_pool_claim_complete", True)
resp = client.post(
"/api/external/pool/claim-complete",
headers=self._auth_headers(),
json={
"account_id": claim_data["account_id"],
"claim_token": claim_data["claim_token"],
"caller_id": "ext-worker-01",
"task_id": "task-ext-comp",
"result": "success",
},
)
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertEqual(data.get("code"), "FEATURE_DISABLED")
self.assertEqual(data.get("data", {}).get("feature"), "pool_claim_complete")
def test_external_pool_requires_pool_access_for_multi_key(self):
client = self.app.test_client()
self._create_external_api_key("partner-a", "multi-pool-deny", pool_access=False)
@@ -285,6 +406,59 @@ class ExternalPoolApiTests(unittest.TestCase):
data = resp.get_json()
self.assertEqual(data.get("code"), "FORBIDDEN")
def test_external_pool_claim_random_requires_pool_access_for_multi_key(self):
client = self.app.test_client()
self._create_external_api_key(
"partner-a", "multi-pool-random-deny", pool_access=False
)
with self.app.app_context():
from outlook_web.repositories import settings as settings_repo
settings_repo.set_setting("pool_external_enabled", "true")
self._insert_pool_account(provider="outlook")
resp = client.post(
"/api/external/pool/claim-random",
headers=self._auth_headers("multi-pool-random-deny"),
json={
"caller_id": "ext-worker-01",
"task_id": "task-ext-no-access",
"provider": "outlook",
},
)
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertEqual(data.get("code"), "FORBIDDEN")
def test_external_pool_returns_api_key_not_configured(self):
client = self.app.test_client()
resp = client.get(
"/api/external/pool/stats",
headers=self._auth_headers("unconfigured-key"),
)
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertEqual(data.get("code"), "API_KEY_NOT_CONFIGURED")
def test_external_pool_ip_not_allowed_in_public_mode(self):
client = self.app.test_client()
self._set_external_api_key("abc123")
with self.app.app_context():
from outlook_web.repositories import settings as settings_repo
settings_repo.set_setting("pool_external_enabled", "true")
self._set_public_mode(True)
self._set_ip_whitelist(["10.0.0.1"])
resp = client.get("/api/external/pool/stats", headers=self._auth_headers())
self.assertEqual(resp.status_code, 403)
data = resp.get_json()
self.assertEqual(data.get("code"), "IP_NOT_ALLOWED")
def test_external_pool_allows_pool_access_for_multi_key(self):
client = self.app.test_client()
self._create_external_api_key("partner-a", "multi-pool-allow", pool_access=True)