diff --git a/atlassian/jira_models.py b/atlassian/jira_models.py new file mode 100644 index 000000000..b78518421 --- /dev/null +++ b/atlassian/jira_models.py @@ -0,0 +1,10 @@ +"""Convenience alias for atlassian.models.jira. + +Allows shorter imports: + from atlassian.jira_models import task, serialize +""" + +from __future__ import annotations + +from atlassian.models.jira import * # noqa: F401,F403 +from atlassian.models.jira import __all__ # noqa: F401 diff --git a/atlassian/models/__init__.py b/atlassian/models/__init__.py new file mode 100644 index 000000000..0cc666fe1 --- /dev/null +++ b/atlassian/models/__init__.py @@ -0,0 +1,5 @@ +"""Atlassian data models.""" + +from __future__ import annotations + +from atlassian.models.jira import * # noqa: F401,F403 diff --git a/atlassian/models/jira/__init__.py b/atlassian/models/jira/__init__.py new file mode 100644 index 000000000..4ad0fdd48 --- /dev/null +++ b/atlassian/models/jira/__init__.py @@ -0,0 +1,109 @@ +"""Type-safe Jira issue models with fluent builders. + +Quick start: + from atlassian.models.jira import task, serialize + + issue = ( + task() + .project("PROJ") + .summary("My task") + .priority("High") + .build() + ) + jira.issue_create(fields=serialize(issue)["fields"]) +""" + +from __future__ import annotations + +from atlassian.models.jira.adf import ADFBuilder, InlineNode, MentionNode, TextNode +from atlassian.models.jira.builders import ( + EpicBuilder, + IssueBuilder, + StoryBuilder, + SubTaskBuilder, + TaskBuilder, + BugBuilder, + bug, + epic, + story, + subtask, + task, +) +from atlassian.models.jira.fields import ( + Component, + CustomField, + IssueFields, + IssueLink, + IssueType, + Parent, + Priority, + PriorityLevel, + Project, + User, + Version, +) +from atlassian.models.jira.issues import ( + Bug, + Epic, + JiraIssue, + Story, + SubTask, + Task, + get_issue_type_registry, + issue_type_for, +) +from atlassian.models.jira.comment import Comment, Visibility +from atlassian.models.jira.serializer import FieldMapping, bulk_serialize, serialize, to_fields_dict +from atlassian.models.jira.transition import Transition, TransitionBuilder +from atlassian.models.jira.update import UpdateBuilder, UpdatePayload +from atlassian.models.jira.validation import ValidationError, validate, validate_or_raise + +__all__ = [ + "ADFBuilder", + "Bug", + "BugBuilder", + "Comment", + "Component", + "CustomField", + "Epic", + "EpicBuilder", + "FieldMapping", + "InlineNode", + "IssueBuilder", + "IssueFields", + "IssueLink", + "IssueType", + "JiraIssue", + "MentionNode", + "Parent", + "Priority", + "PriorityLevel", + "Project", + "Story", + "StoryBuilder", + "SubTask", + "SubTaskBuilder", + "Task", + "TaskBuilder", + "TextNode", + "Transition", + "TransitionBuilder", + "UpdateBuilder", + "UpdatePayload", + "User", + "ValidationError", + "Version", + "Visibility", + "bug", + "bulk_serialize", + "epic", + "get_issue_type_registry", + "issue_type_for", + "serialize", + "story", + "subtask", + "task", + "to_fields_dict", + "validate", + "validate_or_raise", +] diff --git a/atlassian/models/jira/adf.py b/atlassian/models/jira/adf.py new file mode 100644 index 000000000..26677e4ba --- /dev/null +++ b/atlassian/models/jira/adf.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional, Union + + +@dataclass +class TextNode: + text: str + marks: list[dict[str, Any]] = field(default_factory=list) + + def bold(self) -> TextNode: + self.marks.append({"type": "strong"}) + return self + + def italic(self) -> TextNode: + self.marks.append({"type": "em"}) + return self + + def code(self) -> TextNode: + self.marks.append({"type": "code"}) + return self + + def link(self, href: str) -> TextNode: + self.marks.append({"type": "link", "attrs": {"href": href}}) + return self + + def strike(self) -> TextNode: + self.marks.append({"type": "strike"}) + return self + + def to_dict(self) -> dict[str, Any]: + node: dict[str, Any] = {"type": "text", "text": self.text} + if self.marks: + node["marks"] = [dict(m) for m in self.marks] + return node + + +@dataclass(frozen=True) +class MentionNode: + account_id: str + display_text: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "type": "mention", + "attrs": {"id": self.account_id, "text": self.display_text}, + } + + +InlineNode = Union[TextNode, MentionNode] + + +class ADFBuilder: + """Fluent builder that produces an Atlassian Document Format dict.""" + + def __init__(self) -> None: + """Initialize an empty ADF document builder.""" + self._content: list[dict[str, Any]] = [] + + def paragraph(self, *nodes: InlineNode) -> ADFBuilder: + self._content.append( + { + "type": "paragraph", + "content": [n.to_dict() for n in nodes], + } + ) + return self + + def text_paragraph(self, text: str) -> ADFBuilder: + return self.paragraph(TextNode(text)) + + def heading(self, text: str, level: int = 1) -> ADFBuilder: + if level not in range(1, 7): + raise ValueError(f"Heading level must be 1-6, got {level}") + self._content.append( + { + "type": "heading", + "attrs": {"level": level}, + "content": [{"type": "text", "text": text}], + } + ) + return self + + def bullet_list(self, items: list[str]) -> ADFBuilder: + list_items = [ + { + "type": "listItem", + "content": [ + { + "type": "paragraph", + "content": [{"type": "text", "text": item}], + } + ], + } + for item in items + ] + self._content.append({"type": "bulletList", "content": list_items}) + return self + + def ordered_list(self, items: list[str]) -> ADFBuilder: + list_items = [ + { + "type": "listItem", + "content": [ + { + "type": "paragraph", + "content": [{"type": "text", "text": item}], + } + ], + } + for item in items + ] + self._content.append({"type": "orderedList", "content": list_items}) + return self + + def code_block(self, code: str, language: Optional[str] = None) -> ADFBuilder: + node: dict[str, Any] = { + "type": "codeBlock", + "content": [{"type": "text", "text": code}], + } + if language: + node["attrs"] = {"language": language} + self._content.append(node) + return self + + def rule(self) -> ADFBuilder: + self._content.append({"type": "rule"}) + return self + + def blockquote(self, *nodes: InlineNode) -> ADFBuilder: + self._content.append( + { + "type": "blockquote", + "content": [ + { + "type": "paragraph", + "content": [n.to_dict() for n in nodes], + } + ], + } + ) + return self + + def table(self, headers: list[str], rows: list[list[str]]) -> ADFBuilder: + header_row = { + "type": "tableRow", + "content": [ + { + "type": "tableHeader", + "content": [{"type": "paragraph", "content": [{"type": "text", "text": h}]}], + } + for h in headers + ], + } + data_rows = [ + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "content": [{"type": "paragraph", "content": [{"type": "text", "text": cell}]}], + } + for cell in row + ], + } + for row in rows + ] + self._content.append( + { + "type": "table", + "attrs": {"isNumberColumnEnabled": False, "layout": "default"}, + "content": [header_row] + data_rows, + } + ) + return self + + def raw_node(self, node: dict[str, Any]) -> ADFBuilder: + """Escape hatch for ADF node types not covered by dedicated methods.""" + self._content.append(node) + return self + + def build(self) -> dict[str, Any]: + return { + "version": 1, + "type": "doc", + "content": list(self._content), + } diff --git a/atlassian/models/jira/builders.py b/atlassian/models/jira/builders.py new file mode 100644 index 000000000..a17bfe4ac --- /dev/null +++ b/atlassian/models/jira/builders.py @@ -0,0 +1,272 @@ +from __future__ import annotations + +import datetime +from typing import Any, Generic, Optional, Type, TypeVar, Union + +from atlassian.models.jira.adf import ADFBuilder, InlineNode +from atlassian.models.jira.fields import ( + Component, + CustomField, + IssueFields, + IssueLink, + IssueType, + Parent, + Priority, + PriorityLevel, + Project, + User, + Version, +) +from atlassian.models.jira.issues import Bug, Epic, JiraIssue, Story, SubTask, Task +from atlassian.models.jira.serializer import FieldMapping, serialize, to_fields_dict +from atlassian.models.jira.validation import validate_or_raise + +T = TypeVar("T", bound=JiraIssue) + + +class IssueBuilder(Generic[T]): # pylint: disable=too-many-public-methods + """Fluent, type-safe builder for any JiraIssue subclass.""" + + def __init__(self, issue_cls: Type[T]) -> None: + """Initialize the builder for the given issue class.""" + self._issue_cls = issue_cls + self._fields = IssueFields() + + def project(self, key: Optional[str] = None, *, id_: Optional[str] = None) -> IssueBuilder[T]: + self._fields.project = Project(key=key, id=id_) + return self + + def summary(self, text: str) -> IssueBuilder[T]: + self._fields.summary = text + return self + + def description(self, text: str) -> IssueBuilder[T]: + self._fields.description = text + return self + + def description_adf(self, adf: Union[dict[str, Any], ADFBuilder]) -> IssueBuilder[T]: + if isinstance(adf, ADFBuilder): + self._fields.description = adf.build() + else: + self._fields.description = adf + return self + + def description_builder(self) -> _ADFBridge[T]: + return _ADFBridge(self) + + def priority( + self, + name: Optional[str] = None, + *, + id_: Optional[str] = None, + level: Optional[PriorityLevel] = None, + ) -> IssueBuilder[T]: + if level is not None: + self._fields.priority = Priority.from_level(level) + else: + self._fields.priority = Priority(name=name, id=id_) + return self + + def labels(self, *labels: str) -> IssueBuilder[T]: + self._fields.labels = list(labels) + return self + + def add_label(self, label: str) -> IssueBuilder[T]: + self._fields.labels.append(label) + return self + + def components(self, *names: str) -> IssueBuilder[T]: + self._fields.components = [Component(name=n) for n in names] + return self + + def add_component(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> IssueBuilder[T]: + self._fields.components.append(Component(name=name, id=id_)) + return self + + def assignee(self, *, account_id: Optional[str] = None, name: Optional[str] = None) -> IssueBuilder[T]: + self._fields.assignee = User(account_id=account_id, name=name) + return self + + def reporter(self, *, account_id: Optional[str] = None, name: Optional[str] = None) -> IssueBuilder[T]: + self._fields.reporter = User(account_id=account_id, name=name) + return self + + def due_date(self, date: Union[datetime.date, str]) -> IssueBuilder[T]: + if isinstance(date, str): + date = datetime.date.fromisoformat(date) + self._fields.due_date = date + return self + + def fix_versions(self, *names: str) -> IssueBuilder[T]: + self._fields.fix_versions = [Version(name=n) for n in names] + return self + + def add_fix_version(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> IssueBuilder[T]: + self._fields.fix_versions.append(Version(name=name, id=id_)) + return self + + def affected_versions(self, *names: str) -> IssueBuilder[T]: + self._fields.affected_versions = [Version(name=n) for n in names] + return self + + def parent(self, key: Optional[str] = None, *, id_: Optional[str] = None) -> IssueBuilder[T]: + self._fields.parent = Parent(key=key, id=id_) + return self + + def epic_link(self, epic_key: str) -> IssueBuilder[T]: + self._fields.epic_link = epic_key + return self + + def story_points(self, points: float) -> IssueBuilder[T]: + self._fields.story_points = points + return self + + def add_link( + self, + link_type: str, + *, + outward: Optional[str] = None, + inward: Optional[str] = None, + ) -> IssueBuilder[T]: + self._fields.issue_links.append(IssueLink(link_type=link_type, outward_issue=outward, inward_issue=inward)) + return self + + def custom_field(self, field_id: str, value: Any) -> IssueBuilder[T]: + self._fields.custom_fields.append(CustomField(field_id=field_id, value=value)) + return self + + def validate(self) -> IssueBuilder[T]: + """Validate the current fields and raise ValueError if invalid. + + Can be chained: task().project("P").summary("S").validate().build() + """ + issue = self._issue_cls() + issue.fields = self._fields + type_name = self._issue_cls._issue_type_name + if type_name: + issue.fields.issue_type = IssueType(name=type_name) + validate_or_raise(issue) + return self + + def build(self) -> T: + issue = self._issue_cls() + issue.fields = self._fields + type_name = self._issue_cls._issue_type_name + if type_name: + issue.fields.issue_type = IssueType(name=type_name) + return issue + + def build_dict(self, *, mapping: Optional[FieldMapping] = None) -> dict[str, Any]: + """Build the issue and serialize to fields dict for issue_create().""" + return to_fields_dict(self.build(), mapping=mapping) + + def build_payload(self, *, mapping: Optional[FieldMapping] = None) -> dict[str, Any]: + """Build the issue and serialize to full payload for create_issue().""" + return serialize(self.build(), mapping=mapping) + + +class _ADFBridge(Generic[T]): + """Bridges IssueBuilder into ADFBuilder, returning control on done().""" + + def __init__(self, parent: IssueBuilder[T]) -> None: + """Initialize the ADF bridge for the given parent builder.""" + self._parent = parent + self._adf = ADFBuilder() + + def paragraph(self, *nodes: InlineNode) -> _ADFBridge[T]: + self._adf.paragraph(*nodes) + return self + + def text_paragraph(self, text: str) -> _ADFBridge[T]: + self._adf.text_paragraph(text) + return self + + def heading(self, text: str, level: int = 1) -> _ADFBridge[T]: + self._adf.heading(text, level) + return self + + def bullet_list(self, items: list[str]) -> _ADFBridge[T]: + self._adf.bullet_list(items) + return self + + def ordered_list(self, items: list[str]) -> _ADFBridge[T]: + self._adf.ordered_list(items) + return self + + def code_block(self, code: str, language: Optional[str] = None) -> _ADFBridge[T]: + self._adf.code_block(code, language) + return self + + def rule(self) -> _ADFBridge[T]: + self._adf.rule() + return self + + def blockquote(self, *nodes: InlineNode) -> _ADFBridge[T]: + self._adf.blockquote(*nodes) + return self + + def table(self, headers: list[str], rows: list[list[str]]) -> _ADFBridge[T]: + self._adf.table(headers, rows) + return self + + def raw_node(self, node: dict[str, Any]) -> _ADFBridge[T]: + self._adf.raw_node(node) + return self + + def done(self) -> IssueBuilder[T]: + self._parent.description_adf(self._adf) + return self._parent + + +class TaskBuilder(IssueBuilder[Task]): + def __init__(self) -> None: + """Initialize a builder for Task issues.""" + super().__init__(Task) + + +class BugBuilder(IssueBuilder[Bug]): + def __init__(self) -> None: + """Initialize a builder for Bug issues.""" + super().__init__(Bug) + + +class StoryBuilder(IssueBuilder[Story]): + def __init__(self) -> None: + """Initialize a builder for Story issues.""" + super().__init__(Story) + + +class EpicBuilder(IssueBuilder[Epic]): + def __init__(self) -> None: + """Initialize a builder for Epic issues.""" + super().__init__(Epic) + + def epic_name(self, name: str) -> EpicBuilder: + self._fields.epic_name = name + return self + + +class SubTaskBuilder(IssueBuilder[SubTask]): + def __init__(self) -> None: + """Initialize a builder for SubTask issues.""" + super().__init__(SubTask) + + +def task() -> IssueBuilder[Task]: + return IssueBuilder(Task) + + +def bug() -> IssueBuilder[Bug]: + return IssueBuilder(Bug) + + +def story() -> IssueBuilder[Story]: + return IssueBuilder(Story) + + +def epic() -> EpicBuilder: + return EpicBuilder() + + +def subtask() -> SubTaskBuilder: + return SubTaskBuilder() diff --git a/atlassian/models/jira/comment.py b/atlassian/models/jira/comment.py new file mode 100644 index 000000000..e3fe36456 --- /dev/null +++ b/atlassian/models/jira/comment.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional, Union + + +@dataclass(frozen=True) +class Visibility: + """Comment visibility restriction.""" + + type: str + value: str + + def __post_init__(self) -> None: + """Validate visibility type and value.""" + if self.type not in ("role", "group"): + raise ValueError(f"Visibility type must be 'role' or 'group', got '{self.type}'") + if not self.value: + raise ValueError("Visibility requires a 'value'") + + def to_dict(self) -> dict[str, str]: + return {"type": self.type, "value": self.value} + + +@dataclass +class Comment: + """Represents a Jira issue comment. + + Used with Jira.issue_add_comment(issue_key, comment, visibility=...). + Supports both plain text and ADF body. + """ + + body: Union[str, dict[str, Any]] + visibility: Optional[Visibility] = None + + def __post_init__(self) -> None: + """Validate comment body is not empty.""" + if not self.body: + raise ValueError("Comment requires a 'body'") + + def as_args(self) -> dict[str, Any]: + """Return keyword arguments for Jira.issue_add_comment(). + + Usage: + c = Comment("Fixed in PR #42", visibility=Visibility("role", "Developers")) + jira.issue_add_comment("PLAT-123", **c.as_args()) + """ + args: dict[str, Any] = {"comment": self.body} + if self.visibility: + args["visibility"] = self.visibility.to_dict() + return args diff --git a/atlassian/models/jira/fields.py b/atlassian/models/jira/fields.py new file mode 100644 index 000000000..8690a29bf --- /dev/null +++ b/atlassian/models/jira/fields.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import datetime +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, ClassVar, Optional, TypeVar, Union + + +class PriorityLevel(Enum): + HIGHEST = "Highest" + HIGH = "High" + MEDIUM = "Medium" + LOW = "Low" + LOWEST = "Lowest" + + +_NI = TypeVar("_NI", bound="_NameIdEntity") +_KI = TypeVar("_KI", bound="_KeyIdEntity") + + +@dataclass(frozen=True) +class _NameIdEntity: + """Base for frozen entities resolved by name or id.""" + + name: Optional[str] = None + id: Optional[str] = None + + _entity_label: ClassVar[str] = "Entity" + + def __post_init__(self) -> None: + """Validate that at least one identifier is provided.""" + if not self.name and not self.id: + raise ValueError(f"{self._entity_label} requires either 'name' or 'id'") + + def to_dict(self) -> dict[str, Any]: + if self.name: + return {"name": self.name} + return {"id": self.id} + + @classmethod + def from_dict(cls: type[_NI], data: dict[str, Any]) -> _NI: + return cls(name=data.get("name"), id=data.get("id")) + + +@dataclass(frozen=True) +class _KeyIdEntity: + """Base for frozen entities resolved by key or id.""" + + key: Optional[str] = None + id: Optional[str] = None + + _entity_label: ClassVar[str] = "Entity" + + def __post_init__(self) -> None: + """Validate that at least one identifier is provided.""" + if not self.key and not self.id: + raise ValueError(f"{self._entity_label} requires either 'key' or 'id'") + + def to_dict(self) -> dict[str, Any]: + if self.key: + return {"key": self.key} + return {"id": self.id} + + @classmethod + def from_dict(cls: type[_KI], data: dict[str, Any]) -> _KI: + return cls(key=data.get("key"), id=data.get("id")) + + +@dataclass(frozen=True) +class Project(_KeyIdEntity): + _entity_label: ClassVar[str] = "Project" + + +@dataclass(frozen=True) +class IssueType(_NameIdEntity): + _entity_label: ClassVar[str] = "IssueType" + + +@dataclass(frozen=True) +class Priority(_NameIdEntity): + _entity_label: ClassVar[str] = "Priority" + + @classmethod + def from_level(cls, level: PriorityLevel) -> Priority: + return cls(name=level.value) + + +@dataclass(frozen=True) +class User: + account_id: Optional[str] = None + name: Optional[str] = None + + def __post_init__(self) -> None: + """Validate that at least one identifier is provided.""" + if not self.account_id and not self.name: + raise ValueError("User requires either 'account_id' (Cloud) or 'name' (Server)") + + def to_dict(self) -> dict[str, Any]: + if self.account_id: + return {"accountId": self.account_id} + return {"name": self.name} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> User: + return cls(account_id=data.get("accountId"), name=data.get("name")) + + +@dataclass(frozen=True) +class Component(_NameIdEntity): + _entity_label: ClassVar[str] = "Component" + + +@dataclass(frozen=True) +class Version(_NameIdEntity): + _entity_label: ClassVar[str] = "Version" + + +@dataclass(frozen=True) +class Parent(_KeyIdEntity): + _entity_label: ClassVar[str] = "Parent" + + +@dataclass(frozen=True) +class IssueLink: + link_type: str + outward_issue: Optional[str] = None + inward_issue: Optional[str] = None + + def __post_init__(self) -> None: + """Validate that at least one linked issue is provided.""" + if not self.outward_issue and not self.inward_issue: + raise ValueError("IssueLink requires either 'outward_issue' or 'inward_issue'") + + def to_dict(self) -> dict[str, Any]: + entry: dict[str, Any] = {"type": {"name": self.link_type}} + if self.outward_issue: + entry["outwardIssue"] = {"key": self.outward_issue} + if self.inward_issue: + entry["inwardIssue"] = {"key": self.inward_issue} + return entry + + +@dataclass(frozen=True) +class CustomField: + field_id: str + value: Any + + def __post_init__(self) -> None: + """Validate that field_id is not empty.""" + if not self.field_id: + raise ValueError("CustomField requires a 'field_id'") + + +@dataclass +class IssueFields: # pylint: disable=too-many-instance-attributes + project: Optional[Project] = None + issue_type: Optional[IssueType] = None + summary: Optional[str] = None + description: Optional[Union[str, dict[str, Any]]] = None + priority: Optional[Priority] = None + labels: list[str] = field(default_factory=list) + components: list[Component] = field(default_factory=list) + assignee: Optional[User] = None + reporter: Optional[User] = None + parent: Optional[Parent] = None + epic_link: Optional[str] = None + epic_name: Optional[str] = None + fix_versions: list[Version] = field(default_factory=list) + affected_versions: list[Version] = field(default_factory=list) + due_date: Optional[datetime.date] = None + story_points: Optional[float] = None + issue_links: list[IssueLink] = field(default_factory=list) + custom_fields: list[CustomField] = field(default_factory=list) + + @classmethod + def _parse_entity_fields(cls, fields, data, mapping): + if "project" in data and data["project"]: + fields.project = Project.from_dict(data["project"]) + if "issuetype" in data and data["issuetype"]: + fields.issue_type = IssueType.from_dict(data["issuetype"]) + if "priority" in data and data["priority"]: + fields.priority = Priority.from_dict(data["priority"]) + if "assignee" in data and data["assignee"]: + fields.assignee = User.from_dict(data["assignee"]) + if "reporter" in data and data["reporter"]: + fields.reporter = User.from_dict(data["reporter"]) + if "parent" in data and data["parent"]: + fields.parent = Parent.from_dict(data["parent"]) + + @classmethod + def _parse_collection_fields(cls, fields, data): + if "summary" in data: + fields.summary = data["summary"] + if "description" in data: + fields.description = data["description"] + if "labels" in data: + fields.labels = list(data["labels"]) + if "components" in data: + fields.components = [Component.from_dict(c) for c in data["components"]] + if "fixVersions" in data: + fields.fix_versions = [Version.from_dict(v) for v in data["fixVersions"]] + if "versions" in data: + fields.affected_versions = [Version.from_dict(v) for v in data["versions"]] + if "duedate" in data and data["duedate"]: + fields.due_date = datetime.date.fromisoformat(data["duedate"]) + + @classmethod + def _parse_custom_mapped_fields(cls, fields, data, mapping): + if mapping.epic_link_field in data: + fields.epic_link = data[mapping.epic_link_field] + if mapping.epic_name_field in data: + fields.epic_name = data[mapping.epic_name_field] + if mapping.story_points_field in data and data[mapping.story_points_field] is not None: + fields.story_points = data[mapping.story_points_field] + + @classmethod + def from_dict(cls, data: dict[str, Any], *, mapping: Optional[Any] = None) -> IssueFields: + """Parse a Jira REST API fields dict into an IssueFields instance. + + Handles the standard Jira field keys (issuetype, fixVersions, etc.) + and maps them back to Python attribute names. + + Note: this is a partial deserialization. ``custom_fields`` and + ``issue_links`` are not reconstructed (their schema varies per + instance). Use ``serialize()`` output for the authoritative format. + """ + from atlassian.models.jira.serializer import FieldMapping # pylint: disable=import-outside-toplevel + + if mapping is None: + mapping = FieldMapping() + + fields = cls() + cls._parse_entity_fields(fields, data, mapping) + cls._parse_collection_fields(fields, data) + cls._parse_custom_mapped_fields(fields, data, mapping) + return fields diff --git a/atlassian/models/jira/issues.py b/atlassian/models/jira/issues.py new file mode 100644 index 000000000..9efdf7d3d --- /dev/null +++ b/atlassian/models/jira/issues.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, ClassVar + +from atlassian.models.jira.fields import IssueFields, IssueType + + +_ISSUE_TYPE_REGISTRY: dict[str, type[JiraIssue]] = {} + + +def get_issue_type_registry() -> dict[str, type[JiraIssue]]: + return dict(_ISSUE_TYPE_REGISTRY) + + +def issue_type_for(name: str) -> type[JiraIssue]: + try: + return _ISSUE_TYPE_REGISTRY[name] + except KeyError: + raise ValueError(f"Unknown issue type '{name}'. " f"Registered: {sorted(_ISSUE_TYPE_REGISTRY)}") + + +@dataclass +class JiraIssue: + _issue_type_name: ClassVar[str] = "" + + fields: IssueFields = field(default_factory=IssueFields) + + def __init_subclass__(cls, **kwargs: Any) -> None: + """Register subclass in the issue type registry.""" + super().__init_subclass__(**kwargs) + name = getattr(cls, "_issue_type_name", "") + if name: + _ISSUE_TYPE_REGISTRY[name] = cls + + def __post_init__(self) -> None: + """Stamp issue type from class variable.""" + if self._issue_type_name: + self.fields.issue_type = IssueType(name=self._issue_type_name) + + @classmethod + def from_dict(cls, data: dict[str, Any], *, mapping: Any = None) -> JiraIssue: + """Create a JiraIssue from a Jira REST API response dict. + + If a ``fields`` key is present, extracts fields from it. + Otherwise treats the dict itself as the fields block. + + Note: this is a *partial* deserialization. ``custom_fields`` and + ``issue_links`` are not reconstructed because their schema varies + per Jira instance. Use the returned object for inspection or as a + starting point for updates, not as a lossless round-trip. + """ + fields_data = data.get("fields", data) + fields = IssueFields.from_dict(fields_data, mapping=mapping) + + issue_type_name = "" + if fields.issue_type and fields.issue_type.name: + issue_type_name = fields.issue_type.name + + issue_cls = _ISSUE_TYPE_REGISTRY.get(issue_type_name, cls) + issue = issue_cls.__new__(issue_cls) + issue.fields = fields + return issue + + def __repr__(self) -> str: + """Return a concise string representation.""" + parts = [self.__class__.__name__] + if self.fields.project and self.fields.project.key: + parts.append(f"project={self.fields.project.key!r}") + if self.fields.summary: + summary = self.fields.summary + if len(summary) > 50: + summary = summary[:47] + "..." + parts.append(f"summary={summary!r}") + if len(parts) == 1: + return f"{parts[0]}()" + return f"{parts[0]}({', '.join(parts[1:])})" + + +@dataclass +class Task(JiraIssue): + _issue_type_name: ClassVar[str] = "Task" + + +@dataclass +class Bug(JiraIssue): + _issue_type_name: ClassVar[str] = "Bug" + + +@dataclass +class Story(JiraIssue): + _issue_type_name: ClassVar[str] = "Story" + + +@dataclass +class Epic(JiraIssue): + _issue_type_name: ClassVar[str] = "Epic" + + +@dataclass +class SubTask(JiraIssue): + _issue_type_name: ClassVar[str] = "Sub-task" diff --git a/atlassian/models/jira/serializer.py b/atlassian/models/jira/serializer.py new file mode 100644 index 000000000..8cb62893d --- /dev/null +++ b/atlassian/models/jira/serializer.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional + +from atlassian.models.jira.fields import IssueFields, IssueLink +from atlassian.models.jira.issues import JiraIssue + + +@dataclass +class FieldMapping: + """Maps well-known model fields to instance-specific Jira custom field IDs. + + Different Jira instances use different custom field IDs for concepts like + epic link or story points. Override the defaults here. + """ + + epic_link_field: str = "customfield_10014" + story_points_field: str = "customfield_10028" + epic_name_field: str = "customfield_10011" + + +def _ser_entity_fields(f: IssueFields, fields: dict[str, Any]) -> None: + if f.project: + fields["project"] = f.project.to_dict() + if f.issue_type: + fields["issuetype"] = f.issue_type.to_dict() + if f.priority: + fields["priority"] = f.priority.to_dict() + if f.assignee: + fields["assignee"] = f.assignee.to_dict() + if f.reporter: + fields["reporter"] = f.reporter.to_dict() + if f.parent: + fields["parent"] = f.parent.to_dict() + + +def _ser_scalar_fields(f: IssueFields, fields: dict[str, Any]) -> None: + if f.summary is not None: + fields["summary"] = f.summary + if f.description is not None: + fields["description"] = f.description + if f.due_date: + fields["duedate"] = f.due_date.isoformat() + + +def _ser_collection_fields(f: IssueFields, fields: dict[str, Any]) -> None: + if f.labels: + fields["labels"] = list(f.labels) + if f.components: + fields["components"] = [c.to_dict() for c in f.components] + if f.fix_versions: + fields["fixVersions"] = [v.to_dict() for v in f.fix_versions] + if f.affected_versions: + fields["versions"] = [v.to_dict() for v in f.affected_versions] + + +def _ser_custom_and_mapped_fields(f: IssueFields, fields: dict[str, Any], mapping: FieldMapping) -> None: + if f.epic_link: + fields[mapping.epic_link_field] = f.epic_link + if f.epic_name: + fields[mapping.epic_name_field] = f.epic_name + if f.story_points is not None: + fields[mapping.story_points_field] = f.story_points + for cf in f.custom_fields: + fields[cf.field_id] = cf.value + + +def _ser_issue_links(links: list[IssueLink]) -> list[dict[str, Any]]: + """Produce the update payload for issue links. + + Issue links go into the `update.issuelinks` block as "add" operations, + not into the top-level `fields` dict. + """ + result: list[dict[str, Any]] = [] + for link in links: + entry: dict[str, Any] = {"add": {"type": {"name": link.link_type}}} + if link.outward_issue: + entry["add"]["outwardIssue"] = {"key": link.outward_issue} + if link.inward_issue: + entry["add"]["inwardIssue"] = {"key": link.inward_issue} + result.append(entry) + return result + + +def serialize( + issue: JiraIssue, + *, + mapping: Optional[FieldMapping] = None, +) -> dict[str, Any]: + """Convert a JiraIssue into the dict that Jira.create_issue() expects. + + Returns a dict with top-level keys "fields" and optionally "update". + """ + if mapping is None: + mapping = FieldMapping() + + f = issue.fields + fields = {} + update = {} + + _ser_entity_fields(f, fields) + _ser_scalar_fields(f, fields) + _ser_collection_fields(f, fields) + _ser_custom_and_mapped_fields(f, fields, mapping) + + if f.issue_links: + update["issuelinks"] = _ser_issue_links(f.issue_links) + + result = {"fields": fields} + if update: + result["update"] = update + return result + + +def to_fields_dict( + issue: JiraIssue, + *, + mapping: Optional[FieldMapping] = None, +) -> dict[str, Any]: + """Return only the inner fields dict for jira.issue_create(fields=...). + + Use with jira.issue_create(fields=to_fields_dict(issue)). + """ + return serialize(issue, mapping=mapping)["fields"] + + +def bulk_serialize( + issues: list[JiraIssue], + *, + mapping: Optional[FieldMapping] = None, +) -> list[dict[str, Any]]: + """Serialize a list of issues for Jira.create_issues() bulk endpoint. + + Returns a list of dicts, each with "fields" and optionally "update" keys, + matching the format expected by POST /rest/api/2/issue/bulk. + """ + return [serialize(issue, mapping=mapping) for issue in issues] diff --git a/atlassian/models/jira/transition.py b/atlassian/models/jira/transition.py new file mode 100644 index 000000000..b68631c54 --- /dev/null +++ b/atlassian/models/jira/transition.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional + + +@dataclass +class Transition: + """Represents a Jira issue status transition. + + Used with Jira.set_issue_status(issue_key, status_name, fields=..., update=...). + """ + + issue_key: str + status: str + fields: dict[str, Any] = field(default_factory=dict) + update: dict[str, Any] = field(default_factory=dict) + resolution: Optional[str] = None + + def __post_init__(self) -> None: + """Validate required fields and apply resolution.""" + if not self.issue_key: + raise ValueError("Transition requires an 'issue_key'") + if not self.status: + raise ValueError("Transition requires a 'status'") + if self.resolution: + self.fields["resolution"] = {"name": self.resolution} + + def as_args(self) -> dict[str, Any]: + """Return keyword arguments for Jira.set_issue_status(). + + Usage: + t = Transition("PLAT-123", "Done", resolution="Fixed") + jira.set_issue_status(**t.as_args()) + """ + args: dict[str, Any] = { + "issue_key": self.issue_key, + "status_name": self.status, + } + if self.fields: + args["fields"] = self.fields + if self.update: + args["update"] = self.update + return args + + +class TransitionBuilder: + """Fluent builder for issue transitions.""" + + def __init__(self, issue_key: str, status: str) -> None: + """Initialize the builder with issue key and target status.""" + self._issue_key = issue_key + self._status = status + self._fields: dict[str, Any] = {} + self._update: dict[str, Any] = {} + self._resolution: Optional[str] = None + + def resolution(self, name: str) -> TransitionBuilder: + self._resolution = name + return self + + def set_field(self, field_name: str, value: Any) -> TransitionBuilder: + self._fields[field_name] = value + return self + + def set_custom_field(self, field_id: str, value: Any) -> TransitionBuilder: + self._fields[field_id] = value + return self + + def build(self) -> Transition: + return Transition( + issue_key=self._issue_key, + status=self._status, + fields=dict(self._fields), + update=dict(self._update), + resolution=self._resolution, + ) diff --git a/atlassian/models/jira/update.py b/atlassian/models/jira/update.py new file mode 100644 index 000000000..7723eb0ad --- /dev/null +++ b/atlassian/models/jira/update.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional, Type, Union + +from atlassian.models.jira.fields import ( + Component, + Priority, + PriorityLevel, + User, + Version, + _NameIdEntity, +) + + +@dataclass +class UpdatePayload: + issue_key: str + fields: dict[str, Any] = field(default_factory=dict) + update: dict[str, Any] = field(default_factory=dict) + + +class UpdateBuilder: # pylint: disable=too-many-public-methods + """Fluent builder for Jira issue update payloads. + + Produces the dict format expected by Jira.issue_update() and + Jira.update_issue_field(). + + Example: + payload = ( + UpdateBuilder("PLAT-123") + .set_summary("New title") + .set_priority("Critical") + .add_labels("hotfix") + .remove_label("stale") + .add_comment("Fixed in PR #42") + .build() + ) + jira.issue_update(payload.issue_key, payload.fields, update=payload.update) + + """ + + def __init__(self, issue_key: str) -> None: + """Initialize the builder for the given issue key.""" + self._issue_key = issue_key + self._fields: dict[str, Any] = {} + self._update: dict[str, list[dict[str, Any]]] = {} + + def _add_op(self, field_name: str, operation: str, value: Any) -> UpdateBuilder: + self._update.setdefault(field_name, []).append({operation: value}) + return self + + def _entity_op(self, field_name: str, operation: str, entity: _NameIdEntity) -> UpdateBuilder: + return self._add_op(field_name, operation, entity.to_dict()) + + def _set_entity_list( + self, field_name: str, entity_cls: Type[_NameIdEntity], names: tuple[str, ...] + ) -> UpdateBuilder: + self._fields[field_name] = [entity_cls(name=n).to_dict() for n in names] + return self + + def set_summary(self, text: str) -> UpdateBuilder: + self._fields["summary"] = text + return self + + def set_description(self, text: Union[str, dict[str, Any]]) -> UpdateBuilder: + self._fields["description"] = text + return self + + def set_priority( + self, + name: Optional[str] = None, + *, + id_: Optional[str] = None, + level: Optional[PriorityLevel] = None, + ) -> UpdateBuilder: + if level is not None: + p = Priority.from_level(level) + else: + p = Priority(name=name, id=id_) + self._fields["priority"] = p.to_dict() + return self + + def set_assignee(self, *, account_id: Optional[str] = None, name: Optional[str] = None) -> UpdateBuilder: + self._fields["assignee"] = User(account_id=account_id, name=name).to_dict() + return self + + def unassign(self) -> UpdateBuilder: + self._fields["assignee"] = None + return self + + def set_reporter(self, *, account_id: Optional[str] = None, name: Optional[str] = None) -> UpdateBuilder: + self._fields["reporter"] = User(account_id=account_id, name=name).to_dict() + return self + + def set_labels(self, *labels: str) -> UpdateBuilder: + self._fields["labels"] = list(labels) + return self + + def add_labels(self, *labels: str) -> UpdateBuilder: + for label in labels: + self._add_op("labels", "add", label) + return self + + def remove_label(self, label: str) -> UpdateBuilder: + return self._add_op("labels", "remove", label) + + def set_components(self, *names: str) -> UpdateBuilder: + return self._set_entity_list("components", Component, names) + + def add_component(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("components", "add", Component(name=name, id=id_)) + + def remove_component(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("components", "remove", Component(name=name, id=id_)) + + def set_fix_versions(self, *names: str) -> UpdateBuilder: + return self._set_entity_list("fixVersions", Version, names) + + def add_fix_version(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("fixVersions", "add", Version(name=name, id=id_)) + + def remove_fix_version(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("fixVersions", "remove", Version(name=name, id=id_)) + + def set_affected_versions(self, *names: str) -> UpdateBuilder: + return self._set_entity_list("versions", Version, names) + + def add_affected_version(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("versions", "add", Version(name=name, id=id_)) + + def remove_affected_version(self, name: Optional[str] = None, *, id_: Optional[str] = None) -> UpdateBuilder: + return self._entity_op("versions", "remove", Version(name=name, id=id_)) + + def set_due_date(self, date: str) -> UpdateBuilder: + self._fields["duedate"] = date + return self + + def clear_due_date(self) -> UpdateBuilder: + self._fields["duedate"] = None + return self + + def set_custom_field(self, field_id: str, value: Any) -> UpdateBuilder: + self._fields[field_id] = value + return self + + def add_comment(self, body: Union[str, dict[str, Any]]) -> UpdateBuilder: + return self._add_op("comment", "add", {"body": body}) + + def add_issue_link( + self, + link_type: str, + *, + outward: Optional[str] = None, + inward: Optional[str] = None, + ) -> UpdateBuilder: + link: dict[str, Any] = {"type": {"name": link_type}} + if outward: + link["outwardIssue"] = {"key": outward} + if inward: + link["inwardIssue"] = {"key": inward} + return self._add_op("issuelinks", "add", link) + + def build(self) -> UpdatePayload: + return UpdatePayload( + issue_key=self._issue_key, + fields=dict(self._fields), + update=dict(self._update), + ) + + def build_dict(self) -> dict[str, Any]: + """Return the raw dict payload for direct use with issue_update().""" + result: dict[str, Any] = {} + if self._fields: + result["fields"] = dict(self._fields) + if self._update: + result["update"] = dict(self._update) + return result diff --git a/atlassian/models/jira/validation.py b/atlassian/models/jira/validation.py new file mode 100644 index 000000000..49661b609 --- /dev/null +++ b/atlassian/models/jira/validation.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import datetime +from dataclasses import dataclass + +from atlassian.models.jira.issues import JiraIssue, SubTask + + +@dataclass +class ValidationError: + field_name: str + message: str + + +def validate(issue: JiraIssue) -> list[ValidationError]: + """Validate an issue before serialization. Returns empty list if valid.""" + errors: list[ValidationError] = [] + f = issue.fields + + if not f.project: + errors.append(ValidationError("project", "Project is required")) + if not f.summary: + errors.append(ValidationError("summary", "Summary is required")) + if not f.issue_type: + errors.append(ValidationError("issuetype", "Issue type is required")) + + if isinstance(issue, SubTask) and not f.parent: + errors.append(ValidationError("parent", "Sub-task requires a parent issue")) + + if f.summary and len(f.summary) > 255: + errors.append(ValidationError("summary", "Summary must be 255 characters or fewer")) + + if f.story_points is not None and f.story_points < 0: + errors.append(ValidationError("story_points", "Story points cannot be negative")) + + if f.due_date is not None and not isinstance(f.due_date, datetime.date): + errors.append(ValidationError("duedate", "due_date must be a datetime.date")) + + if f.description and isinstance(f.description, dict): + if f.description.get("type") != "doc" or f.description.get("version") != 1: + errors.append( + ValidationError( + "description", + "ADF description must have type='doc' and version=1", + ) + ) + + return errors + + +def validate_or_raise(issue: JiraIssue) -> None: + """Validate and raise ValueError if any errors found.""" + errors = validate(issue) + if errors: + details = "; ".join(f"{e.field_name}: {e.message}" for e in errors) + raise ValueError(f"Issue validation failed: {details}") diff --git a/atlassian/models/py.typed b/atlassian/models/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_jira_models.py b/tests/test_jira_models.py new file mode 100644 index 000000000..d57fd78bf --- /dev/null +++ b/tests/test_jira_models.py @@ -0,0 +1,745 @@ +from __future__ import annotations + +import datetime +import re + +import pytest + +from atlassian.models.jira.adf import ADFBuilder, MentionNode, TextNode +from atlassian.models.jira.builders import ( + EpicBuilder, + IssueBuilder, + SubTaskBuilder, + bug, + epic, + story, + subtask, + task, +) +from atlassian.models.jira.comment import Comment, Visibility +from atlassian.models.jira.fields import ( + Component, + CustomField, + IssueFields, + IssueLink, + IssueType, + Parent, + Priority, + PriorityLevel, + Project, + User, + Version, +) +from atlassian.models.jira.issues import ( + Bug, + Epic, + JiraIssue, + Story, + SubTask, + Task, + get_issue_type_registry, + issue_type_for, +) +from atlassian.models.jira.serializer import FieldMapping, bulk_serialize, serialize, to_fields_dict +from atlassian.models.jira.transition import Transition, TransitionBuilder +from atlassian.models.jira.update import UpdateBuilder, UpdatePayload +from atlassian.models.jira.validation import validate, validate_or_raise + + +def test_project_key_only_to_dict(): + p = Project(key="ABC") + assert p.to_dict() == {"key": "ABC"} + + +def test_project_id_only_to_dict(): + p = Project(id="10000") + assert p.to_dict() == {"id": "10000"} + + +def test_project_both_prefers_key_in_to_dict(): + p = Project(key="ABC", id="10000") + assert p.to_dict() == {"key": "ABC"} + + +def test_project_neither_raises(): + with pytest.raises(ValueError, match="key' or 'id"): + Project() + + +def test_issue_type_name_only(): + it = IssueType(name="Bug") + assert it.to_dict() == {"name": "Bug"} + + +def test_issue_type_id_only(): + it = IssueType(id="10001") + assert it.to_dict() == {"id": "10001"} + + +def test_issue_type_neither_raises(): + with pytest.raises(ValueError, match="name' or 'id"): + IssueType() + + +def test_priority_creation_and_to_dict(): + assert Priority(name="High").to_dict() == {"name": "High"} + assert Priority(id="2").to_dict() == {"id": "2"} + + +def test_priority_from_level(): + p = Priority.from_level(PriorityLevel.MEDIUM) + assert p.name == "Medium" + assert p.to_dict() == {"name": "Medium"} + + +def test_priority_neither_raises(): + with pytest.raises(ValueError, match="name' or 'id"): + Priority() + + +def test_issue_type_both_name_and_id_prefers_name_in_to_dict(): + assert IssueType(name="Task", id="3").to_dict() == {"name": "Task"} + + +def test_user_cloud_and_server_to_dict(): + assert User(account_id="acc-1").to_dict() == {"accountId": "acc-1"} + assert User(name="jdoe").to_dict() == {"name": "jdoe"} + + +def test_user_neither_raises(): + with pytest.raises(ValueError, match="account_id"): + User() + + +def test_component_version_parent_to_dict(): + assert Component(name="UI").to_dict() == {"name": "UI"} + assert Version(name="1.0").to_dict() == {"name": "1.0"} + assert Parent(key="FOO-1").to_dict() == {"key": "FOO-1"} + + +def test_component_version_parent_neither_raises(): + with pytest.raises(ValueError, match="name' or 'id"): + Component() + with pytest.raises(ValueError, match="name' or 'id"): + Version() + with pytest.raises(ValueError, match="key' or 'id"): + Parent() + + +def test_issue_link_outward_to_dict(): + link = IssueLink("Blocks", outward_issue="A-1") + assert link.to_dict() == { + "type": {"name": "Blocks"}, + "outwardIssue": {"key": "A-1"}, + } + + +def test_issue_link_inward_to_dict(): + link = IssueLink("Duplicate", inward_issue="B-2") + assert link.to_dict() == { + "type": {"name": "Duplicate"}, + "inwardIssue": {"key": "B-2"}, + } + + +def test_issue_link_both_to_dict(): + link = IssueLink("Relates", outward_issue="A-1", inward_issue="B-2") + d = link.to_dict() + assert d["outwardIssue"] == {"key": "A-1"} + assert d["inwardIssue"] == {"key": "B-2"} + + +def test_issue_link_neither_raises(): + with pytest.raises(ValueError, match="outward_issue|inward_issue"): + IssueLink("Blocks") + + +def test_custom_field_empty_id_raises(): + with pytest.raises(ValueError, match="field_id"): + CustomField(field_id="", value=1) + + +def test_adf_empty_doc_structure(): + doc = ADFBuilder().build() + assert doc == {"version": 1, "type": "doc", "content": []} + + +def test_adf_paragraph_and_text_paragraph(): + b = ADFBuilder() + b.paragraph(TextNode("a"), TextNode("b")) + b.text_paragraph("c") + doc = b.build() + assert doc["content"][0] == { + "type": "paragraph", + "content": [ + {"type": "text", "text": "a"}, + {"type": "text", "text": "b"}, + ], + } + assert doc["content"][1] == { + "type": "paragraph", + "content": [{"type": "text", "text": "c"}], + } + + +def test_adf_heading_levels(): + doc = ADFBuilder().heading("T", level=3).build() + assert doc["content"][0] == { + "type": "heading", + "attrs": {"level": 3}, + "content": [{"type": "text", "text": "T"}], + } + + +def test_adf_heading_invalid_level(): + with pytest.raises(ValueError, match="Heading level"): + ADFBuilder().heading("x", level=0) + + +def test_adf_bullet_and_ordered_list(): + doc = ADFBuilder().bullet_list(["a", "b"]).ordered_list(["c"]).build() + assert doc["content"][0]["type"] == "bulletList" + assert doc["content"][1]["type"] == "orderedList" + assert len(doc["content"][0]["content"]) == 2 + + +def test_adf_code_block_with_and_without_language(): + d1 = ADFBuilder().code_block("x = 1").build()["content"][0] + assert d1["type"] == "codeBlock" + assert "attrs" not in d1 + d2 = ADFBuilder().code_block("print()", language="python").build()["content"][0] + assert d2["attrs"] == {"language": "python"} + + +def test_adf_rule(): + assert ADFBuilder().rule().build()["content"][0] == {"type": "rule"} + + +def test_adf_text_marks(): + node = TextNode("hi").bold().italic().code().link("https://e.example").strike() + assert node.to_dict() == { + "type": "text", + "text": "hi", + "marks": [ + {"type": "strong"}, + {"type": "em"}, + {"type": "code"}, + {"type": "link", "attrs": {"href": "https://e.example"}}, + {"type": "strike"}, + ], + } + + +def test_adf_mention_node(): + assert MentionNode("acc", "Bob").to_dict() == { + "type": "mention", + "attrs": {"id": "acc", "text": "Bob"}, + } + assert MentionNode("acc").to_dict()["attrs"]["text"] == "" + + +def test_adf_table(): + doc = ADFBuilder().table(["H"], [["c"]]).build() + tbl = doc["content"][0] + assert tbl["type"] == "table" + assert tbl["attrs"] == {"isNumberColumnEnabled": False, "layout": "default"} + assert len(tbl["content"]) == 2 + + +def test_adf_blockquote(): + doc = ADFBuilder().blockquote(TextNode("q")).build() + bq = doc["content"][0] + assert bq["type"] == "blockquote" + assert bq["content"][0]["type"] == "paragraph" + + +def test_adf_raw_node_escape_hatch(): + raw = {"type": "extension", "attrs": {"foo": "bar"}} + doc = ADFBuilder().raw_node(raw).build() + assert doc["content"][0] == raw + + +def test_issue_classes_and_issue_type_name(): + assert Task._issue_type_name == "Task" + assert Bug._issue_type_name == "Bug" + assert Story._issue_type_name == "Story" + assert Epic._issue_type_name == "Epic" + assert SubTask._issue_type_name == "Sub-task" + + +def test_post_init_stamps_issue_type(): + t = Task() + assert t.fields.issue_type == IssueType(name="Task") + + +def test_registry_has_five_types(): + reg = get_issue_type_registry() + assert set(reg.keys()) == {"Task", "Bug", "Story", "Epic", "Sub-task"} + assert reg["Task"] is Task + + +def test_issue_type_for_known_and_unknown(): + assert issue_type_for("Story") is Story + with pytest.raises(ValueError, match="Unknown issue type"): + issue_type_for("Unknown") + + +def test_serialize_minimal_fields_shape(): + issue = Task() + issue.fields = IssueFields( + project=Project(key="P"), + summary="S", + issue_type=IssueType(name="Task"), + ) + assert serialize(issue) == { + "fields": { + "project": {"key": "P"}, + "issuetype": {"name": "Task"}, + "summary": "S", + } + } + + +def test_serialize_issue_links_update_block(): + issue = Task() + issue.fields = IssueFields( + project=Project(key="P"), + summary="S", + issue_type=IssueType(name="Task"), + issue_links=[IssueLink("Blocks", outward_issue="X-1")], + ) + out = serialize(issue) + assert out["update"] == { + "issuelinks": [ + { + "add": { + "type": {"name": "Blocks"}, + "outwardIssue": {"key": "X-1"}, + }, + }, + ], + } + + +def test_to_fields_dict_strips_wrapper(): + issue = Task() + issue.fields.project = Project(key="K") + issue.fields.summary = "Hi" + d = to_fields_dict(issue) + assert d == serialize(issue)["fields"] + assert "update" not in to_fields_dict(issue) + + +def test_field_mapping_overrides_epic_link_and_story_points_keys(): + issue = Task() + issue.fields.project = Project(key="P") + issue.fields.summary = "S" + issue.fields.epic_link = "E-1" + issue.fields.story_points = 3.5 + m = FieldMapping(epic_link_field="customfield_9001", story_points_field="customfield_9002") + fields = serialize(issue, mapping=m)["fields"] + assert fields["customfield_9001"] == "E-1" + assert fields["customfield_9002"] == 3.5 + + +def test_serialize_all_field_types_and_custom_fields(): + due = datetime.date(2026, 4, 1) + issue = Task() + issue.fields = IssueFields( + project=Project(key="PR"), + issue_type=IssueType(name="Task"), + summary="Sum", + description="plain", + priority=Priority.from_level(PriorityLevel.HIGH), + labels=["l1"], + components=[Component(name="C1")], + assignee=User(account_id="a1"), + reporter=User(name="rep"), + parent=Parent(key="P-9"), + fix_versions=[Version(name="2.0")], + affected_versions=[Version(name="1.0")], + due_date=due, + story_points=2.0, + custom_fields=[CustomField("customfield_50000", {"x": 1})], + ) + f = serialize(issue)["fields"] + assert f["project"] == {"key": "PR"} + assert f["issuetype"] == {"name": "Task"} + assert f["summary"] == "Sum" + assert f["description"] == "plain" + assert f["priority"] == {"name": "High"} + assert f["labels"] == ["l1"] + assert f["components"] == [{"name": "C1"}] + assert f["assignee"] == {"accountId": "a1"} + assert f["reporter"] == {"name": "rep"} + assert f["parent"] == {"key": "P-9"} + assert f["fixVersions"] == [{"name": "2.0"}] + assert f["versions"] == [{"name": "1.0"}] + assert f["duedate"] == "2026-04-01" + assert f["customfield_50000"] == {"x": 1} + + +def test_serialize_omits_empty_optional_lists_and_none_fields(): + issue = Task() + issue.fields.project = Project(key="P") + issue.fields.summary = "S" + keys = set(serialize(issue)["fields"].keys()) + assert "labels" not in keys + assert "components" not in keys + assert "assignee" not in keys + assert "duedate" not in keys + + +def test_issue_builder_chaining_and_build_types(): + b: IssueBuilder[Task] = task() + b.project(key="X").summary("Y").priority(level=PriorityLevel.LOW).labels("a").due_date("2026-01-02") + issue = b.build() + assert isinstance(issue, Task) + assert issue.fields.project == Project(key="X") + assert issue.fields.summary == "Y" + assert issue.fields.priority == Priority.from_level(PriorityLevel.LOW) + assert issue.fields.labels == ["a"] + assert issue.fields.due_date == datetime.date(2026, 1, 2) + + +def test_factory_functions_return_expected_builder_types(): + assert isinstance(task(), IssueBuilder) + assert isinstance(bug(), IssueBuilder) + assert isinstance(story(), IssueBuilder) + assert isinstance(epic(), EpicBuilder) + assert isinstance(subtask(), SubTaskBuilder) + + +def test_build_dict_and_build_payload(): + b = story().project(key="S").summary("st") + assert isinstance(b.build(), Story) + d = b.build_dict() + assert "fields" not in d + assert d["summary"] == "st" + payload = b.build_payload() + assert set(payload.keys()) == {"fields"} + assert payload["fields"]["summary"] == "st" + + +def test_adf_bridge_description_builder_done(): + issue = bug().project(key="P").summary("S").description_builder().text_paragraph("Hello").done().build() + desc = issue.fields.description + assert isinstance(desc, dict) + assert desc["type"] == "doc" + assert desc["version"] == 1 + assert desc["content"][0]["type"] == "paragraph" + + +def test_epic_builder_epic_name_sets_field(): + issue = epic().project(key="E").summary("Epic").epic_name("My Epic").build() + assert issue.fields.epic_name == "My Epic" + fields = to_fields_dict(issue) + assert fields["customfield_10011"] == "My Epic" + + +def test_epic_builder_epic_name_respects_field_mapping(): + issue = epic().project(key="E").summary("Epic").epic_name("My Epic").build() + mapping = FieldMapping(epic_name_field="customfield_99999") + fields = to_fields_dict(issue, mapping=mapping) + assert fields["customfield_99999"] == "My Epic" + assert "customfield_10011" not in fields + + +def test_subtask_builder_parent_method(): + issue = subtask().project(key="P").summary("Sub").parent(key="P-1").build() + assert issue.fields.parent == Parent(key="P-1") + + +def test_builder_custom_field_in_serialize(): + payload = task().project(key="P").summary("S").custom_field("customfield_70000", [1, 2]).build_payload() + assert payload["fields"]["customfield_70000"] == [1, 2] + + +def test_validate_empty_for_complete_task(): + issue = task().project(key="P").summary("OK").build() + assert validate(issue) == [] + + +def test_validate_missing_project_summary_issue_type(): + issue = JiraIssue() + issue.fields = IssueFields() + errs = validate(issue) + names = {e.field_name for e in errs} + assert names == {"project", "summary", "issuetype"} + + +def test_validate_subtask_without_parent(): + issue = subtask().project(key="P").summary("S").build() + names = [e.field_name for e in validate(issue)] + assert "parent" in names + + +def test_validate_summary_too_long(): + issue = task().project(key="P").summary("x" * 256).build() + msgs = [e.message for e in validate(issue) if e.field_name == "summary"] + assert any("255" in m for m in msgs) + + +def test_validate_negative_story_points(): + issue = task().project(key="P").summary("S").story_points(-1).build() + assert any(e.field_name == "story_points" for e in validate(issue)) + + +def test_validate_invalid_adf_structure(): + issue = task().project(key="P").summary("S").build() + issue.fields.description = {"type": "wrong", "version": 1} + assert any(e.field_name == "description" for e in validate(issue)) + + +def test_validate_or_raise_joins_errors(): + issue = JiraIssue() + issue.fields = IssueFields() + with pytest.raises(ValueError, match=re.compile("project:|summary:|issuetype:", re.DOTALL)): + validate_or_raise(issue) + + +def test_e2e_task_builder_serialize_matches_api_shape(): + payload = ( + task() + .project(key="DEMO") + .summary("Do work") + .priority(level=PriorityLevel.MEDIUM) + .assignee(account_id="712345:abc") + .due_date("2026-12-31") + .build_payload() + ) + assert payload == { + "fields": { + "project": {"key": "DEMO"}, + "issuetype": {"name": "Task"}, + "summary": "Do work", + "priority": {"name": "Medium"}, + "assignee": {"accountId": "712345:abc"}, + "duedate": "2026-12-31", + }, + } + + +def test_e2e_bug_with_adf_description_in_fields(): + adf = ADFBuilder().heading("Title").text_paragraph("Body").build() + fields = bug().project(key="B").summary("Crash").description_adf(adf).build_dict() + assert fields["description"] == adf + assert fields["description"]["type"] == "doc" + assert fields["description"]["version"] == 1 + + +def test_e2e_epic_custom_field_mapping(): + mapping = FieldMapping( + epic_link_field="customfield_80001", + story_points_field="customfield_80002", + ) + fields = ( + epic() + .project(key="E") + .summary("Roadmap") + .epic_link("E-100") + .story_points(8) + .custom_field("customfield_91000", "extra") + .build_dict(mapping=mapping) + ) + assert fields["customfield_80001"] == "E-100" + assert fields["customfield_80002"] == 8 + assert fields["customfield_91000"] == "extra" + assert fields["issuetype"] == {"name": "Epic"} + + +def test_update_builder_set_summary(): + p = UpdateBuilder("PLAT-1").set_summary("New title").build() + assert isinstance(p, UpdatePayload) + assert p.issue_key == "PLAT-1" + assert p.fields["summary"] == "New title" + + +def test_update_builder_set_priority(): + p = UpdateBuilder("PLAT-2").set_priority("Critical").build() + assert p.fields["priority"] == {"name": "Critical"} + + +def test_update_builder_add_and_remove_labels(): + p = UpdateBuilder("PLAT-3").add_labels("a", "b").remove_label("stale").build() + assert p.update["labels"] == [ + {"add": "a"}, + {"add": "b"}, + {"remove": "stale"}, + ] + + +def test_update_builder_set_assignee_and_unassign(): + assigned = UpdateBuilder("PLAT-4").set_assignee(account_id="acc-9").build() + assert assigned.fields["assignee"] == {"accountId": "acc-9"} + unassigned = UpdateBuilder("PLAT-4").unassign().build() + assert unassigned.fields["assignee"] is None + + +def test_update_builder_add_component(): + p = UpdateBuilder("PLAT-5").add_component("UI").build() + assert p.update["components"] == [{"add": {"name": "UI"}}] + + +def test_update_builder_add_comment(): + p = UpdateBuilder("PLAT-6").add_comment("note").build() + assert p.update["comment"] == [{"add": {"body": "note"}}] + + +def test_update_builder_add_issue_link(): + p = UpdateBuilder("PLAT-7").add_issue_link("Blocks", outward="OTHER-1").build() + assert p.update["issuelinks"] == [ + { + "add": { + "type": {"name": "Blocks"}, + "outwardIssue": {"key": "OTHER-1"}, + }, + }, + ] + + +def test_update_builder_build_dict_format(): + d = UpdateBuilder("PLAT-8").set_summary("S").add_labels("x").build_dict() + assert d == { + "fields": {"summary": "S"}, + "update": {"labels": [{"add": "x"}]}, + } + + +def test_transition_basic(): + t = Transition("FOO-1", "In Progress") + assert t.issue_key == "FOO-1" + assert t.status == "In Progress" + assert t.fields == {} + assert t.update == {} + assert t.resolution is None + + +def test_transition_with_resolution(): + t = Transition("FOO-2", "Done", resolution="Fixed") + assert t.resolution == "Fixed" + assert t.fields["resolution"] == {"name": "Fixed"} + + +def test_transition_as_args(): + t = Transition("FOO-3", "Done", fields={"customfield_1": "v"}, update={"comment": []}) + assert t.as_args() == { + "issue_key": "FOO-3", + "status_name": "Done", + "fields": {"customfield_1": "v"}, + "update": {"comment": []}, + } + + +def test_transition_builder_chain(): + t = TransitionBuilder("FOO-4", "Done").resolution("Won't Do").set_field("timeSpent", "1h").build() + assert isinstance(t, Transition) + assert t.resolution == "Won't Do" + assert t.fields["resolution"] == {"name": "Won't Do"} + assert t.fields["timeSpent"] == "1h" + + +def test_comment_plain_text(): + c = Comment("hello") + assert c.body == "hello" + assert c.visibility is None + + +def test_comment_with_visibility(): + vis = Visibility("role", "Developers") + c = Comment("secret", visibility=vis) + assert c.visibility == vis + + +def test_comment_as_args(): + c = Comment("body", visibility=Visibility("group", "jira-users")) + assert c.as_args() == { + "comment": "body", + "visibility": {"type": "group", "value": "jira-users"}, + } + + +def test_visibility_invalid_type_raises(): + with pytest.raises(ValueError, match="role' or 'group"): + Visibility("team", "x") + + +def test_project_from_dict(): + assert Project.from_dict({"key": "ABC"}) == Project(key="ABC") + assert Project.from_dict({"id": "9"}) == Project(id="9") + + +def test_user_from_dict_cloud_and_server(): + assert User.from_dict({"accountId": "a1"}) == User(account_id="a1") + assert User.from_dict({"name": "jdoe"}) == User(name="jdoe") + + +def test_issue_fields_from_dict_round_trip(): + issue = task().project(key="RT").summary("Round trip").priority(level=PriorityLevel.HIGH).labels("l1", "l2").build() + raw_fields = serialize(issue)["fields"] + parsed = IssueFields.from_dict(raw_fields) + assert parsed.project == Project(key="RT") + assert parsed.summary == "Round trip" + assert parsed.priority == Priority(name="High") + assert parsed.labels == ["l1", "l2"] + assert parsed.issue_type == IssueType(name="Task") + + +def test_jira_issue_from_dict_returns_correct_type(): + data = { + "project": {"key": "P"}, + "summary": "Bug body", + "issuetype": {"name": "Bug"}, + } + issue = JiraIssue.from_dict(data) + assert isinstance(issue, Bug) + assert issue.fields.summary == "Bug body" + + +def test_jira_issue_from_dict_with_fields_wrapper(): + data = { + "fields": { + "project": {"key": "P"}, + "summary": "Wrapped", + "issuetype": {"name": "Task"}, + }, + } + issue = JiraIssue.from_dict(data) + assert isinstance(issue, Task) + assert issue.fields.project == Project(key="P") + + +def test_jira_issue_repr(): + issue = JiraIssue() + issue.fields = IssueFields(project=Project(key="DEMO"), summary="Short title") + assert repr(issue) == "JiraIssue(project='DEMO', summary='Short title')" + issue.fields.summary = "y" * 60 + assert repr(issue) == "JiraIssue(project='DEMO', summary='" + "y" * 47 + "...')" + + +def test_builder_validate_passes_for_valid_issue(): + b = task().project(key="P").summary("OK").validate() + issue = b.build() + assert validate(issue) == [] + + +def test_builder_validate_raises_for_invalid_issue(): + with pytest.raises(ValueError, match="Issue validation failed"): + task().summary("Missing project").validate() + + +def test_bulk_serialize_produces_list(): + i1 = task().project(key="P").summary("One").build() + i2 = task().project(key="P").summary("Two").build() + out = bulk_serialize([i1, i2]) + assert isinstance(out, list) + assert len(out) == 2 + assert out[0]["fields"]["summary"] == "One" + assert out[1]["fields"]["summary"] == "Two" + + +def test_bulk_serialize_with_mapping(): + mapping = FieldMapping(epic_link_field="customfield_777") + issue = task().project(key="P").summary("S").epic_link("E-99").build() + out = bulk_serialize([issue], mapping=mapping) + assert out[0]["fields"]["customfield_777"] == "E-99"