Initial import of cf-temp-email deploy CLI
This commit is contained in:
295
tests/test_cloudflare.py
Normal file
295
tests/test_cloudflare.py
Normal file
@@ -0,0 +1,295 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import respx
|
||||
|
||||
from cf_temp_email_deploy.cloudflare import CloudflareClient
|
||||
from cf_temp_email_deploy.errors import CloudflareAPIError, ConfigError
|
||||
from cf_temp_email_deploy.models import CloudflareConfig
|
||||
|
||||
|
||||
def build_config() -> CloudflareConfig:
|
||||
return CloudflareConfig(
|
||||
account_name="demo-account",
|
||||
zone_name="example.com",
|
||||
api_token="token-value",
|
||||
api_email="demo@example.com",
|
||||
global_api_key="global-key",
|
||||
)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_verify_token_uses_bearer_token() -> None:
|
||||
route = respx.get("https://api.cloudflare.com/client/v4/user/tokens/verify").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={"success": True, "result": {"status": "active"}},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.verify_token()
|
||||
|
||||
assert route.called is True
|
||||
assert route.calls[0].request.headers["Authorization"] == "Bearer token-value"
|
||||
assert result["status"] == "active"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_resolve_account_uses_global_key_headers() -> None:
|
||||
route = respx.get("https://api.cloudflare.com/client/v4/accounts").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": [{"id": "acc-1", "name": "demo-account"}],
|
||||
"result_info": {"page": 1, "per_page": 50, "count": 1, "total_pages": 1},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.resolve_account(account_name="demo-account")
|
||||
|
||||
request = route.calls[0].request
|
||||
assert request.headers["X-Auth-Email"] == "demo@example.com"
|
||||
assert request.headers["X-Auth-Key"] == "global-key"
|
||||
assert result["id"] == "acc-1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_resolve_zone_uses_account_filter() -> None:
|
||||
route = respx.get("https://api.cloudflare.com/client/v4/zones").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": [{"id": "zone-1", "name": "example.com"}],
|
||||
"result_info": {"page": 1, "per_page": 50, "count": 1, "total_pages": 1},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.resolve_zone(zone_name="example.com", account_id="acc-1")
|
||||
|
||||
request = route.calls[0].request
|
||||
assert "account.id=acc-1" in str(request.url)
|
||||
assert request.headers["Authorization"] == "Bearer token-value"
|
||||
assert result["id"] == "zone-1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_email_routing_request_falls_back_to_global_key() -> None:
|
||||
route = respx.get("https://api.cloudflare.com/client/v4/zones/zone-1/email/routing/rules/catch_all").mock(
|
||||
side_effect=[
|
||||
httpx.Response(
|
||||
403,
|
||||
json={
|
||||
"success": False,
|
||||
"errors": [{"message": "Use X-Auth-Email and X-Auth-Key for this endpoint."}],
|
||||
},
|
||||
),
|
||||
httpx.Response(
|
||||
200,
|
||||
json={"success": True, "result": {"tag": "catch_all"}},
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.request_email_routing("GET", "/zones/zone-1/email/routing/rules/catch_all")
|
||||
|
||||
assert len(route.calls) == 2
|
||||
assert route.calls[0].request.headers["Authorization"] == "Bearer token-value"
|
||||
assert route.calls[1].request.headers["X-Auth-Key"] == "global-key"
|
||||
assert result["result"]["tag"] == "catch_all"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_request_raises_cloudflare_error_for_unsuccessful_payload() -> None:
|
||||
respx.get("https://api.cloudflare.com/client/v4/user/tokens/verify").mock(
|
||||
return_value=httpx.Response(
|
||||
403,
|
||||
json={"success": False, "errors": [{"message": "forbidden"}]},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
try:
|
||||
client.verify_token()
|
||||
except CloudflareAPIError as exc:
|
||||
assert "forbidden" in str(exc)
|
||||
else: # pragma: no cover
|
||||
raise AssertionError("expected CloudflareAPIError")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_ensure_pages_project_creates_when_missing() -> None:
|
||||
get_route = respx.get(
|
||||
"https://api.cloudflare.com/client/v4/accounts/acc-1/pages/projects/demo-pages"
|
||||
).mock(return_value=httpx.Response(404, json={"success": False, "errors": [{"message": "not found"}]}))
|
||||
post_route = respx.post(
|
||||
"https://api.cloudflare.com/client/v4/accounts/acc-1/pages/projects"
|
||||
).mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": {"id": "proj-1", "name": "demo-pages", "subdomain": "demo-pages.pages.dev"},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.ensure_pages_project(
|
||||
account_id="acc-1",
|
||||
project_name="demo-pages",
|
||||
production_branch="production",
|
||||
)
|
||||
|
||||
assert get_route.called is True
|
||||
assert post_route.called is True
|
||||
assert result["subdomain"] == "demo-pages.pages.dev"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_ensure_cname_record_updates_existing_record() -> None:
|
||||
list_route = respx.get("https://api.cloudflare.com/client/v4/zones/zone-1/dns_records").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": [
|
||||
{
|
||||
"id": "dns-1",
|
||||
"type": "CNAME",
|
||||
"name": "email.example.com",
|
||||
"content": "old.pages.dev",
|
||||
"proxied": True,
|
||||
"ttl": 1,
|
||||
}
|
||||
],
|
||||
"result_info": {"page": 1, "per_page": 50, "count": 1, "total_pages": 1},
|
||||
},
|
||||
)
|
||||
)
|
||||
update_route = respx.put(
|
||||
"https://api.cloudflare.com/client/v4/zones/zone-1/dns_records/dns-1"
|
||||
).mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": {
|
||||
"id": "dns-1",
|
||||
"type": "CNAME",
|
||||
"name": "email.example.com",
|
||||
"content": "new.pages.dev",
|
||||
"proxied": True,
|
||||
"ttl": 1,
|
||||
},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.ensure_cname_record(
|
||||
zone_id="zone-1",
|
||||
name="email.example.com",
|
||||
content="new.pages.dev",
|
||||
)
|
||||
|
||||
assert list_route.called is True
|
||||
assert update_route.called is True
|
||||
assert result["content"] == "new.pages.dev"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_ensure_catch_all_worker_skips_when_already_targeted() -> None:
|
||||
get_route = respx.get(
|
||||
"https://api.cloudflare.com/client/v4/zones/zone-1/email/routing/rules/catch_all"
|
||||
).mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"success": True,
|
||||
"result": {
|
||||
"id": "rule-1",
|
||||
"enabled": True,
|
||||
"actions": [{"type": "worker", "value": ["email-api"]}],
|
||||
},
|
||||
},
|
||||
)
|
||||
)
|
||||
put_route = respx.put(
|
||||
"https://api.cloudflare.com/client/v4/zones/zone-1/email/routing/rules/catch_all"
|
||||
).mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={"success": True, "result": {"id": "rule-1", "enabled": True}},
|
||||
)
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.ensure_catch_all_worker(zone_id="zone-1", script_name="email-api")
|
||||
|
||||
assert get_route.called is True
|
||||
assert put_route.called is False
|
||||
assert result["id"] == "rule-1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_wait_for_pages_domain_active_polls_until_ready() -> None:
|
||||
route = respx.get(
|
||||
"https://api.cloudflare.com/client/v4/accounts/acc-1/pages/projects/demo-pages/domains/email.example.com"
|
||||
).mock(
|
||||
side_effect=[
|
||||
httpx.Response(
|
||||
200,
|
||||
json={"success": True, "result": {"name": "email.example.com", "status": "pending"}},
|
||||
),
|
||||
httpx.Response(
|
||||
200,
|
||||
json={"success": True, "result": {"name": "email.example.com", "status": "active"}},
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with CloudflareClient(build_config()) as client:
|
||||
result = client.wait_for_pages_domain_active(
|
||||
account_id="acc-1",
|
||||
project_name="demo-pages",
|
||||
domain_name="email.example.com",
|
||||
timeout_seconds=1.0,
|
||||
poll_interval_seconds=0.0,
|
||||
)
|
||||
|
||||
assert len(route.calls) == 2
|
||||
assert result["status"] == "active"
|
||||
|
||||
|
||||
def test_catch_all_points_to_worker_accepts_nested_value_shapes() -> None:
|
||||
rule = {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
{
|
||||
"type": "worker",
|
||||
"value": [
|
||||
{
|
||||
"service": {
|
||||
"name": "email-api",
|
||||
}
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
assert CloudflareClient.catch_all_points_to_worker(rule, "email-api") is True
|
||||
|
||||
|
||||
def test_is_authentication_error_accepts_global_key_config_errors() -> None:
|
||||
error = ConfigError("旧式鉴权需要同时提供 api_email 与 global_api_key。")
|
||||
|
||||
assert CloudflareClient.is_authentication_error(error) is True
|
||||
Reference in New Issue
Block a user