diff --git a/plane/api/base_resource.py b/plane/api/base_resource.py
index bd8bb4a..2f89e1e 100644
--- a/plane/api/base_resource.py
+++ b/plane/api/base_resource.py
@@ -37,11 +37,18 @@ def _get(self, endpoint: str, params: Mapping[str, Any] | None = None) -> Any:
return self._handle_response(response)
def _post(
- self, endpoint: str, data: Mapping[str, Any] | list[Any] | None = None
+ self,
+ endpoint: str,
+ data: Mapping[str, Any] | list[Any] | None = None,
+ params: Mapping[str, Any] | None = None,
) -> Any:
url = self._build_url(endpoint)
response = self.session.post(
- url, headers=self._headers(), json=data, timeout=self.config.timeout
+ url,
+ headers=self._headers(),
+ json=data,
+ params=params,
+ timeout=self.config.timeout,
)
return self._handle_response(response)
@@ -59,10 +66,19 @@ def _patch(self, endpoint: str, data: Mapping[str, Any] | None = None) -> Any:
)
return self._handle_response(response)
- def _delete(self, endpoint: str, data: Mapping[str, Any] | None = None) -> None:
+ def _delete(
+ self,
+ endpoint: str,
+ data: Mapping[str, Any] | None = None,
+ params: Mapping[str, Any] | None = None,
+ ) -> None:
url = self._build_url(endpoint)
response = self.session.delete(
- url, headers=self._headers(), json=data, timeout=self.config.timeout
+ url,
+ headers=self._headers(),
+ json=data,
+ params=params,
+ timeout=self.config.timeout,
)
self._handle_response(response)
diff --git a/plane/api/customers/base.py b/plane/api/customers/base.py
index 65994bb..8d37048 100644
--- a/plane/api/customers/base.py
+++ b/plane/api/customers/base.py
@@ -3,7 +3,9 @@
from plane.api.base_resource import BaseResource
from plane.api.customers.properties import CustomerProperties
+from plane.api.customers.property_values import CustomerPropertyValues
from plane.api.customers.requests import CustomerRequests
+from plane.api.customers.work_items import CustomerWorkItems
from plane.models.customers import (
CreateCustomer,
Customer,
@@ -18,7 +20,9 @@ def __init__(self, config: Any) -> None:
# Initialize sub-resources
self.properties = CustomerProperties(config)
+ self.property_values = CustomerPropertyValues(config)
self.requests = CustomerRequests(config)
+ self.work_items = CustomerWorkItems(config)
def list(
self, workspace_slug: str, params: Mapping[str, Any] | None = None
@@ -27,7 +31,7 @@ def list(
Args:
workspace_slug: The workspace slug identifier
- params: Optional query parameters
+ params: Optional query parameters, e.g. `query` to search by name
"""
response = self._get(f"{workspace_slug}/customers", params=params)
return PaginatedCustomerResponse.model_validate(response)
@@ -46,7 +50,7 @@ def retrieve(
return Customer.model_validate(response)
def create(self, workspace_slug: str, data: CreateCustomer) -> Customer:
- """Create a new customer.
+ """Create a customer, or update the one it matches.
Args:
workspace_slug: The workspace slug identifier
@@ -68,11 +72,33 @@ def update(self, workspace_slug: str, customer_id: str, data: UpdateCustomer) ->
)
return Customer.model_validate(response)
- def delete(self, workspace_slug: str, customer_id: str) -> None:
- """Delete a customer by ID.
+ def delete(
+ self,
+ workspace_slug: str,
+ customer_id: str | None = None,
+ external_source: str | None = None,
+ external_id: str | None = None,
+ ) -> None:
+ """Delete a customer, addressed by ID or by its external reference.
+
+ Pass `customer_id`, or both `external_source` and `external_id` for a
+ customer synced from an external system. Deleting by an external reference
+ that matches nothing succeeds silently.
Args:
workspace_slug: The workspace slug identifier
customer_id: UUID of the customer
+ external_source: External system the customer came from
+ external_id: The customer's identifier in that system
+
+ Raises:
+ ValueError: If neither a customer_id nor a full external reference is given
"""
- return self._delete(f"{workspace_slug}/customers/{customer_id}")
+ if customer_id:
+ return self._delete(f"{workspace_slug}/customers/{customer_id}")
+ if external_source and external_id:
+ return self._delete(
+ f"{workspace_slug}/customers",
+ params={"external_source": external_source, "external_id": external_id},
+ )
+ raise ValueError("Provide customer_id, or both external_source and external_id.")
diff --git a/plane/api/customers/property_values.py b/plane/api/customers/property_values.py
new file mode 100644
index 0000000..b2a7897
--- /dev/null
+++ b/plane/api/customers/property_values.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+from typing import Any
+
+from plane.api.base_resource import BaseResource
+from plane.models.customers import (
+ CustomerPropertyValues as CustomerPropertyValuesResponse,
+)
+from plane.models.customers import (
+ SetCustomerPropertyValue,
+ SetCustomerPropertyValues,
+)
+
+
+class CustomerPropertyValues(BaseResource):
+ """API resource for the values a customer holds for its properties.
+
+ Values are addressed as a mapping of property id to a list of strings, for
+ every property type — dates as ``YYYY-MM-DD``, booleans as ``"True"`` /
+ ``"False"``, options and relations as UUIDs. Properties that are not
+ ``is_active`` are left out of reads.
+
+ Writes replace the values of the properties they name, rather than merging
+ into them. Properties absent from the payload keep their current values.
+ """
+
+ def __init__(self, config: Any) -> None:
+ super().__init__(config, "/workspaces/")
+
+ def list(self, workspace_slug: str, customer_id: str) -> dict[str, list[str]]:
+ """Retrieve every property value set on a customer.
+
+ Properties that have no value set are absent from the mapping.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+
+ Returns:
+ Property id mapped to that property's values
+ """
+ response = self._get(f"{workspace_slug}/customers/{customer_id}/property-values")
+ return CustomerPropertyValuesResponse.model_validate(response or {}).root
+
+ def create(
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ data: SetCustomerPropertyValues,
+ ) -> None:
+ """Set several of a customer's property values at once.
+
+ Acts as an upsert: the properties named are given these values, replacing
+ any they already held. Properties absent from the payload are untouched.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ data: Property ids mapped to their new values
+
+ Raises:
+ HttpError: 400 if a value fails its property's validation
+ """
+ self._post(
+ f"{workspace_slug}/customers/{customer_id}/property-values",
+ data.model_dump(exclude_none=True),
+ )
+
+ def retrieve(self, workspace_slug: str, customer_id: str, property_id: str) -> list[str]:
+ """Retrieve the values a customer holds for one property.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ property_id: UUID of the customer property
+
+ Returns:
+ The property's values, empty when none are set
+
+ Raises:
+ HttpError: 404 if the customer or property does not exist
+ """
+ response = self._get(
+ f"{workspace_slug}/customers/{customer_id}/property-values/{property_id}"
+ )
+ values = CustomerPropertyValuesResponse.model_validate(response or {}).root
+ return values.get(str(property_id), [])
+
+ def update(
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ property_id: str,
+ data: SetCustomerPropertyValue,
+ ) -> None:
+ """Replace the values a customer holds for one property.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ property_id: UUID of the customer property
+ data: The property's new values
+
+ Raises:
+ HttpError: 400 if the property is required and the values are empty,
+ or if a value fails the property's validation
+ HttpError: 404 if the customer or property does not exist
+ """
+ self._patch(
+ f"{workspace_slug}/customers/{customer_id}/property-values/{property_id}",
+ data.model_dump(exclude_none=True),
+ )
diff --git a/plane/api/customers/requests.py b/plane/api/customers/requests.py
index ee43ed1..61250d1 100644
--- a/plane/api/customers/requests.py
+++ b/plane/api/customers/requests.py
@@ -3,7 +3,9 @@
from plane.api.base_resource import BaseResource
from plane.models.customers import (
+ CreateCustomerRequest,
CustomerRequest,
+ PaginatedCustomerRequestResponse,
UpdateCustomerRequest,
)
@@ -17,18 +19,20 @@ def list(
workspace_slug: str,
customer_id: str,
params: Mapping[str, Any] | None = None,
- ) -> list[CustomerRequest]:
- """List customer requests.
+ ) -> PaginatedCustomerRequestResponse:
+ """List customer requests (paginated).
+
+ Returns one page (1000 by default). Pass `per_page`/`cursor` in params and
+ follow `next_cursor` to page through the rest.
Args:
workspace_slug: The workspace slug identifier
customer_id: UUID of the customer
- params: Optional query parameters
+ params: Optional query parameters, e.g. `query` to search by name,
+ `per_page` to size the page, `cursor` to walk pages
"""
response = self._get(f"{workspace_slug}/customers/{customer_id}/requests", params=params)
- if isinstance(response, list):
- return [CustomerRequest.model_validate(item) for item in response]
- return []
+ return PaginatedCustomerRequestResponse.model_validate(response)
def retrieve(self, workspace_slug: str, customer_id: str, request_id: str) -> CustomerRequest:
"""Retrieve a customer request by ID.
@@ -42,14 +46,23 @@ def retrieve(self, workspace_slug: str, customer_id: str, request_id: str) -> Cu
return CustomerRequest.model_validate(response)
def create(
- self, workspace_slug: str, customer_id: str, data: CustomerRequest
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ data: CreateCustomerRequest | CustomerRequest,
) -> CustomerRequest:
"""Create a new customer request.
+ `data.work_item_ids` links work items to the customer as the request is
+ created. The returned request does not echo them back; read them with
+ `client.customers.work_items.list()`.
+
Args:
workspace_slug: The workspace slug identifier
customer_id: UUID of the customer
- data: Customer request data
+ data: Customer request data. Prefer `CreateCustomerRequest`;
+ `CustomerRequest` is still accepted for callers written against
+ earlier versions.
"""
response = self._post(
f"{workspace_slug}/customers/{customer_id}/requests",
diff --git a/plane/api/customers/work_items.py b/plane/api/customers/work_items.py
new file mode 100644
index 0000000..97d13e7
--- /dev/null
+++ b/plane/api/customers/work_items.py
@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from plane.api.base_resource import BaseResource
+from plane.models.customers import (
+ CustomerWorkItem,
+ LinkCustomerWorkItems,
+ LinkCustomerWorkItemsResponse,
+)
+
+
+class CustomerWorkItems(BaseResource):
+ """API resource for the work items linked to a customer.
+
+ Links may hang off the customer alone, or off one of its requests when a
+ `customer_request_id` is supplied.
+ """
+
+ def __init__(self, config: Any) -> None:
+ super().__init__(config, "/workspaces/")
+
+ def list(
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ customer_request_id: str | None = None,
+ search: str | None = None,
+ params: Mapping[str, Any] | None = None,
+ ) -> list[CustomerWorkItem]:
+ """List the work items linked to a customer.
+
+ This endpoint is not paginated and returns every linked work item.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ customer_request_id: Only return work items linked via this request
+ search: Match on work item name, sequence ID or project identifier
+ params: Additional query parameters
+ """
+ query: dict[str, Any] = dict(params or {})
+ if customer_request_id:
+ query["customer_request_id"] = customer_request_id
+ if search:
+ query["search"] = search
+
+ response = self._get(
+ f"{workspace_slug}/customers/{customer_id}/issues", params=query or None
+ )
+ return [CustomerWorkItem.model_validate(item) for item in response]
+
+ def create(
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ data: LinkCustomerWorkItems,
+ customer_request_id: str | None = None,
+ ) -> LinkCustomerWorkItemsResponse:
+ """Link one or more work items to a customer.
+
+ The API rejects the whole call if any work item in `data.work_item_ids`
+ does not exist. Re-linking an already linked work item is a no-op.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ data: The work items to link
+ customer_request_id: Attach the links to this request of the customer
+ """
+ query = {"customer_request_id": customer_request_id} if customer_request_id else None
+ response = self._post(
+ f"{workspace_slug}/customers/{customer_id}/issues",
+ data.model_dump(exclude_none=True, by_alias=True),
+ params=query,
+ )
+ return LinkCustomerWorkItemsResponse.model_validate(response)
+
+ def delete(
+ self,
+ workspace_slug: str,
+ customer_id: str,
+ work_item_id: str,
+ customer_request_id: str | None = None,
+ ) -> None:
+ """Unlink a work item from a customer.
+
+ Args:
+ workspace_slug: The workspace slug identifier
+ customer_id: UUID of the customer
+ work_item_id: UUID of the work item to unlink
+ customer_request_id: Only drop the link made via this request;
+ without it every link between the customer and the work item goes
+
+ Raises:
+ HttpError: 404 if no such link exists
+ """
+ query = {"customer_request_id": customer_request_id} if customer_request_id else None
+ return self._delete(
+ f"{workspace_slug}/customers/{customer_id}/issues/{work_item_id}", params=query
+ )
diff --git a/plane/models/customers.py b/plane/models/customers.py
index 417a245..924a256 100644
--- a/plane/models/customers.py
+++ b/plane/models/customers.py
@@ -1,6 +1,6 @@
from typing import Any
-from pydantic import BaseModel, ConfigDict, field_serializer, model_validator
+from pydantic import BaseModel, ConfigDict, Field, RootModel, field_serializer, model_validator
from .enums import PropertyType, RelationType
from .pagination import PaginatedResponse
@@ -36,6 +36,9 @@ class Customer(BaseModel):
stage: str | None = None
contract_status: str | None = None
revenue: str | None = None
+ external_source: str | None = None
+ external_id: str | None = None
+ archived_at: str | None = None
created_by: str | None = None
updated_by: str | None = None
logo_asset: str | None = None
@@ -43,7 +46,13 @@ class Customer(BaseModel):
class CreateCustomer(BaseModel):
- """Request model for creating a customer."""
+ """Request model for creating a customer.
+
+ The create endpoint is an upsert. When `external_id` and `external_source`
+ are both set, an existing customer matching them is updated; otherwise a
+ customer with the same `name` is updated. Only when neither matches is a new
+ customer created.
+ """
model_config = ConfigDict(extra="ignore", populate_by_name=True)
@@ -59,6 +68,8 @@ class CreateCustomer(BaseModel):
stage: str | None = None
contract_status: str | None = None
revenue: str | None = None
+ external_source: str | None = None
+ external_id: str | None = None
created_by: str | None = None
updated_by: str | None = None
logo_asset: str | None = None
@@ -81,11 +92,61 @@ class UpdateCustomer(BaseModel):
stage: str | None = None
contract_status: str | None = None
revenue: str | None = None
+ external_source: str | None = None
+ external_id: str | None = None
created_by: str | None = None
updated_by: str | None = None
logo_asset: str | None = None
+class CustomerPropertyOption(BaseModel):
+ """Customer property option model (values of an OPTION property)."""
+
+ model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+ id: str | None = None
+ deleted_at: str | None = None
+ created_at: str | None = None
+ updated_at: str | None = None
+ name: str
+ sort_order: float | None = None
+ description: str | None = None
+ logo_props: Any | None = None
+ is_active: bool | None = None
+ is_default: bool | None = None
+ external_source: str | None = None
+ external_id: str | None = None
+ created_by: str | None = None
+ updated_by: str | None = None
+ workspace: str | None = None
+ property: str | None = None
+ parent: str | None = None
+
+
+class CreateCustomerPropertyOption(BaseModel):
+ """Request model for an OPTION property's option, sent inline on create."""
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ name: str
+ description: str | None = None
+ is_default: bool | None = None
+
+
+class UpdateCustomerPropertyOption(BaseModel):
+ """Request model for an OPTION property's option, sent inline on update.
+
+ Set `id` to update an existing option; omit it to create a new one.
+ """
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ id: str | None = None
+ name: str | None = None
+ description: str | None = None
+ is_default: bool | None = None
+
+
class CustomerProperty(BaseModel):
"""Customer property model."""
@@ -113,13 +174,14 @@ class CustomerProperty(BaseModel):
created_by: str | None = None
updated_by: str | None = None
workspace: str | None = None
+ options: list[CustomerPropertyOption] | None = None
@field_serializer("property_type")
- def serialize_property_type(self, value: PropertyType) -> str:
+ def serialize_property_type(self, value: PropertyType) -> str | None:
return value.value if value else None
@field_serializer("relation_type")
- def serialize_relation_type(self, value: RelationType) -> str:
+ def serialize_relation_type(self, value: RelationType) -> str | None:
return value.value if value else None
@@ -143,13 +205,14 @@ class CreateCustomerProperty(BaseModel):
validation_rules: Any | None = None
external_source: str | None = None
external_id: str | None = None
+ options: list[CreateCustomerPropertyOption] | None = None
@field_serializer("property_type")
- def serialize_property_type(self, value: PropertyType) -> str:
+ def serialize_property_type(self, value: PropertyType) -> str | None:
return value.value if value else None
@field_serializer("relation_type")
- def serialize_relation_type(self, value: RelationType) -> str:
+ def serialize_relation_type(self, value: RelationType) -> str | None:
return value.value if value else None
@model_validator(mode="after")
@@ -172,7 +235,7 @@ def validate_settings_and_relation_type(self) -> "CreateCustomerProperty":
if prop_type == PropertyType.DATETIME:
if settings is None:
raise ValueError(
- "settings with DateAttributeSettings is required for DATETIME " "properties"
+ "settings with DateAttributeSettings is required for DATETIME properties"
)
if not isinstance(settings, DateAttributeSettings):
raise ValueError("settings must be DateAttributeSettings for DATETIME properties")
@@ -206,13 +269,14 @@ class UpdateCustomerProperty(BaseModel):
external_id: str | None = None
created_by: str | None = None
updated_by: str | None = None
+ options: list[UpdateCustomerPropertyOption] | None = None
@field_serializer("property_type")
- def serialize_property_type(self, value: PropertyType) -> str:
+ def serialize_property_type(self, value: PropertyType) -> str | None:
return value.value if value else None
@field_serializer("relation_type")
- def serialize_relation_type(self, value: RelationType) -> str:
+ def serialize_relation_type(self, value: RelationType) -> str | None:
return value.value if value else None
@model_validator(mode="after")
@@ -267,6 +331,24 @@ class CustomerRequest(BaseModel):
description_html: str | None = None
link: str | None = None
work_item_ids: list[str] | None = None
+ attachment_count: int | None = None
+ created_at: str | None = None
+
+
+class CreateCustomerRequest(BaseModel):
+ """Request model for creating a customer request.
+
+ `work_item_ids` links work items to the customer at creation time. It is
+ write-only — the created request echoed back does not include it.
+ """
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ name: str
+ description: Any | None = None
+ description_html: str | None = None
+ link: str | None = None
+ work_item_ids: list[str] | None = None
class UpdateCustomerRequest(BaseModel):
@@ -281,6 +363,83 @@ class UpdateCustomerRequest(BaseModel):
work_item_ids: list[str] | None = None
+class CustomerWorkItem(BaseModel):
+ """A work item linked to a customer."""
+
+ model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+ id: str
+ name: str | None = None
+ state_id: str | None = None
+ priority: str | None = None
+ sequence_id: int | None = None
+ project_id: str | None = None
+ project_identifier: str | None = Field(None, alias="project__identifier")
+ created_at: str | None = None
+ updated_at: str | None = None
+
+
+class LinkCustomerWorkItems(BaseModel):
+ """Request model for linking work items to a customer."""
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ work_item_ids: list[str] = Field(..., alias="issue_ids")
+
+
+class LinkedCustomerWorkItem(BaseModel):
+ """A work item as echoed back by the link endpoint."""
+
+ model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+ id: str
+ name: str | None = None
+ sequence_id: int | None = None
+ project_id: str | None = None
+ project_identifier: str | None = Field(None, alias="project__identifier")
+
+
+class LinkCustomerWorkItemsResponse(BaseModel):
+ """Response returned when linking work items to a customer."""
+
+ model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+ message: str | None = None
+ linked_work_items: list[LinkedCustomerWorkItem] = Field(
+ default_factory=list, alias="linked_issues"
+ )
+
+
+class SetCustomerPropertyValues(BaseModel):
+ """Request model for setting several customer property values at once.
+
+ Maps property id to its list of values. Every value is sent as a string
+ regardless of the property's type.
+ """
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ customer_property_values: dict[str, list[str]]
+
+
+class SetCustomerPropertyValue(BaseModel):
+ """Request model for setting a single customer property's values."""
+
+ model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+ values: list[str]
+
+
+class CustomerPropertyValues(RootModel[dict[str, list[str]]]):
+ """Response model for a customer's property values.
+
+ Maps property id to that property's values. Every value is a string whatever
+ the property's type.
+ """
+
+ root: dict[str, list[str]]
+
+
class PaginatedCustomerResponse(PaginatedResponse):
"""Paginated response for customers list endpoint.
@@ -303,3 +462,15 @@ class PaginatedCustomerPropertyResponse(PaginatedResponse):
model_config = ConfigDict(extra="allow", populate_by_name=True)
results: list[CustomerProperty]
+
+
+class PaginatedCustomerRequestResponse(PaginatedResponse):
+ """Paginated response for customer requests list endpoint.
+
+ All pagination fields from PaginatedResponse are required.
+ The results field contains the list of CustomerRequest objects.
+ """
+
+ model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+ results: list[CustomerRequest]
diff --git a/pyproject.toml b/pyproject.toml
index ee42cd6..b10ada4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "plane-sdk"
-version = "0.2.19"
+version = "0.2.20"
description = "Python SDK for Plane API"
readme = "README.md"
requires-python = ">=3.10"
diff --git a/tests/unit/test_customers.py b/tests/unit/test_customers.py
index 2be9564..8b0e354 100644
--- a/tests/unit/test_customers.py
+++ b/tests/unit/test_customers.py
@@ -1,9 +1,42 @@
"""Unit tests for Customers API resource (smoke tests with real HTTP requests)."""
+import re
+import time
+import unicodedata
+import uuid
+
import pytest
from plane.client import PlaneClient
-from plane.models.customers import CreateCustomer, UpdateCustomer
+from plane.errors.errors import HttpError
+from plane.models.customers import (
+ CreateCustomer,
+ CreateCustomerProperty,
+ CreateCustomerPropertyOption,
+ CreateCustomerRequest,
+ LinkCustomerWorkItems,
+ SetCustomerPropertyValue,
+ SetCustomerPropertyValues,
+ UpdateCustomer,
+ UpdateCustomerProperty,
+ UpdateCustomerPropertyOption,
+ UpdateCustomerRequest,
+)
+from plane.models.enums import PropertyType
+from plane.models.projects import Project
+from plane.models.work_item_property_configurations import TextAttributeSettings
+from plane.models.work_items import CreateWorkItem
+
+
+def _unique(prefix: str) -> str:
+ """Build a name that will not collide with other runs."""
+ return f"{prefix} {int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"
+
+
+def slugify(value: str) -> str:
+ value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
+ value = re.sub(r"[^\w\s-]", "", value.lower())
+ return re.sub(r"[-\s]+", "-", value).strip("-_")
class TestCustomersAPI:
@@ -18,11 +51,18 @@ def test_list_customers(self, client: PlaneClient, workspace_slug: str) -> None:
assert isinstance(response.results, list)
def test_list_customers_with_params(self, client: PlaneClient, workspace_slug: str) -> None:
- """Test listing customers with query parameters."""
- response = client.customers.list(workspace_slug, params={"limit": 10})
+ """The page size is honoured. The paginator reads `per_page`, not `limit`."""
+ response = client.customers.list(workspace_slug, params={"per_page": 2})
assert response is not None
assert hasattr(response, "results")
- assert len(response.results) <= 10
+ assert len(response.results) <= 2
+
+ def test_list_customers_paginates(self, client: PlaneClient, workspace_slug: str) -> None:
+ """A workspace holding more customers than a page reports the rest."""
+ response = client.customers.list(workspace_slug, params={"per_page": 1})
+ if response.total_count > 1:
+ assert len(response.results) == 1
+ assert response.next_page_results is True
class TestCustomersAPICRUD:
@@ -31,9 +71,8 @@ class TestCustomersAPICRUD:
@pytest.fixture
def customer_data(self) -> CreateCustomer:
"""Create test customer data."""
- import time
return CreateCustomer(
- name=f"Test Customer {int(time.time())}",
+ name=_unique("Test Customer"),
email=f"test{int(time.time())}@example.com",
)
@@ -55,7 +94,7 @@ def test_create_customer(
assert customer is not None
assert customer.id is not None
assert customer.name == customer_data.name
-
+
# Cleanup
try:
client.customers.delete(workspace_slug, customer.id)
@@ -71,9 +110,553 @@ def test_retrieve_customer(self, client: PlaneClient, workspace_slug: str, custo
def test_update_customer(self, client: PlaneClient, workspace_slug: str, customer) -> None:
"""Test updating a customer."""
- update_data = UpdateCustomer(name="Updated Customer Name")
+ new_name = _unique("Updated Customer")
+ update_data = UpdateCustomer(name=new_name)
updated = client.customers.update(workspace_slug, customer.id, update_data)
assert updated is not None
assert updated.id == customer.id
- assert updated.name == "Updated Customer Name"
+ assert updated.name == new_name
+
+ def test_create_customer_with_full_payload(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """Every field the create endpoint accepts survives the round trip."""
+ data = CreateCustomer(
+ name=_unique("Full Customer"),
+ description_html="A customer
",
+ email="full@example.com",
+ website_url="https://example.com",
+ domain="example.com",
+ employees=42,
+ stage="lead",
+ contract_status="active",
+ revenue="1000000",
+ )
+ customer = client.customers.create(workspace_slug, data)
+ try:
+ assert customer.name == data.name
+ assert customer.email == data.email
+ assert customer.domain == data.domain
+ assert customer.employees == 42
+ assert customer.revenue == "1000000"
+ finally:
+ try:
+ client.customers.delete(workspace_slug, customer.id)
+ except Exception:
+ pass
+
+ def test_delete_customer(self, client: PlaneClient, workspace_slug: str) -> None:
+ """A deleted customer is gone."""
+ customer = client.customers.create(workspace_slug, CreateCustomer(name=_unique("Doomed")))
+ client.customers.delete(workspace_slug, customer.id)
+
+ with pytest.raises(HttpError):
+ client.customers.retrieve(workspace_slug, customer.id)
+
+
+class TestCustomersUpsert:
+ """The create endpoint upserts rather than always creating."""
+
+ def test_create_with_same_name_updates_existing(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """Creating twice under one name returns the same customer, updated."""
+ name = _unique("Upsert By Name")
+ first = client.customers.create(workspace_slug, CreateCustomer(name=name, employees=1))
+ try:
+ second = client.customers.create(
+ workspace_slug, CreateCustomer(name=name, employees=99)
+ )
+ assert second.id == first.id
+ assert second.employees == 99
+ finally:
+ try:
+ client.customers.delete(workspace_slug, first.id)
+ except Exception:
+ pass
+
+ def test_create_with_same_external_ref_updates_existing(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """An external reference matches an existing customer even under a new name."""
+ external_id = uuid.uuid4().hex
+ first = client.customers.create(
+ workspace_slug,
+ CreateCustomer(
+ name=_unique("External A"),
+ external_source="pytest",
+ external_id=external_id,
+ ),
+ )
+ try:
+ renamed = _unique("External B")
+ second = client.customers.create(
+ workspace_slug,
+ CreateCustomer(
+ name=renamed,
+ external_source="pytest",
+ external_id=external_id,
+ ),
+ )
+ assert second.id == first.id
+ assert second.name == renamed
+ finally:
+ try:
+ client.customers.delete(workspace_slug, first.id)
+ except Exception:
+ pass
+
+ def test_delete_by_external_id(self, client: PlaneClient, workspace_slug: str) -> None:
+ """A customer can be deleted through its external reference."""
+ external_id = uuid.uuid4().hex
+ customer = client.customers.create(
+ workspace_slug,
+ CreateCustomer(
+ name=_unique("External Delete"),
+ external_source="pytest",
+ external_id=external_id,
+ ),
+ )
+ client.customers.delete(workspace_slug, external_source="pytest", external_id=external_id)
+
+ with pytest.raises(HttpError):
+ client.customers.retrieve(workspace_slug, customer.id)
+
+ def test_delete_by_unknown_external_id_is_silent(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """Deleting an external reference that matches nothing does not raise."""
+ client.customers.delete(
+ workspace_slug, external_source="pytest", external_id=uuid.uuid4().hex
+ )
+
+
+@pytest.fixture
+def customer(client: PlaneClient, workspace_slug: str):
+ """A customer shared by the request, issue and property-value tests."""
+ customer = client.customers.create(workspace_slug, CreateCustomer(name=_unique("Customer")))
+ yield customer
+ try:
+ client.customers.delete(workspace_slug, customer.id)
+ except Exception:
+ pass
+
+
+class TestCustomerProperties:
+ """Test customer property operations."""
+
+ @pytest.fixture
+ def text_property(self, client: PlaneClient, workspace_slug: str):
+ """A workspace-level TEXT property, torn down after the test."""
+ prop = client.customers.properties.create(
+ workspace_slug,
+ CreateCustomerProperty(
+ name="ignored",
+ display_name=_unique("Tier"),
+ property_type=PropertyType.TEXT,
+ settings=TextAttributeSettings(display_format="single-line"),
+ ),
+ )
+ yield prop
+ try:
+ client.customers.properties.delete(workspace_slug, prop.id)
+ except Exception:
+ pass
+
+ def test_list_properties(self, client: PlaneClient, workspace_slug: str) -> None:
+ """Listing properties returns a paginated response."""
+ response = client.customers.properties.list(workspace_slug)
+ assert hasattr(response, "results")
+ assert isinstance(response.results, list)
+
+ def test_create_text_property(self, client: PlaneClient, workspace_slug: str, text_property):
+ """A TEXT property carries its settings."""
+ assert text_property.id is not None
+ assert text_property.property_type == PropertyType.TEXT
+ assert text_property.display_name.startswith("Tier")
+
+ def test_name_is_slugified_display_name(
+ self, client: PlaneClient, workspace_slug: str, text_property
+ ) -> None:
+ """The stored name is the slug of display_name, not the name that was sent."""
+ assert text_property.name == slugify(text_property.display_name)
+ assert text_property.name != "ignored"
+
+ def test_created_property_appears_in_list(
+ self, client: PlaneClient, workspace_slug: str, text_property
+ ) -> None:
+ """A created property is listed."""
+ response = client.customers.properties.list(workspace_slug)
+ assert text_property.id in [p.id for p in response.results]
+
+ def test_retrieve_property(self, client: PlaneClient, workspace_slug: str, text_property):
+ """A property can be fetched by id."""
+ retrieved = client.customers.properties.retrieve(workspace_slug, text_property.id)
+ assert retrieved.id == text_property.id
+
+ def test_update_property(self, client: PlaneClient, workspace_slug: str, text_property):
+ """display_name is updatable, and re-slugs the stored name with it."""
+ new_display_name = _unique("Updated Tier")
+ updated = client.customers.properties.update(
+ workspace_slug,
+ text_property.id,
+ UpdateCustomerProperty(display_name=new_display_name),
+ )
+ assert updated.display_name == new_display_name
+ assert updated.name == slugify(new_display_name)
+
+ def test_create_option_property_with_options(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """An OPTION property is created together with its options."""
+ prop = client.customers.properties.create(
+ workspace_slug,
+ CreateCustomerProperty(
+ name="ignored",
+ display_name=_unique("Segment"),
+ property_type=PropertyType.OPTION,
+ options=[
+ CreateCustomerPropertyOption(name="SMB", is_default=True),
+ CreateCustomerPropertyOption(name="Enterprise"),
+ ],
+ ),
+ )
+ try:
+ assert prop.options is not None
+ assert {o.name for o in prop.options} == {"SMB", "Enterprise"}
+ finally:
+ try:
+ client.customers.properties.delete(workspace_slug, prop.id)
+ except Exception:
+ pass
+
+ def test_update_option_property_adds_option(
+ self, client: PlaneClient, workspace_slug: str
+ ) -> None:
+ """An option can be appended to an existing OPTION property."""
+ prop = client.customers.properties.create(
+ workspace_slug,
+ CreateCustomerProperty(
+ name="ignored",
+ display_name=_unique("Segment"),
+ property_type=PropertyType.OPTION,
+ options=[CreateCustomerPropertyOption(name="SMB")],
+ ),
+ )
+ try:
+ updated = client.customers.properties.update(
+ workspace_slug,
+ prop.id,
+ UpdateCustomerProperty(options=[UpdateCustomerPropertyOption(name="Mid-Market")]),
+ )
+ assert "Mid-Market" in {o.name for o in (updated.options or [])}
+ finally:
+ try:
+ client.customers.properties.delete(workspace_slug, prop.id)
+ except Exception:
+ pass
+
+ def test_delete_property(self, client: PlaneClient, workspace_slug: str) -> None:
+ """A deleted property is gone."""
+ prop = client.customers.properties.create(
+ workspace_slug,
+ CreateCustomerProperty(
+ name="ignored",
+ display_name=_unique("Doomed"),
+ property_type=PropertyType.TEXT,
+ settings=TextAttributeSettings(display_format="single-line"),
+ ),
+ )
+ client.customers.properties.delete(workspace_slug, prop.id)
+
+ with pytest.raises(HttpError):
+ client.customers.properties.retrieve(workspace_slug, prop.id)
+
+ def test_text_property_requires_settings(self) -> None:
+ """A TEXT property without settings is rejected before it reaches the API."""
+ with pytest.raises(ValueError, match="TextAttributeSettings is required"):
+ CreateCustomerProperty(
+ name="no_settings",
+ display_name="No Settings",
+ property_type=PropertyType.TEXT,
+ )
+
+ def test_relation_property_requires_relation_type(self) -> None:
+ """A RELATION property without relation_type is rejected before it reaches the API."""
+ with pytest.raises(ValueError, match="relation_type is required"):
+ CreateCustomerProperty(
+ name="no_relation",
+ display_name="No Relation",
+ property_type=PropertyType.RELATION,
+ )
+
+ def test_update_model_keeps_immutable_fields(self) -> None:
+ """The update model keeps fields the API refuses to change."""
+ fields = set(UpdateCustomerProperty.model_fields)
+ assert {"property_type", "is_multi", "settings"} <= fields
+
+
+class TestCustomerPropertyValues:
+ """Test customer property value operations."""
+
+ @pytest.fixture
+ def text_property(self, client: PlaneClient, workspace_slug: str):
+ """A TEXT property to hold values against."""
+ prop = client.customers.properties.create(
+ workspace_slug,
+ CreateCustomerProperty(
+ name="ignored",
+ display_name=_unique("Notes"),
+ property_type=PropertyType.TEXT,
+ settings=TextAttributeSettings(display_format="single-line"),
+ ),
+ )
+ yield prop
+ try:
+ client.customers.properties.delete(workspace_slug, prop.id)
+ except Exception:
+ pass
+
+ def test_list_values_empty(self, client: PlaneClient, workspace_slug: str, customer) -> None:
+ """A customer with no values set reads back a mapping without them."""
+ values = client.customers.property_values.list(workspace_slug, customer.id)
+ assert isinstance(values, dict)
+
+ def test_set_and_list_values(
+ self, client: PlaneClient, workspace_slug: str, customer, text_property
+ ) -> None:
+ """Values set in bulk read back."""
+ client.customers.property_values.create(
+ workspace_slug,
+ customer.id,
+ SetCustomerPropertyValues(customer_property_values={text_property.id: ["hello"]}),
+ )
+ values = client.customers.property_values.list(workspace_slug, customer.id)
+ assert values.get(text_property.id) == ["hello"]
+
+ def test_update_and_retrieve_single_value(
+ self, client: PlaneClient, workspace_slug: str, customer, text_property
+ ) -> None:
+ """A single property's values can be replaced and read back."""
+ client.customers.property_values.update(
+ workspace_slug,
+ customer.id,
+ text_property.id,
+ SetCustomerPropertyValue(values=["first"]),
+ )
+ assert client.customers.property_values.retrieve(
+ workspace_slug, customer.id, text_property.id
+ ) == ["first"]
+ client.customers.property_values.update(
+ workspace_slug,
+ customer.id,
+ text_property.id,
+ SetCustomerPropertyValue(values=["second"]),
+ )
+ assert client.customers.property_values.retrieve(
+ workspace_slug, customer.id, text_property.id
+ ) == ["second"]
+
+ def test_retrieve_unset_value_is_empty(
+ self, client: PlaneClient, workspace_slug: str, customer, text_property
+ ) -> None:
+ """A property with no value reads back as an empty list, not an error."""
+ assert (
+ client.customers.property_values.retrieve(workspace_slug, customer.id, text_property.id)
+ == []
+ )
+
+
+class TestCustomerRequests:
+ """Test customer request operations."""
+
+ @pytest.fixture
+ def customer_request(self, client: PlaneClient, workspace_slug: str, customer):
+ """A request on the shared customer."""
+ req = client.customers.requests.create(
+ workspace_slug,
+ customer.id,
+ CreateCustomerRequest(
+ name=_unique("Request"),
+ description_html="Please build this
",
+ ),
+ )
+ yield req
+ try:
+ client.customers.requests.delete(workspace_slug, customer.id, req.id)
+ except Exception:
+ pass
+
+ def test_create_request(self, customer_request) -> None:
+ """A request comes back with an id."""
+ assert customer_request.id is not None
+ assert customer_request.name is not None
+
+ def test_retrieve_request(
+ self, client: PlaneClient, workspace_slug: str, customer, customer_request
+ ) -> None:
+ """A request can be fetched by id."""
+ retrieved = client.customers.requests.retrieve(
+ workspace_slug, customer.id, customer_request.id
+ )
+ assert retrieved.id == customer_request.id
+
+ def test_update_request(
+ self, client: PlaneClient, workspace_slug: str, customer, customer_request
+ ) -> None:
+ """A request's name is updatable."""
+ new_name = _unique("Updated Request")
+ updated = client.customers.requests.update(
+ workspace_slug,
+ customer.id,
+ customer_request.id,
+ UpdateCustomerRequest(name=new_name),
+ )
+ assert updated.name == new_name
+
+ def test_delete_request(self, client: PlaneClient, workspace_slug: str, customer) -> None:
+ """A deleted request is gone."""
+ req = client.customers.requests.create(
+ workspace_slug, customer.id, CreateCustomerRequest(name=_unique("Doomed Request"))
+ )
+ client.customers.requests.delete(workspace_slug, customer.id, req.id)
+
+ with pytest.raises(HttpError):
+ client.customers.requests.retrieve(workspace_slug, customer.id, req.id)
+
+ def test_update_model_keeps_work_item_ids(self) -> None:
+ """The update model keeps work_item_ids even though the API discards it.
+
+ Removing it would break callers that already set it. The docstring warns
+ that it does nothing and points at customers.work_items instead.
+ """
+ assert "work_item_ids" in UpdateCustomerRequest.model_fields
+
+ def test_list_returns_envelope(
+ self, client: PlaneClient, workspace_slug: str, customer, customer_request
+ ) -> None:
+ """list() returns the paginated envelope.
+
+ It used to return [] unconditionally by testing a paginated dict with
+ isinstance(response, list); now it parses the envelope.
+ """
+ response = client.customers.requests.list(workspace_slug, customer.id)
+ assert isinstance(response.results, list)
+ assert isinstance(response.total_count, int)
+ assert customer_request.id in [r.id for r in response.results]
+
+
+class TestCustomerWorkItems:
+ """Test linking work items to customers."""
+
+ @pytest.fixture
+ def work_item(self, client: PlaneClient, workspace_slug: str, project: Project):
+ """A work item to link."""
+ item = client.work_items.create(
+ workspace_slug, project.id, CreateWorkItem(name=_unique("Linked Work Item"))
+ )
+ yield item
+ try:
+ client.work_items.delete(workspace_slug, project.id, item.id)
+ except Exception:
+ pass
+
+ def test_list_issues_empty(self, client: PlaneClient, workspace_slug: str, customer) -> None:
+ """A customer with nothing linked lists no work items."""
+ assert client.customers.work_items.list(workspace_slug, customer.id) == []
+
+ def test_link_and_list_issue(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """A linked work item is listed against the customer."""
+ response = client.customers.work_items.create(
+ workspace_slug, customer.id, LinkCustomerWorkItems(work_item_ids=[work_item.id])
+ )
+ assert work_item.id in [i.id for i in response.linked_work_items]
+
+ linked = client.customers.work_items.list(workspace_slug, customer.id)
+ assert work_item.id in [i.id for i in linked]
+
+ def test_linked_issue_carries_project_identifier(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """The project__identifier key is surfaced as project_identifier."""
+ client.customers.work_items.create(
+ workspace_slug, customer.id, LinkCustomerWorkItems(work_item_ids=[work_item.id])
+ )
+ linked = client.customers.work_items.list(workspace_slug, customer.id)
+ issue = next(i for i in linked if i.id == work_item.id)
+ assert issue.project_identifier is not None
+
+ def test_unlink_issue(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """An unlinked work item drops off the customer."""
+ client.customers.work_items.create(
+ workspace_slug, customer.id, LinkCustomerWorkItems(work_item_ids=[work_item.id])
+ )
+ client.customers.work_items.delete(workspace_slug, customer.id, work_item.id)
+
+ linked = client.customers.work_items.list(workspace_slug, customer.id)
+ assert work_item.id not in [i.id for i in linked]
+
+ def test_unlink_unlinked_issue_raises(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """Unlinking a work item that is not linked is a 404."""
+ with pytest.raises(HttpError):
+ client.customers.work_items.delete(workspace_slug, customer.id, work_item.id)
+
+ def test_link_unknown_issue_raises(
+ self, client: PlaneClient, workspace_slug: str, customer
+ ) -> None:
+ """Linking a work item that does not exist is rejected."""
+ with pytest.raises(HttpError):
+ client.customers.work_items.create(
+ workspace_slug,
+ customer.id,
+ LinkCustomerWorkItems(work_item_ids=[str(uuid.uuid4())]),
+ )
+
+ def test_link_via_request_and_filter(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """Links made through a request are filterable by that request."""
+ req = client.customers.requests.create(
+ workspace_slug, customer.id, CreateCustomerRequest(name=_unique("Scoped Request"))
+ )
+ try:
+ client.customers.work_items.create(
+ workspace_slug,
+ customer.id,
+ LinkCustomerWorkItems(work_item_ids=[work_item.id]),
+ customer_request_id=req.id,
+ )
+ scoped = client.customers.work_items.list(
+ workspace_slug, customer.id, customer_request_id=req.id
+ )
+ assert work_item.id in [i.id for i in scoped]
+ finally:
+ try:
+ client.customers.requests.delete(workspace_slug, customer.id, req.id)
+ except Exception:
+ pass
+
+ def test_create_request_with_work_item_ids_links_them(
+ self, client: PlaneClient, workspace_slug: str, customer, work_item
+ ) -> None:
+ """work_item_ids on create links the work items to the customer."""
+ req = client.customers.requests.create(
+ workspace_slug,
+ customer.id,
+ CreateCustomerRequest(name=_unique("Linking Request"), work_item_ids=[work_item.id]),
+ )
+ try:
+ linked = client.customers.work_items.list(workspace_slug, customer.id)
+ assert work_item.id in [i.id for i in linked]
+ finally:
+ try:
+ client.customers.requests.delete(workspace_slug, customer.id, req.id)
+ except Exception:
+ pass