Coverage for functions\flipdare\error\message

1#!/usr/bin/env python

4# This file is part of Flipdare's proprietary software and contains

5# confidential and copyrighted material. Unauthorised copying,

6# modification, distribution, or use of this file is strictly

7# prohibited without prior written permission from Flipdare Pty Ltd.

9# This software includes third-party components licensed under MIT,

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.

11#

13from __future__ import annotations

15import re

16from collections import defaultdict

17from pydantic import TypeAdapter

18from typing import Any, TypeGuard, cast, override

19from collections.abc import Mapping

20from datetime import datetime

21from tabulate import tabulate

22from dataclasses import dataclass

23from jinja2.exceptions import TemplateSyntaxError

24from pydantic_core import ValidationError

25from flipdare.app_types import JsonDict

26from flipdare.constants import NO_DOC_ID

27from flipdare.error.app_error import AppError

28from flipdare.generated.shared.app_error_code import AppErrorCode

29from flipdare.message.user_error_code import UserErrorCode

30from flipdare.result.app_result import AppResult, ResultExtraType

31from flipdare.error.app_error_protocol import AppErrorProtocol

32from flipdare.error.log_context import LogContext

33from flipdare.error.stack_util import StackUtil

34from flipdare.generated.shared.backend.system_log_type import SystemLogType

35from flipdare.util.ansi_codes import AnsiCodes

36from flipdare.generated.shared.firestore_collections import FirestoreCollections

37from flipdare.job.trigger_data import TriggerData, UpdateTriggerData

38from flipdare.util.debug_util import stringify_debug

39from flipdare.util.time_util import TimeUtil

41__all__ = [

42 "JinjaTmplErrorMsgFormat",

43 "ValidationErrorMsgFormat",

44 "ConfigErrorMsgFormat",

45 "JobErrorErrorMsgFormat",

46 "InfoMsgFormat",

47 "ErrorMsgFormat",

48 "AppErrorMsgFormat",

49 "TriggerErrorMsgFormat",

50 "MessageContext",

51 "MessageSection",

52 "BaseMsgFormat",

53]

55_C = AnsiCodes

56_OccuredAtKey = "Occurred At"

57_MAX_CODE_WIDTH = 24

60SITE_PACKAGES_RE = re.compile(

61 r"(?:[A-Z]:[\\/]|/).*?lib[\\/].*?site-packages[\\/]", re.IGNORECASE | re.DOTALL

62)

63SITE_PACKAGES_REPLACEMENT = r"../site-packages/"

65FLIPDARE_PACKAGES_RE = re.compile(r"(?:[A-Z]:[\\/]|/).*?flipdare[\\/]", re.IGNORECASE | re.DOTALL)

67SCHEMA_PATTERN = r"(_?schema)+"

70def _sanitize_message_info(s: str) -> str:

71 s = SITE_PACKAGES_RE.sub(SITE_PACKAGES_REPLACEMENT, s)

72 return FLIPDARE_PACKAGES_RE.sub("", s)

75def _format_extra_info(extra: dict[str, Any] | list[str] | str) -> str:

76 match extra:

77 case dict():

78 adapter = TypeAdapter(dict)

79 pretty_bytes = adapter.dump_json(extra, indent=4)

80 return _sanitize_message_info(pretty_bytes.decode())

81 case list():

82 return "\n".join(f"- {_sanitize_message_info(item)}" for item in extra)

83 case _:

84 return _sanitize_message_info(str(extra))

87class _guards: # noqa: N801

88 @staticmethod

89 def is_validation_error(error: Any) -> TypeGuard[ValidationError]:

90 return isinstance(error, ValidationError)

92 @staticmethod

93 def is_list_error(error: Any) -> TypeGuard[list[str]]:

94 return isinstance(error, list) and all(isinstance(e, str) for e in error)

97class BaseMsgFormat:

98 __slots__ = (

99 "_context_msg",

100 "_error_msg",

101 "_log_type",

102 "_message",

103 "_occurred_at",

104 "_section_msg",

105 "_source",

106 "_user_error_code",

107 )

108

109 _message: str

110 _log_type: SystemLogType

111 _context_msg: MessageContext

112 _section_msg: MessageSection | None

113 _occurred_at: str

114

115 def __init__(

116 self,

117 log_type: SystemLogType,

118 source: str,

119 user_error_code: str,

120 message: str,

121 context: MessageContext,

122 section: MessageSection | None = None,

123 occurred_at: datetime | None = None,

124 ) -> None:

125

126 self._source = source

127 self._log_type = log_type

128

129 if occurred_at is None:

130 occurred_at = TimeUtil.get_current_utc_dt()

131 self._occurred_at = TimeUtil.formatted_user(occurred_at)

132

133 self._user_error_code = user_error_code

134 self._message = message

135 self._context_msg = context

136 self._section_msg = section

137

138 # preload the error message, so there are not repeated calls to formatting ..

139 self._error_msg = self._preload_error()

140

141 @property

142 def message(self) -> str:

143 return self._message

144

145 @property

146 def source(self) -> str:

147 return self._source

148

149 @property

150 def log_type(self) -> SystemLogType:

151 return self._log_type

152

153 def extra(self) -> JsonDict | None:

154 # this is used if a formatter is present in the log context.

155 # we parse the section/context into a dict[str,Any]

156 if self._section_msg is None:

157 return None

158

159 return self._section_msg.to_dict()

160

161 @property

162 def user_error_code(self) -> str:

163 # this is what is returned to the user so they can use

164 # that to contact support to help identify issues.

165 return self._user_error_code

166

167 @override

168 def __str__(self) -> str:

169 return self._error_msg

170

171 @override

172 def __repr__(self) -> str:

173 return self._error_msg

174

175 def _preload_error(self) -> str:

176 c = _ErrorColors(log_type=self._log_type)

177

178 msg = f"\n\n{c.base}" + f"== {self._log_type.name} ==" * 8 + f"{c.reset}\n\n"

179 msg += (

180 f"{c.highlight}{self._user_error_code}:{c.reset} {c.bold}{self._message}{c.reset}\n\n"

181 )

182

183 msg += self._context_msg.table(self._occurred_at) + "\n\n"

184 if self._section_msg is not None:

185 table = self._section_msg.table()

186 msg += table + "\n\n" if table is not None else ""

187

188 msg += f"\n\n{c.base}" + f"== {self._log_type.name} END ==" * 6 + f"{c.reset}\n\n"

189

190 return msg

191

192

193class JinjaTmplErrorMsgFormat(BaseMsgFormat):

194 def __init__(

195 self,

196 template_name: str,

197 error: Exception,

198 jinja_label: str | None = None,

199 ) -> None:

200 ctx = MessageContext(log_type=SystemLogType.ERROR)

201

202 if isinstance(error, TemplateSyntaxError):

203 ctx.add("Type", type(error).__name__)

204 ctx.add("Message", error.message)

205 ctx.add("Filename", error.filename)

206 ctx.add("Line Number", error.lineno)

207 else:

208 ctx.add("Type", type(error).__name__)

209 ctx.add("Message", str(error))

210

211 if jinja_label is not None:

212 ctx.add("Jinja Label", jinja_label)

213

214 section = MessageSection(log_type=SystemLogType.ERROR)

215 section.add("Stack Trace", StackUtil().get_flipdare_stack())

216

217 super().__init__(

218 log_type=SystemLogType.ERROR,

219 source=f"jinja_template:{template_name}",

220 user_error_code=f"jinja_{template_name}",

221 message=f"Error parsing Jinja template '{template_name}'",

222 context=ctx,

223 section=section,

224 )

225

226

227class ValidationErrorMsgFormat(BaseMsgFormat):

228 __slots__ = ("_class_name",)

229 # ^(?:_?schema)+ matches _Schema, schema, _Schema_schema, etc. at start

230 SCHEMA_PATTERN = r"(_?schema)+"

231

232 def __init__(

233 self,

234 class_type: type[Any],

235 error: Exception | list[str],

236 parse_failed: bool = False,

237 message: str | None = None,

238 ) -> None:

239 class_name = class_type.__name__

240 error_ct = 1

241 if _guards.is_validation_error(error):

242 error_ct = len(error.errors())

243 elif _guards.is_list_error(error):

244 error_ct = len(error)

245

246 # format the class name for user error codes

247 user_error_code = UserErrorCode.validation(class_type, error_ct, parse_failed=parse_failed)

248

249 ctx = MessageContext(log_type=SystemLogType.ERROR)

250 ctx.add("Class", class_name)

251 ctx.add("Error Count", error_ct)

252 ctx.add("Error Code", user_error_code)

253

254 # format the error details

255 section = MessageSection(log_type=SystemLogType.ERROR)

256

257 if _guards.is_validation_error(error):

258 for err in error.errors():

259 loc_value = " -> ".join(str(loc) for loc in err.get("loc", []))

260 section.add("Validation", f"{loc_value}: {err.get('msg', '')}")

261 elif _guards.is_list_error(error):

262 section.add("Validation", "\n".join(error))

263 else:

264 section.add("Validation", str(error))

265

266 validation_msg = f"Validation error for {class_name}: {error_ct} error(s) found."

267 message = validation_msg if message is None else f"{message}\n\n{validation_msg}"

268

269 self._class_name = class_name

270 super().__init__(

271 log_type=SystemLogType.ERROR,

272 source=f"validation:{class_name}",

273 message=message,

274 user_error_code=user_error_code,

275 context=ctx,

276 section=section,

277 )

278

279 @property

280 def class_name(self) -> str:

281 return self._class_name

282

283

284class TriggerErrorMsgFormat(BaseMsgFormat):

285 __slots__ = ("_extra",)

286

287 def __init__(

288 self,

289 validator: TriggerData[Any, Any],

290 message: str | None = None,

291 ) -> None:

292 if message is None:

293 message = "Trigger validation failed"

294

295 ctx = MessageContext(log_type=SystemLogType.ERROR)

296 ctx.add("Job Name", validator.job_type)

297 ctx.add("Event", type(validator.event).__name__)

298 ctx.add("Wrapper Class", validator.wrapper_class.__name__)

299 ctx.add("Data Class", type(validator._result).__name__)

300 ctx.add("Doc ID", validator.doc_id or NO_DOC_ID)

301

302 section = MessageSection(log_type=SystemLogType.ERROR)

303

304 evt = validator.event

305 section.add("Event", stringify_debug(cast("Mapping[str, Any]", evt)))

306

307 params_msg = (

308 stringify_debug(validator.params)

309 if validator.params is not None

310 else "No params found."

311 )

312 section.add("Params", params_msg)

313

314 extra = validator.data or {}

315 self._extra = extra

316

317 validator_msg = stringify_debug(self._extra) if self._extra else "No data found."

318 section.add("Data", validator_msg)

319

320 if isinstance(validator, UpdateTriggerData):

321 if validator.before_data is not None:

322 section.add("Before Data", stringify_debug(validator.before_data))

323 else:

324 section.add("Before Data", "No before data found.")

325

326 if validator.errors is not None:

327 validation_errors = validator.errors

328 if len(validation_errors) > 0:

329 section.add("Validation Errors", _format_extra_info(validation_errors))

330 else:

331 section.add("Validation Errors", "No validation errors found.")

332

333 super().__init__(

334 log_type=SystemLogType.ERROR,

335 source=f"trigger:{validator.job_type}",

336 user_error_code=UserErrorCode.from_trigger_data(validator),

337 message=message,

338 context=ctx,

339 section=section,

340 )

341

342 @override

343 def extra(self) -> JsonDict:

344 return self._extra

345

346

347def _format_app_result(result: AppResult[Any]) -> defaultdict[str, list[list[str]]] | None:

348 errors = result._errors

349 warnings = result._warnings

350

351 entries: defaultdict[str, list[list[str]]] = defaultdict(list)

352

353 if len(errors) > 0:

354 extra_info: dict[AppErrorProtocol, ResultExtraType] = result._extra_error_info

355

356 for task, errs in result._errors.items():

357 for err in errs:

358 extra_info_str = ""

359 if err.error_code in extra_info:

360 error_info = extra_info[err.error_code]

361 extra_info_str = (

362 error_info if isinstance(error_info, str) else stringify_debug(error_info)

363 )

364 msg = f"{err.message}\n{extra_info_str}" if extra_info_str else err.message

365 entries["ERROR"].append([err.error_code.display_title, task, msg])

366

367 if len(warnings) > 0:

368 for task, warns in warnings.items():

369 for warn in warns:

370 entries["WARN"].append(["Warning", task, warn.message])

371

372 return entries if len(entries) > 0 else None

373

374

375class AppResultErrorMsgFormat(BaseMsgFormat):

376 __slots__ = ()

377

378 def __init__(

379 self,

380 result: AppResult[Any],

381 duration: int | None = None,

382 ) -> None:

383 error_code = result.main_error or AppErrorCode.SERVER

384 outcome = result.outcome

385 main_task = result.main_task

386

387 ctx = MessageContext(log_type=SystemLogType.ERROR)

388 ctx.add("Error Code", f"{error_code.value}/{error_code.category}")

389 ctx.add("Main Task", main_task)

390 ctx.add("Duration", f"{duration or 'N/A'} seconds")

391 ctx.add("Outcome", outcome.value)

392 ctx.add("Doc ID", result.doc_id)

393

394 errors = result._errors

395 warnings = result._warnings

396

397 section = MessageSection(log_type=SystemLogType.ERROR)

398

399 formatted_result = _format_app_result(result)

400 if formatted_result is not None:

401 for title, rows in formatted_result.items():

402 for row in rows:

403 section.add_row(title, row)

404

405 super().__init__(

406 log_type=SystemLogType.ERROR,

407 source=f"app_result:{result.doc_id}",

408 user_error_code=error_code.value,

409 message=f"AppResult contains {len(errors)} error(s) and {len(warnings)} warning(s).",

410 context=ctx,

411 section=section,

412 )

413

414

415class ConfigErrorMsgFormat(BaseMsgFormat):

416 __slots__ = ()

417

418 def __init__(

419 self,

420 missing_keys: list[str] | None = None,

421 template_errors: list[str] | None = None,

422 config_error: Exception | None = None,

423 ) -> None:

424 ctx = MessageContext(log_type=SystemLogType.ERROR)

425 ctx.add("Error Type", "Configuration error")

426

427 missing_keys = missing_keys or []

428 template_errors = template_errors or []

429

430 ctx.add("Missing Keys Count", len(missing_keys))

431 ctx.add("Template Errors Count", len(template_errors))

432

433 section = MessageSection(log_type=SystemLogType.ERROR)

434 for missing_key in missing_keys:

435 section.add("Missing Key", missing_key)

436 for template_error in template_errors:

437 section.add("Template Error", template_error)

438

439 section.add("Stack Trace", StackUtil().get_flipdare_stack())

440

441 message = (

442 f"Configuration Error: {len(missing_keys)} missing config keys, "

443 f"{len(template_errors)} template errors"

444 )

445 if config_error is not None:

446 message += ", Exception thrown"

447 section.add("Config Exception", _format_extra_info(str(config_error)))

448

449 super().__init__(

450 log_type=SystemLogType.ERROR,

451 source=StackUtil().get_caller_str(),

452 user_error_code="config_error",

453 message=message,

454 context=ctx,

455 section=section,

456 )

457

458

459class JobErrorErrorMsgFormat(BaseMsgFormat):

460 __slots__ = ()

461

462 def __init__(

463 self,

464 error_code: AppErrorProtocol,

465 source: FirestoreCollections | str | None = None,

466 job_str: str | None = None,

467 doc_id: str | None = None,

468 message: str | None = None,

469 detailed_error: str | None = None,

470 error: str | Exception | None = None,

471 is_missing: bool = False,

472 ) -> None:

473

474 actual_message = message

475 if is_missing:

476 if source == FirestoreCollections.USER:

477 actual_message = f"User {doc_id} not found during {job_str}: {message}"

478 else:

479 actual_message = (

480 f"Missing document {doc_id} in {source} during {job_str}: {message}"

481 )

482 if actual_message is None:

483 actual_message = (

484 f"Error during {job_str} for document {doc_id} in {source}:"

485 " No additional message provided."

486 )

487

488 ctx = MessageContext(log_type=SystemLogType.ERROR)

489 ctx.add("Error Code", error_code.value)

490 ctx.add("Category", error_code.category.value)

491 ctx.add("Source", source if source is not None else "N/A")

492 ctx.add("Job", job_str if job_str is not None else "N/A")

493 ctx.add("Doc ID", doc_id if doc_id is not None else "N/A")

494

495 section = MessageSection(log_type=SystemLogType.ERROR)

496 if detailed_error is not None:

497 section.add("Detail", detailed_error)

498

499 if error is not None:

500 section.add("Error Details", str(error))

501

502 super().__init__(

503 log_type=SystemLogType.ERROR,

504 source=source or StackUtil().get_caller_str(),

505 user_error_code=error_code.value,

506 message=actual_message,

507 context=ctx,

508 section=section,

509 )

510

511

512class InfoMsgFormat(BaseMsgFormat):

513 __slots__ = ()

514

515 def __init__(

516 self,

517 ctx: LogContext,

518 ) -> None:

519

520 msg_ctx = MessageContext(log_type=SystemLogType.INFO)

521 msg_ctx.add("Category", ctx.category)

522 msg_ctx.add("Called From", ctx.called_by)

523 msg_ctx.add("Source", ctx.source)

524 msg_ctx.add("Doc ID", ctx.doc_id)

525 msg_ctx.add("Job Type", ctx.job_type or "N/A")

526 msg_ctx.add("Collection", ctx.collection or "N/A")

527

528 super().__init__(

529 log_type=SystemLogType.INFO,

530 source=ctx.source,

531 message=ctx.message,

532 user_error_code="info",

533 context=msg_ctx,

534 )

535

536

537class AppErrorMsgFormat(BaseMsgFormat):

538 def __init__(self, err: AppError) -> None:

539

540 msg_ctx = MessageContext(log_type=SystemLogType.INFO)

541 msg_ctx.add("Source", err.source)

542 msg_ctx.add("Category", err.category)

543 msg_ctx.add("Error Code", err.error_code.value)

544 msg_ctx.add("Http Code", err.http_code or "N/A")

545 msg_ctx.add("Title", err.title or "N/A")

546

547 section = MessageSection(log_type=SystemLogType.ERROR)

548

549 cause_message = err.cause_message

550 if cause_message is not None:

551 section.add("Cause", _sanitize_message_info(cause_message))

552

553 ex = err.cause

554 if ex is not None:

555 section.add("Exception", _sanitize_message_info(str(ex)))

556

557 super().__init__(

558 log_type=SystemLogType.ERROR,

559 source=err.source,

560 user_error_code=err.error_code.value,

561 message=err.message,

562 context=msg_ctx,

563 section=section,

564 )

565

566

567class ErrorMsgFormat(BaseMsgFormat):

568 __slots__ = ()

569

570 def __init__(

571 self,

572 ctx: LogContext,

573 ) -> None:

574

575 error_code_str = ctx.error_code_str

576

577 msg_ctx = MessageContext(log_type=ctx.log_type)

578 msg_ctx.add("Category", ctx.category)

579 msg_ctx.add("Called From", ctx.called_by)

580 msg_ctx.add("Error Code", error_code_str)

581 msg_ctx.add("Source", ctx.source)

582 msg_ctx.add("Job Type", ctx.job_type)

583 msg_ctx.add("Doc ID", ctx.doc_id)

584

585 section = MessageSection(log_type=ctx.log_type)

586 if result := ctx.result:

587 formatted_result = _format_app_result(result)

588 if formatted_result is not None:

589 for title, rows in formatted_result.items():

590 for row in rows:

591 section.add_row(title, row)

592 if extra := ctx.extra:

593 section.add("Extra", _format_extra_info(extra))

594 if ctx.stack_trace:

595 section.add("Stack Trace", ctx.stack_trace)

596

597 super().__init__(

598 log_type=ctx.log_type,

599 source=ctx.source,

600 message=ctx.message,

601 user_error_code=error_code_str,

602 context=msg_ctx,

603 section=section,

604 )

605

606

607@dataclass(frozen=True, kw_only=True)

608class _ErrorColors:

609 log_type: SystemLogType

610

611 @property

612 def reset(self) -> str:

613 return _C.RESET_COLOR

614

615 @property

616 def bold(self) -> str:

617 return _C.BOLD

618

619 @property

620 def base(self) -> str:

621 match self.log_type:

622 case SystemLogType.ERROR:

623 return _C.RED

624 case SystemLogType.WARNING:

625 return _C.ORANGE

626 case _:

627 return _C.LIGHT_GREEN

628

629 @property

630 def highlight(self) -> str:

631 match self.log_type:

632 case SystemLogType.ERROR:

633 return _C.LIGHT_RED

634 case SystemLogType.WARNING:

635 return _C.LIGHT_YELLOW

636 case _:

637 return _C.LIGHT_GREEN

638

639

640class MessageContext:

641 __slots__ = ("_log_type", "_messages")

642

643 def __init__(self, log_type: SystemLogType) -> None:

644 self._log_type = log_type

645 self._messages: list[tuple[str, Any]] = []

646

647 def add(self, key: str, value: Any) -> None:

648 self._messages.append((key, value))

649

650 def table(self, occurred_at: str) -> str:

651 c = _ErrorColors(log_type=self._log_type)

652

653 messages = list(self._messages)

654 if _OccuredAtKey not in [key for key, _ in messages]:

655 messages.insert(0, (_OccuredAtKey, occurred_at))

656

657 # format conttent

658 context: list[list[str]] = [[f"{c.bold}{c.highlight}Context{c.reset}{c.reset}", ""]]

659 for key, value in messages:

660 context.append([f"{key:^20}", f"{c.base}{value}{c.reset}"])

661

662 return tabulate(context, tablefmt="grid", maxcolwidths=[_MAX_CODE_WIDTH, None])

663

664

665class MessageSection:

666 __slots__ = ("_log_type", "_sections")

667

668 def __init__(self, log_type: SystemLogType) -> None:

669 self._log_type = log_type

670 self._sections: defaultdict[str, list[list[str]]] = defaultdict(list)

671

672 def add(self, title: str, value: str) -> None:

673 self._sections[title].append([value])

674

675 def add_row(self, title: str, row: list[str]) -> None:

676 self._sections[title].append(row)

677

678 def to_dict(self) -> JsonDict:

679 # this is used if a formatter is present in the log context.

680 # we parse the section/context into a dict[str,Any]

681 return self._sections

682

683 def table(self) -> str | None:

684 c = _ErrorColors(log_type=self._log_type)

685 sections = self._sections

686

687 extra: list[list[str]] = []

688

689 # we need to pad the row so we need to know the max cols for all sections

690 max_cols = (

691 max(len(row) for rows in sections.values() for row in rows) if len(sections) > 0 else 0

692 )

693

694 for title, content in sections.items():

695 title_str = f"{c.bold}{c.highlight}{title}{c.reset}"

696

697 # if content is a single entry, format

698 # otherwise we assume it is already formatted as a row and just add it

699 content_cols = max(len(row) for row in content)

700

701 if content_cols == 1:

702 content_msg = "\n".join(cell[0] for cell in content)

703 content_msg = _sanitize_message_info(content_msg.strip())

704 row = [title_str, content_msg] + [""] * (max_cols - 2)

705 extra.append(row)

706 else:

707 row = [title_str] + [""] * (max_cols - 1)

708 extra.append(row)

709 for content_row in content:

710 sanitized_row = [_sanitize_message_info(cell) for cell in content_row]

711 padded_row = sanitized_row + [""] * (max_cols - len(sanitized_row))

712 extra.append(padded_row)

713

714 if not extra:

715 return None

716

717 # note setting max_col_width causes tabulate to strip newlines..

718 max_col_widths: list[int | None] = []

719 if max_cols == 2: # noqa: PLR2004

720 # dont inclide a code width

721 max_col_widths = [None, None]

722 else:

723 max_col_widths = [_MAX_CODE_WIDTH] + [None] * (max_cols - 1)

724

725 return tabulate(extra, tablefmt="grid", maxcolwidths=max_col_widths)

Coverage for functions \ flipdare \ error \ message_format.py: 88%

394 statements