| 213 | |
| 214 | |
| 215 | class IssueUpdate: |
| 216 | def __init__(self, issue, github_session, git_repo): |
| 217 | self.issue = issue |
| 218 | self.update_payload = {} |
| 219 | self.github_session = github_session |
| 220 | self.git_repo = git_repo |
| 221 | self._pr_cache = {} |
| 222 | self.has_changes = False # New flag to track if changes are made |
| 223 | self.transform = None |
| 224 | logger_extra = { |
| 225 | 'issue_id': issue.id, |
| 226 | 'current_transform': None, |
| 227 | } |
| 228 | self.logger = IssueLoggerAdapter(logging.getLogger(__name__), extra=logger_extra) |
| 229 | |
| 230 | def set_transform(self, transform): |
| 231 | self.transform = transform |
| 232 | self.logger.extra['current_transform'] = transform |
| 233 | |
| 234 | def get_raw_custom_field(self, field_id): |
| 235 | cf = self.issue.custom_fields.get(field_id) |
| 236 | try: |
| 237 | return cf.value if cf else None |
| 238 | except redminelib.exceptions.ResourceAttrError: |
| 239 | return None |
| 240 | |
| 241 | def get_custom_field(self, field_id): |
| 242 | """ Get the custom field, first from update_payload otherwise issue """ |
| 243 | custom_fields = self.update_payload.setdefault("custom_fields", []) |
| 244 | for field in custom_fields: |
| 245 | if field.get('id') == field_id: |
| 246 | return field['value'] |
| 247 | return self.get_raw_custom_field(field_id) |
| 248 | |
| 249 | def add_or_update_custom_field(self, field_id, value): |
| 250 | """Helper to add or update a custom field in the payload.""" |
| 251 | custom_fields = self.update_payload.setdefault("custom_fields", []) |
| 252 | found = False |
| 253 | current_value = self.get_custom_field(field_id) # Get current value from issue or payload |
| 254 | |
| 255 | if current_value == value: |
| 256 | # Value is already the same, no change needed |
| 257 | self.logger.debug(f"Field {field_id} is already set to '{value}'. No update needed.") |
| 258 | return False |
| 259 | |
| 260 | self.logger.debug(f"Updating custom field {field_id} from '{current_value}' to '{value}'.") |
| 261 | for field in custom_fields: |
| 262 | if field.get('id') == field_id: |
| 263 | field['value'] = value |
| 264 | found = True |
| 265 | break |
| 266 | if not found: |
| 267 | custom_fields.append({'id': field_id, 'value': value}) |
| 268 | self.has_changes = True # Mark that a change has been made |
| 269 | return True |
| 270 | |
| 271 | def change_field(self, field, value): |
| 272 | self.logger.debug(f"Changing field '{field}' to '{value}'.") |
no outgoing calls
no test coverage detected