Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ features:
- Create application
- Redeploy application
- Create TCP redirection
- Add a domain to an application
- List domains
- Get primary domain
- Custom CA bundle and mTLS client certificate support
Expand Down
42 changes: 42 additions & 0 deletions src/clever_cloud/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -631,6 +631,48 @@ async def create_tcp_redirection(
)
return TcpRedirection.from_api_response(data)

async def create_domain(
self,
owner_id: str,
app_id: str,
*,
domain: str,
) -> Domain:
"""Attach a domain (vhost) to an application.

Args:
owner_id: Organisation or user ID that owns the application
app_id: Application ID to attach the domain to
domain: Fully-qualified domain name, e.g. ``"app.example.com"``.
The platform also accepts a wildcard (``"*.example.com"``) and
a path suffix (``"example.com/api"``). A trailing slash is
ignored, so the value round-trips with :attr:`Domain.domain`.

Returns:
The domain now attached to the application. This endpoint answers
with an empty body on some deployments; the returned value is then
built from the requested name.

Raises:
ValueError: If ``domain`` is empty.
NotFoundError: If the organisation or the application does not
exist.
HttpError: If the domain is invalid, or is already attached to
another application.
"""
owner = encode_path_segment(owner_id, name="owner_id")
app = encode_path_segment(app_id, name="app_id")
# Stripped before encoding: a trailing slash would otherwise survive as
# %2F and reach the API as part of the name.
fqdn = domain.rstrip("/") if isinstance(domain, str) else domain
vhost = encode_path_segment(fqdn, name="domain")
data = await self._request(
"PUT", f"/v2/organisations/{owner}/applications/{app}/vhosts/{vhost}"
)
if data is None:
return Domain(domain=fqdn, is_primary=False)
return Domain.from_api_response(data)

async def list_domains(self, owner_id: str, app_id: str) -> list[Domain]:
"""List all domains (vhosts) for an application.

Expand Down
6 changes: 5 additions & 1 deletion src/clever_cloud/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,8 +176,12 @@ class Domain:
@classmethod
def from_api_response(cls, data: Any, *, is_primary: bool = False) -> Self:
data = _mapping(data, model="Domain")
fqdn = _require_str(data, "fqdn", model="Domain").rstrip("/")
if not fqdn:
msg = "Domain: missing or invalid required field 'fqdn'"
raise InvalidResponseError(msg)
return cls(
domain=_require_str(data, "fqdn", model="Domain").rstrip("/"),
domain=fqdn,
is_primary=is_primary,
)

Expand Down
47 changes: 47 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -533,6 +533,53 @@ def test_no_deprecation_warning_when_creating_the_transport(


class TestEndpoints:
async def test_create_domain(
self,
make_client: Callable[..., CleverCloudClient],
record_requests: tuple[list[httpx.Request], Callable[..., object]],
) -> None:
seen, factory = record_requests
client = make_client(factory(200, {"fqdn": "app.example.test"}))
async with client:
domain = await client.create_domain(
"orga_1", "app_1", domain="app.example.test"
)
assert domain.domain == "app.example.test"
assert domain.is_primary is False
assert seen[0].method == "PUT"
assert seen[0].url.path.endswith("/vhosts/app.example.test")

async def test_create_domain_accepts_an_empty_body(
self, make_client: Callable[..., CleverCloudClient]
) -> None:
"""Some deployments answer the PUT with no content."""
client = make_client(lambda r: httpx.Response(200))
async with client:
domain = await client.create_domain(
"orga_1", "app_1", domain="app.example.test/"
)
assert domain.domain == "app.example.test"

async def test_create_domain_encodes_a_path_suffix(
self,
make_client: Callable[..., CleverCloudClient],
record_requests: tuple[list[httpx.Request], Callable[..., object]],
) -> None:
"""A path suffix belongs to the vhost name, not to the request route."""
seen, factory = record_requests
client = make_client(factory(200))
async with client:
await client.create_domain("orga_1", "app_1", domain="example.test/api")
assert seen[0].url.raw_path.endswith(b"/vhosts/example.test%2Fapi")

async def test_create_domain_rejects_an_empty_name(
self, make_client: Callable[..., CleverCloudClient]
) -> None:
client = make_client(lambda r: httpx.Response(200))
async with client:
with pytest.raises(ValueError, match="domain"):
await client.create_domain("orga_1", "app_1", domain="/")

async def test_list_domains(
self, make_client: Callable[..., CleverCloudClient]
) -> None:
Expand Down
5 changes: 5 additions & 0 deletions tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,11 @@ def test_missing_fqdn_is_rejected(self) -> None:
with pytest.raises(InvalidResponseError, match="'fqdn'"):
Domain.from_api_response({})

@pytest.mark.parametrize("fqdn", ["/", "///"])
def test_slash_only_fqdn_is_rejected(self, fqdn: str) -> None:
with pytest.raises(InvalidResponseError, match="'fqdn'"):
Domain.from_api_response({"fqdn": fqdn})


class TestTcpRedirection:
def test_parses_namespace_and_port(self) -> None:
Expand Down
Loading