Coverage for functions \ flipdare \ util \ debug_util.py: 100%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

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. 

8# 

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

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

11# 

12 

13import json 

14from collections.abc import Mapping 

15from typing import Any 

16 

17from google.cloud.firestore import Increment 

18from google.cloud.firestore_v1.transforms import Sentinel 

19from google.cloud.firestore_v1.base_query import FieldFilter, BaseCompositeFilter 

20 

21from google.cloud.firestore_v1 import Query, CollectionReference 

22 

23__all__ = ["stringify_debug", "stringify_query"] 

24 

25 

26# NOTE: ------------------------------------------------------------------- 

27# NOTE: This should not be returned to the user, only for internal debugging 

28# NOTE: ------------------------------------------------------------------- 

29 

30 

31def stringify_debug(data: dict[str, Any] | Mapping[str, Any]) -> str: 

32 

33 def json_fallback(obj: Any) -> str: 

34 """Handles Firestore-specific types and any other unknowns.""" 

35 if isinstance(obj, Sentinel): 

36 return "<sentinel-timestamp>" 

37 elif isinstance(obj, Increment): 

38 return "(Increment)" 

39 # Fallback for any other custom objects (like your Models) 

40 return str(obj) 

41 

42 def transform(item: Any, key_name: str | None = None) -> Any: 

43 # 1. Key-based masking 

44 if key_name: 

45 if key_name.endswith("At"): 

46 return "<timestamp>" 

47 if key_name.startswith("hash"): 

48 return "<hash>" 

49 

50 # 2. Handle None explicitly (to get "NONE" instead of null) 

51 if item is None: 

52 return "NONE" 

53 

54 # 3. Recursive traversal 

55 if isinstance(item, dict): 

56 return {k: transform(v, k) for k, v in item.items()} # type: ignore 

57 if isinstance(item, list): 

58 return [transform(i) for i in item] 

59 

60 return item 

61 

62 # special case, one string key, one string value. 

63 if isinstance(data, dict) and len(data) == 1: 

64 only_key, only_value = next(iter(data.items())) 

65 if isinstance(only_value, str): 

66 return f"{only_key}={only_value}" 

67 

68 # Clean the data keys/None values first 

69 clean_data = transform(data) 

70 

71 # Dump with the Firestore-type fallback 

72 return json.dumps(clean_data, indent=2, default=json_fallback, sort_keys=True) 

73 

74 

75def stringify_query(query: Query | CollectionReference) -> str: # pragma: no cover 

76 msg = str(query) 

77 parent = getattr(query, "_parent", None) 

78 if parent: 

79 msg = f"{parent.id!s} -> {msg}" 

80 

81 filter_parts = [] 

82 # New-style composite filter (query.where(filter=...)) is stored in _query_filter 

83 query_filter = getattr(query, "_query_filter", None) 

84 if query_filter: 

85 filter_parts.append(_filter_to_string(query_filter)) 

86 

87 # Old-style field filters (query.where("field", "op", value)) 

88 field_filters = getattr(query, "_field_filters", []) 

89 filter_parts.extend(_filter_to_string(f) for f in field_filters) 

90 

91 filter_str = "\t" + " AND ".join(filter_parts) if filter_parts else "No filters" 

92 

93 msg += "\n------------------------ Filter -----------------------\n" 

94 msg += f"{filter_str}" 

95 msg += "\n-------------------------------------------------------\n" 

96 return msg 

97 

98 

99def _filter_to_string(filter_obj: Any) -> str: 

100 if isinstance(filter_obj, FieldFilter): 

101 # Access field_path, op_string, and value from the FieldFilter object 

102 return f"{filter_obj.field_path} {filter_obj.op_string} {filter_obj.value}" 

103 

104 elif isinstance(filter_obj, BaseCompositeFilter): 

105 # Recursively handle nested filters in AND/OR blocks 

106 op = "AND" if "And" in str(type(filter_obj)) else "OR" 

107 parts = [_filter_to_string(f) for f in filter_obj.filters] 

108 return f"({f' {op} '.join(parts)})" 

109 

110 return str(filter_obj)