+5 -3

src/atpasser/blob/__init__.py

··· 1 import cid 2 import multihash, hashlib 3 0 4 def generateCID(file): 5 hasher = hashlib.new("sha-256") 6 while True: 7 chunk = file.read(8192) 8 - if not chunk: break 0 9 hasher.update(chunk) 10 - 11 digest = hasher.digest 12 mh = multihash.encode(digest, "sha-256") 13 14 - return cid.CIDv1(codec='raw', multihash=mh)

··· 1 import cid 2 import multihash, hashlib 3 4 + 5 def generateCID(file): 6 hasher = hashlib.new("sha-256") 7 while True: 8 chunk = file.read(8192) 9 + if not chunk: 10 + break 11 hasher.update(chunk) 12 + 13 digest = hasher.digest 14 mh = multihash.encode(digest, "sha-256") 15 16 + return cid.CIDv1(codec="raw", multihash=mh)

+1 -1

src/atpasser/data/__init__.py

··· 46 "dumps", 47 "load", 48 "loads", 49 - ]

··· 46 "dumps", 47 "load", 48 "loads", 49 + ]

+41 -37

src/atpasser/data/decoder.py

··· 13 14 class JsonDecoder(json.JSONDecoder): 15 """A JSON decoder that supports ATProto data types. 16 - 17 This decoder extends the standard JSON decoder to handle ATProto-specific 18 data types, including bytes, CID links, and typed objects. 19 - 20 Attributes: 21 type_hook_registry: Registry for type-specific hooks. 22 encoding: The encoding to use for string deserialization. 23 """ 24 - 25 def __init__( 26 self, 27 *, ··· 29 type_hook_registry: Optional[Any] = None, 30 type_processor_registry: Optional[Any] = None, 31 encoding: str = "utf-8", 32 - **kwargs: Any 33 ) -> None: 34 """Initialize the JSON decoder. 35 - 36 Args: 37 object_hook: Optional function to call with each decoded object. 38 type_hook_registry: Registry for type-specific hooks. ··· 45 type_hook_registry = type_processor_registry.to_hook_registry() 46 elif type_hook_registry is None: 47 from .hooks import get_global_registry 0 48 type_hook_registry = get_global_registry() 49 - 50 # Create a combined object hook that calls both the custom hook and our hook 51 combined_hook = self._create_combined_hook(object_hook, type_hook_registry) 52 - 53 super().__init__(object_hook=combined_hook, **kwargs) 54 self.type_hook_registry = type_hook_registry 55 self.type_processor_registry = type_processor_registry 56 self.encoding = encoding 57 - 58 def _create_combined_hook( 59 self, 60 custom_hook: Optional[Callable[[Dict[str, Any]], Any]], 61 - type_hook_registry: Optional[Any] 62 ) -> Callable[[Dict[str, Any]], Any]: 63 """Create a combined object hook function. 64 - 65 Args: 66 custom_hook: Optional custom object hook function. 67 type_hook_registry: Registry for type-specific hooks. 68 - 69 Returns: 70 A combined object hook function. 71 """ 0 72 def combined_hook(obj: Dict[str, Any]) -> Any: 73 # First, apply our ATProto-specific decoding 74 decoded_obj = self._atproto_object_hook(obj) 75 - 76 # Then, apply the custom hook if provided 77 if custom_hook is not None: 78 decoded_obj = custom_hook(decoded_obj) 79 - 80 return decoded_obj 81 - 82 return combined_hook 83 - 84 def _atproto_object_hook(self, obj: Dict[str, Any]) -> Any: 85 """Handle ATProto-specific object decoding. 86 - 87 Args: 88 obj: The object to decode. 89 - 90 Returns: 91 The decoded object. 92 """ ··· 96 # If there are other keys, this is invalid 97 raise ValueError(f"Invalid $bytes object: {obj}") 98 return base64.b64decode(obj["$bytes"].encode(self.encoding)) 99 - 100 # Handle $link key (CID parsing) 101 elif "$link" in obj: 102 if len(obj) != 1: 103 # If there are other keys, this is invalid 104 raise ValueError(f"Invalid $link object: {obj}") 105 return make_cid(obj["$link"]) 106 - 107 # Handle $type key (typed objects) 108 elif "$type" in obj: 109 type_value = obj["$type"] 110 remaining_obj = {k: v for k, v in obj.items() if k != "$type"} 111 - 112 # Check if there's a registered type handler 113 if self.type_hook_registry is not None: 114 handler = self.type_hook_registry.get_handler(type_value) 115 if handler is not None: 116 return handler(remaining_obj) 117 - 118 # If no handler is registered, return a typed object 119 return TypedObject(type_value, remaining_obj) 120 - 121 # Handle nested objects recursively 122 elif isinstance(obj, dict): 123 - return {k: self._atproto_object_hook(v) if isinstance(v, dict) else v 124 - for k, v in obj.items()} 125 - 0 0 126 return obj 127 128 129 class TypedObject: 130 """A typed object in the ATProto data model. 131 - 132 This class represents an object with a $type field in the ATProto data model. 133 - 134 Attributes: 135 type: The type of the object. 136 data: The data associated with the object. 137 """ 138 - 139 def __init__(self, type_name: str, data: Dict[str, Any]) -> None: 140 """Initialize a typed object. 141 - 142 Args: 143 type_name: The type of the object. 144 data: The data associated with the object. 145 """ 146 self.type_name = type_name 147 self.data = data 148 - 149 def __repr__(self) -> str: 150 """Return a string representation of the typed object. 151 - 152 Returns: 153 A string representation of the typed object. 154 """ 155 return f"TypedObject(type_name={self.type_name!r}, data={self.data!r})" 156 - 157 def __eq__(self, other: Any) -> bool: 158 """Check if two typed objects are equal. 159 - 160 Args: 161 other: The object to compare with. 162 - 163 Returns: 164 True if the objects are equal, False otherwise. 165 """ 166 if not isinstance(other, TypedObject): 167 return False 168 return self.type_name == other.type_name and self.data == other.data 169 - 170 def __atproto_json_encode__(self) -> Dict[str, Any]: 171 """Encode the typed object to a JSON-serializable format. 172 - 173 Returns: 174 A JSON-serializable representation of the typed object. 175 """ 176 result = {"$type": self.type_name} 177 result.update(self.data) 178 - return result

··· 13 14 class JsonDecoder(json.JSONDecoder): 15 """A JSON decoder that supports ATProto data types. 16 + 17 This decoder extends the standard JSON decoder to handle ATProto-specific 18 data types, including bytes, CID links, and typed objects. 19 + 20 Attributes: 21 type_hook_registry: Registry for type-specific hooks. 22 encoding: The encoding to use for string deserialization. 23 """ 24 + 25 def __init__( 26 self, 27 *, ··· 29 type_hook_registry: Optional[Any] = None, 30 type_processor_registry: Optional[Any] = None, 31 encoding: str = "utf-8", 32 + **kwargs: Any, 33 ) -> None: 34 """Initialize the JSON decoder. 35 + 36 Args: 37 object_hook: Optional function to call with each decoded object. 38 type_hook_registry: Registry for type-specific hooks. ··· 45 type_hook_registry = type_processor_registry.to_hook_registry() 46 elif type_hook_registry is None: 47 from .hooks import get_global_registry 48 + 49 type_hook_registry = get_global_registry() 50 + 51 # Create a combined object hook that calls both the custom hook and our hook 52 combined_hook = self._create_combined_hook(object_hook, type_hook_registry) 53 + 54 super().__init__(object_hook=combined_hook, **kwargs) 55 self.type_hook_registry = type_hook_registry 56 self.type_processor_registry = type_processor_registry 57 self.encoding = encoding 58 + 59 def _create_combined_hook( 60 self, 61 custom_hook: Optional[Callable[[Dict[str, Any]], Any]], 62 + type_hook_registry: Optional[Any], 63 ) -> Callable[[Dict[str, Any]], Any]: 64 """Create a combined object hook function. 65 + 66 Args: 67 custom_hook: Optional custom object hook function. 68 type_hook_registry: Registry for type-specific hooks. 69 + 70 Returns: 71 A combined object hook function. 72 """ 73 + 74 def combined_hook(obj: Dict[str, Any]) -> Any: 75 # First, apply our ATProto-specific decoding 76 decoded_obj = self._atproto_object_hook(obj) 77 + 78 # Then, apply the custom hook if provided 79 if custom_hook is not None: 80 decoded_obj = custom_hook(decoded_obj) 81 + 82 return decoded_obj 83 + 84 return combined_hook 85 + 86 def _atproto_object_hook(self, obj: Dict[str, Any]) -> Any: 87 """Handle ATProto-specific object decoding. 88 + 89 Args: 90 obj: The object to decode. 91 + 92 Returns: 93 The decoded object. 94 """ ··· 98 # If there are other keys, this is invalid 99 raise ValueError(f"Invalid $bytes object: {obj}") 100 return base64.b64decode(obj["$bytes"].encode(self.encoding)) 101 + 102 # Handle $link key (CID parsing) 103 elif "$link" in obj: 104 if len(obj) != 1: 105 # If there are other keys, this is invalid 106 raise ValueError(f"Invalid $link object: {obj}") 107 return make_cid(obj["$link"]) 108 + 109 # Handle $type key (typed objects) 110 elif "$type" in obj: 111 type_value = obj["$type"] 112 remaining_obj = {k: v for k, v in obj.items() if k != "$type"} 113 + 114 # Check if there's a registered type handler 115 if self.type_hook_registry is not None: 116 handler = self.type_hook_registry.get_handler(type_value) 117 if handler is not None: 118 return handler(remaining_obj) 119 + 120 # If no handler is registered, return a typed object 121 return TypedObject(type_value, remaining_obj) 122 + 123 # Handle nested objects recursively 124 elif isinstance(obj, dict): 125 + return { 126 + k: self._atproto_object_hook(v) if isinstance(v, dict) else v 127 + for k, v in obj.items() 128 + } 129 + 130 return obj 131 132 133 class TypedObject: 134 """A typed object in the ATProto data model. 135 + 136 This class represents an object with a $type field in the ATProto data model. 137 + 138 Attributes: 139 type: The type of the object. 140 data: The data associated with the object. 141 """ 142 + 143 def __init__(self, type_name: str, data: Dict[str, Any]) -> None: 144 """Initialize a typed object. 145 + 146 Args: 147 type_name: The type of the object. 148 data: The data associated with the object. 149 """ 150 self.type_name = type_name 151 self.data = data 152 + 153 def __repr__(self) -> str: 154 """Return a string representation of the typed object. 155 + 156 Returns: 157 A string representation of the typed object. 158 """ 159 return f"TypedObject(type_name={self.type_name!r}, data={self.data!r})" 160 + 161 def __eq__(self, other: Any) -> bool: 162 """Check if two typed objects are equal. 163 + 164 Args: 165 other: The object to compare with. 166 + 167 Returns: 168 True if the objects are equal, False otherwise. 169 """ 170 if not isinstance(other, TypedObject): 171 return False 172 return self.type_name == other.type_name and self.data == other.data 173 + 174 def __atproto_json_encode__(self) -> Dict[str, Any]: 175 """Encode the typed object to a JSON-serializable format. 176 + 177 Returns: 178 A JSON-serializable representation of the typed object. 179 """ 180 result = {"$type": self.type_name} 181 result.update(self.data) 182 + return result

+10 -10

src/atpasser/data/encoder.py

··· 13 14 class JsonEncoder(json.JSONEncoder): 15 """A JSON encoder that supports ATProto data types. 16 - 17 This encoder extends the standard JSON encoder to handle ATProto-specific 18 data types, including bytes, CID links, and typed objects. 19 - 20 Attributes: 21 encoding (str): The encoding to use for string serialization. 22 type_processor_registry: Registry for type-specific processors. 23 """ 24 - 25 def __init__( 26 self, 27 *, 28 encoding: str = "utf-8", 29 type_processor_registry: Optional[Any] = None, 30 - **kwargs: Any 31 ) -> None: 32 """Initialize the JSON encoder. 33 - 34 Args: 35 encoding: The encoding to use for string serialization. 36 type_processor_registry: Registry for type-specific processors. ··· 39 super().__init__(**kwargs) 40 self.encoding = encoding 41 self.type_processor_registry = type_processor_registry 42 - 43 def default(self, o: Any) -> Any: 44 """Convert an object to a serializable format. 45 - 46 Args: 47 o: The object to serialize. 48 - 49 Returns: 50 A serializable representation of the object. 51 - 52 Raises: 53 TypeError: If the object is not serializable. 54 """ ··· 79 return [self.default(item) for item in o] 80 else: 81 # Use the parent class for other types 82 - return super().default(o)

··· 13 14 class JsonEncoder(json.JSONEncoder): 15 """A JSON encoder that supports ATProto data types. 16 + 17 This encoder extends the standard JSON encoder to handle ATProto-specific 18 data types, including bytes, CID links, and typed objects. 19 + 20 Attributes: 21 encoding (str): The encoding to use for string serialization. 22 type_processor_registry: Registry for type-specific processors. 23 """ 24 + 25 def __init__( 26 self, 27 *, 28 encoding: str = "utf-8", 29 type_processor_registry: Optional[Any] = None, 30 + **kwargs: Any, 31 ) -> None: 32 """Initialize the JSON encoder. 33 + 34 Args: 35 encoding: The encoding to use for string serialization. 36 type_processor_registry: Registry for type-specific processors. ··· 39 super().__init__(**kwargs) 40 self.encoding = encoding 41 self.type_processor_registry = type_processor_registry 42 + 43 def default(self, o: Any) -> Any: 44 """Convert an object to a serializable format. 45 + 46 Args: 47 o: The object to serialize. 48 + 49 Returns: 50 A serializable representation of the object. 51 + 52 Raises: 53 TypeError: If the object is not serializable. 54 """ ··· 79 return [self.default(item) for item in o] 80 else: 81 # Use the parent class for other types 82 + return super().default(o)

+51 -46

src/atpasser/data/hooks.py

··· 9 from typing import Any, Callable, Dict, Optional, TypeVar, Union 10 11 # Type variable for the decorated function 12 - F = TypeVar('F', bound=Callable[..., Any]) 13 14 15 class TypeHookRegistry: 16 """Registry for type-specific hooks in the ATProto JSON decoder. 17 - 18 This class maintains a registry of type-specific hooks that can be used 19 to customize the decoding of objects with $type keys in the ATProto data model. 20 - 21 Attributes: 22 _handlers: Dictionary mapping type names to handler functions. 23 """ 24 - 25 def __init__(self) -> None: 26 """Initialize the type hook registry.""" 27 self._handlers: Dict[str, Callable[[Dict[str, Any]], Any]] = {} 28 - 29 def register(self, type_name: str) -> Callable[[F], F]: 30 """Register a type handler function. 31 - 32 This method can be used as a decorator to register a function as a handler 33 for a specific type. 34 - 35 Args: 36 type_name: The name of the type to handle. 37 - 38 Returns: 39 A decorator function that registers the decorated function as a handler. 40 - 41 Example: 42 >>> registry = TypeHookRegistry() 43 - >>> 44 >>> @registry.register("app.bsky.feed.post") 45 ... def handle_post(data: Dict[str, Any]) -> Any: 46 ... return Post(**data) 47 """ 0 48 def decorator(func: F) -> F: 49 self._handlers[type_name] = func 50 return func 51 - 52 return decorator 53 - 54 - def register_handler(self, type_name: str, handler: Callable[[Dict[str, Any]], Any]) -> None: 0 0 55 """Register a type handler function directly. 56 - 57 Args: 58 type_name: The name of the type to handle. 59 handler: The function to call when decoding objects of this type. 60 - 61 Example: 62 >>> registry = TypeHookRegistry() 63 - >>> 64 >>> def handle_post(data: Dict[str, Any]) -> Any: 65 ... return Post(**data) 66 - >>> 67 >>> registry.register_handler("app.bsky.feed.post", handle_post) 68 """ 69 self._handlers[type_name] = handler 70 - 71 def unregister(self, type_name: str) -> None: 72 """Unregister a type handler function. 73 - 74 Args: 75 type_name: The name of the type to unregister. 76 """ 77 if type_name in self._handlers: 78 del self._handlers[type_name] 79 - 80 def get_handler(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 81 """Get the handler function for a specific type. 82 - 83 Args: 84 type_name: The name of the type to get the handler for. 85 - 86 Returns: 87 The handler function for the specified type, or None if no handler 88 is registered. 89 """ 90 return self._handlers.get(type_name) 91 - 92 def has_handler(self, type_name: str) -> bool: 93 """Check if a handler is registered for a specific type. 94 - 95 Args: 96 type_name: The name of the type to check. 97 - 98 Returns: 99 True if a handler is registered for the specified type, False otherwise. 100 """ 101 return type_name in self._handlers 102 - 103 def clear(self) -> None: 104 """Clear all registered handlers.""" 105 self._handlers.clear() 106 - 107 def get_registered_types(self) -> set: 108 """Get the set of all registered type names. 109 - 110 Returns: 111 A set of all registered type names. 112 """ ··· 119 120 def type_handler(type_name: str) -> Callable[[F], F]: 121 """Register a global type handler function. 122 - 123 This decorator registers a function as a global handler for a specific type 124 in the ATProto data model. 125 - 126 Args: 127 type_name: The name of the type to handle. 128 - 129 Returns: 130 A decorator function that registers the decorated function as a handler. 131 - 132 Example: 133 >>> @type_handler("app.bsky.feed.post") 134 ... def handle_post(data: Dict[str, Any]) -> Any: ··· 139 140 def get_global_registry() -> TypeHookRegistry: 141 """Get the global type hook registry. 142 - 143 Returns: 144 The global TypeHookRegistry instance. 145 """ 146 return _global_registry 147 148 149 - def register_type_handler(type_name: str, handler: Callable[[Dict[str, Any]], Any]) -> None: 0 0 150 """Register a global type handler function directly. 151 - 152 Args: 153 type_name: The name of the type to handle. 154 handler: The function to call when decoding objects of this type. 155 - 156 Example: 157 >>> def handle_post(data: Dict[str, Any]) -> Any: 158 ... return Post(**data) 159 - >>> 160 >>> register_type_handler("app.bsky.feed.post", handle_post) 161 """ 162 _global_registry.register_handler(type_name, handler) ··· 164 165 def unregister_type_handler(type_name: str) -> None: 166 """Unregister a global type handler function. 167 - 168 Args: 169 type_name: The name of the type to unregister. 170 """ ··· 173 174 def get_type_handler(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 175 """Get the global handler function for a specific type. 176 - 177 Args: 178 type_name: The name of the type to get the handler for. 179 - 180 Returns: 181 The handler function for the specified type, or None if no handler 182 is registered. ··· 186 187 def has_type_handler(type_name: str) -> bool: 188 """Check if a global handler is registered for a specific type. 189 - 190 Args: 191 type_name: The name of the type to check. 192 - 193 Returns: 194 True if a handler is registered for the specified type, False otherwise. 195 """ ··· 203 204 def get_registered_types() -> set: 205 """Get the set of all globally registered type names. 206 - 207 Returns: 208 A set of all registered type names. 209 """ ··· 212 213 def create_registry() -> TypeHookRegistry: 214 """Create a new type hook registry. 215 - 216 This function creates a new, independent registry that can be used 217 instead of the global registry. 218 - 219 Returns: 220 A new TypeHookRegistry instance. 221 """ 222 - return TypeHookRegistry()

··· 9 from typing import Any, Callable, Dict, Optional, TypeVar, Union 10 11 # Type variable for the decorated function 12 + F = TypeVar("F", bound=Callable[..., Any]) 13 14 15 class TypeHookRegistry: 16 """Registry for type-specific hooks in the ATProto JSON decoder. 17 + 18 This class maintains a registry of type-specific hooks that can be used 19 to customize the decoding of objects with $type keys in the ATProto data model. 20 + 21 Attributes: 22 _handlers: Dictionary mapping type names to handler functions. 23 """ 24 + 25 def __init__(self) -> None: 26 """Initialize the type hook registry.""" 27 self._handlers: Dict[str, Callable[[Dict[str, Any]], Any]] = {} 28 + 29 def register(self, type_name: str) -> Callable[[F], F]: 30 """Register a type handler function. 31 + 32 This method can be used as a decorator to register a function as a handler 33 for a specific type. 34 + 35 Args: 36 type_name: The name of the type to handle. 37 + 38 Returns: 39 A decorator function that registers the decorated function as a handler. 40 + 41 Example: 42 >>> registry = TypeHookRegistry() 43 + >>> 44 >>> @registry.register("app.bsky.feed.post") 45 ... def handle_post(data: Dict[str, Any]) -> Any: 46 ... return Post(**data) 47 """ 48 + 49 def decorator(func: F) -> F: 50 self._handlers[type_name] = func 51 return func 52 + 53 return decorator 54 + 55 + def register_handler( 56 + self, type_name: str, handler: Callable[[Dict[str, Any]], Any] 57 + ) -> None: 58 """Register a type handler function directly. 59 + 60 Args: 61 type_name: The name of the type to handle. 62 handler: The function to call when decoding objects of this type. 63 + 64 Example: 65 >>> registry = TypeHookRegistry() 66 + >>> 67 >>> def handle_post(data: Dict[str, Any]) -> Any: 68 ... return Post(**data) 69 + >>> 70 >>> registry.register_handler("app.bsky.feed.post", handle_post) 71 """ 72 self._handlers[type_name] = handler 73 + 74 def unregister(self, type_name: str) -> None: 75 """Unregister a type handler function. 76 + 77 Args: 78 type_name: The name of the type to unregister. 79 """ 80 if type_name in self._handlers: 81 del self._handlers[type_name] 82 + 83 def get_handler(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 84 """Get the handler function for a specific type. 85 + 86 Args: 87 type_name: The name of the type to get the handler for. 88 + 89 Returns: 90 The handler function for the specified type, or None if no handler 91 is registered. 92 """ 93 return self._handlers.get(type_name) 94 + 95 def has_handler(self, type_name: str) -> bool: 96 """Check if a handler is registered for a specific type. 97 + 98 Args: 99 type_name: The name of the type to check. 100 + 101 Returns: 102 True if a handler is registered for the specified type, False otherwise. 103 """ 104 return type_name in self._handlers 105 + 106 def clear(self) -> None: 107 """Clear all registered handlers.""" 108 self._handlers.clear() 109 + 110 def get_registered_types(self) -> set: 111 """Get the set of all registered type names. 112 + 113 Returns: 114 A set of all registered type names. 115 """ ··· 122 123 def type_handler(type_name: str) -> Callable[[F], F]: 124 """Register a global type handler function. 125 + 126 This decorator registers a function as a global handler for a specific type 127 in the ATProto data model. 128 + 129 Args: 130 type_name: The name of the type to handle. 131 + 132 Returns: 133 A decorator function that registers the decorated function as a handler. 134 + 135 Example: 136 >>> @type_handler("app.bsky.feed.post") 137 ... def handle_post(data: Dict[str, Any]) -> Any: ··· 142 143 def get_global_registry() -> TypeHookRegistry: 144 """Get the global type hook registry. 145 + 146 Returns: 147 The global TypeHookRegistry instance. 148 """ 149 return _global_registry 150 151 152 + def register_type_handler( 153 + type_name: str, handler: Callable[[Dict[str, Any]], Any] 154 + ) -> None: 155 """Register a global type handler function directly. 156 + 157 Args: 158 type_name: The name of the type to handle. 159 handler: The function to call when decoding objects of this type. 160 + 161 Example: 162 >>> def handle_post(data: Dict[str, Any]) -> Any: 163 ... return Post(**data) 164 + >>> 165 >>> register_type_handler("app.bsky.feed.post", handle_post) 166 """ 167 _global_registry.register_handler(type_name, handler) ··· 169 170 def unregister_type_handler(type_name: str) -> None: 171 """Unregister a global type handler function. 172 + 173 Args: 174 type_name: The name of the type to unregister. 175 """ ··· 178 179 def get_type_handler(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 180 """Get the global handler function for a specific type. 181 + 182 Args: 183 type_name: The name of the type to get the handler for. 184 + 185 Returns: 186 The handler function for the specified type, or None if no handler 187 is registered. ··· 191 192 def has_type_handler(type_name: str) -> bool: 193 """Check if a global handler is registered for a specific type. 194 + 195 Args: 196 type_name: The name of the type to check. 197 + 198 Returns: 199 True if a handler is registered for the specified type, False otherwise. 200 """ ··· 208 209 def get_registered_types() -> set: 210 """Get the set of all globally registered type names. 211 + 212 Returns: 213 A set of all registered type names. 214 """ ··· 217 218 def create_registry() -> TypeHookRegistry: 219 """Create a new type hook registry. 220 + 221 This function creates a new, independent registry that can be used 222 instead of the global registry. 223 + 224 Returns: 225 A new TypeHookRegistry instance. 226 """ 227 + return TypeHookRegistry()

+120 -114

src/atpasser/data/types.py

··· 10 from .hooks import TypeHookRegistry 11 12 # Type variable for the decorated class 13 - T = TypeVar('T') 14 15 16 class TypeProcessor: 17 """A type processor for ATProto JSON objects. 18 - 19 This class represents a processor for a specific type in the ATProto data model. 20 It contains information about how to convert JSON data to Python objects and 21 vice versa. 22 - 23 Attributes: 24 type_name: The name of the type this processor handles. 25 decoder: The function to decode JSON data to a Python object. 26 encoder: The function to encode a Python object to JSON data. 27 priority: The priority of this processor (higher values = higher priority). 28 """ 29 - 30 def __init__( 31 self, 32 type_name: str, 33 decoder: Optional[Callable[[Dict[str, Any]], Any]] = None, 34 encoder: Optional[Callable[[Any], Dict[str, Any]]] = None, 35 - priority: int = 0 36 ) -> None: 37 """Initialize a type processor. 38 - 39 Args: 40 type_name: The name of the type this processor handles. 41 decoder: The function to decode JSON data to a Python object. ··· 46 self.decoder = decoder 47 self.encoder = encoder 48 self.priority = priority 49 - 50 def decode(self, data: Dict[str, Any]) -> Any: 51 """Decode JSON data to a Python object. 52 - 53 Args: 54 data: The JSON data to decode. 55 - 56 Returns: 57 The decoded Python object. 58 - 59 Raises: 60 ValueError: If no decoder is registered. 61 """ 62 if self.decoder is None: 63 raise ValueError(f"No decoder registered for type {self.type_name}") 64 return self.decoder(data) 65 - 66 def encode(self, obj: Any) -> Dict[str, Any]: 67 """Encode a Python object to JSON data. 68 - 69 Args: 70 obj: The Python object to encode. 71 - 72 Returns: 73 The encoded JSON data. 74 - 75 Raises: 76 ValueError: If no encoder is registered. 77 """ ··· 82 83 class TypeProcessorRegistry: 84 """Registry for type processors in the ATProto JSON decoder. 85 - 86 This class maintains a registry of type processors that can be used 87 to customize the encoding and decoding of objects with $type keys in 88 the ATProto data model. 89 - 90 Attributes: 91 _processors: Dictionary mapping type names to processor lists. 92 """ 93 - 94 def __init__(self) -> None: 95 """Initialize the type processor registry.""" 96 self._processors: Dict[str, List[TypeProcessor]] = {} 97 - 98 def register_processor(self, processor: TypeProcessor) -> None: 99 """Register a type processor. 100 - 101 Args: 102 processor: The type processor to register. 103 """ 104 if processor.type_name not in self._processors: 105 self._processors[processor.type_name] = [] 106 - 107 self._processors[processor.type_name].append(processor) 108 # Sort processors by priority (descending) 109 - self._processors[processor.type_name].sort(key=lambda p: p.priority, reverse=True) 110 - 0 0 111 def register( 112 - self, 113 - type_name: str, 114 - priority: int = 0 115 ) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]: 116 """Register a type decoder function. 117 - 118 This method can be used as a decorator to register a function as a decoder 119 for a specific type. 120 - 121 Args: 122 type_name: The name of the type to handle. 123 priority: The priority of this processor (higher values = higher priority). 124 - 125 Returns: 126 A decorator function that registers the decorated function as a decoder. 127 - 128 Example: 129 >>> registry = TypeProcessorRegistry() 130 - >>> 131 >>> @registry.register("app.bsky.feed.post", priority=10) 132 ... def decode_post(data: Dict[str, Any]) -> Any: 133 ... return Post(**data) 134 """ 135 - def decorator(func: Callable[[Dict[str, Any]], Any]) -> Callable[[Dict[str, Any]], Any]: 0 0 0 136 processor = TypeProcessor(type_name, decoder=func, priority=priority) 137 self.register_processor(processor) 138 return func 139 - 140 return decorator 141 - 142 def register_encoder( 143 - self, 144 - type_name: str, 145 - priority: int = 0 146 ) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]: 147 """Register a type encoder function. 148 - 149 This method can be used as a decorator to register a function as an encoder 150 for a specific type. 151 - 152 Args: 153 type_name: The name of the type to handle. 154 priority: The priority of this processor (higher values = higher priority). 155 - 156 Returns: 157 A decorator function that registers the decorated function as an encoder. 158 - 159 Example: 160 >>> registry = TypeProcessorRegistry() 161 - >>> 162 >>> @registry.register_encoder("app.bsky.feed.post", priority=10) 163 ... def encode_post(post: Post) -> Dict[str, Any]: 164 ... return {"text": post.text, "createdAt": post.created_at} 165 """ 166 - def decorator(func: Callable[[Any], Dict[str, Any]]) -> Callable[[Any], Dict[str, Any]]: 0 0 0 167 # Check if a processor for this type already exists 168 if type_name in self._processors: 169 for processor in self._processors[type_name]: ··· 173 break 174 else: 175 # No decoder found, create a new processor 176 - processor = TypeProcessor(type_name, encoder=func, priority=priority) 0 0 177 self.register_processor(processor) 178 else: 179 # No processor exists, create a new one 180 processor = TypeProcessor(type_name, encoder=func, priority=priority) 181 self.register_processor(processor) 182 - 183 return func 184 - 185 return decorator 186 - 187 def register_class( 188 - self, 189 - type_name: str, 190 - priority: int = 0 191 ) -> Callable[[Type[T]], Type[T]]: 192 """Register a class for both encoding and decoding. 193 - 194 This method can be used as a decorator to register a class for both 195 encoding and decoding of a specific type. 196 - 197 The class must have a class method `from_json` that takes a dictionary 198 and returns an instance of the class, and an instance method `to_json` 199 that returns a dictionary. 200 - 201 Args: 202 type_name: The name of the type to handle. 203 priority: The priority of this processor (higher values = higher priority). 204 - 205 Returns: 206 A decorator function that registers the decorated class. 207 - 208 Example: 209 >>> registry = TypeProcessorRegistry() 210 - >>> 211 >>> @registry.register_class("app.bsky.feed.post", priority=10) 212 ... class Post: 213 ... def __init__(self, text: str, created_at: str) -> None: 214 ... self.text = text 215 ... self.created_at = created_at 216 - ... 217 ... @classmethod 218 ... def from_json(cls, data: Dict[str, Any]) -> "Post": 219 ... return cls(data["text"], data["createdAt"]) 220 - ... 221 ... def to_json(self) -> Dict[str, Any]: 222 ... return {"text": self.text, "createdAt": self.created_at} 223 """ 0 224 def decorator(cls: Type[T]) -> Type[T]: 225 # Create decoder from class method 226 if hasattr(cls, "from_json"): ··· 232 # Create a decoder that passes the data as keyword arguments 233 decoder = lambda data: cls(**data) 234 else: 235 - raise ValueError(f"Class {cls.__name__} has no from_json method or compatible __init__") 236 - 0 0 237 # Create encoder from instance method 238 if hasattr(cls, "to_json"): 239 encoder = lambda obj: obj.to_json() 240 else: 241 raise ValueError(f"Class {cls.__name__} has no to_json method") 242 - 243 # Register the processor 244 - processor = TypeProcessor(type_name, decoder=decoder, encoder=encoder, priority=priority) 0 0 245 self.register_processor(processor) 246 - 247 return cls 248 - 249 return decorator 250 - 251 def unregister(self, type_name: str, priority: Optional[int] = None) -> None: 252 """Unregister type processors. 253 - 254 Args: 255 type_name: The name of the type to unregister. 256 priority: If specified, only unregister processors with this priority. ··· 264 else: 265 # Remove all processors for this type 266 del self._processors[type_name] 267 - 268 def get_decoder(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 269 """Get the decoder function for a specific type. 270 - 271 Args: 272 type_name: The name of the type to get the decoder for. 273 - 274 Returns: 275 The decoder function for the specified type, or None if no decoder 276 is registered. ··· 279 # Return the decoder of the highest priority processor 280 return self._processors[type_name][0].decoder 281 return None 282 - 283 def get_encoder(self, type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]: 284 """Get the encoder function for a specific type. 285 - 286 Args: 287 type_name: The name of the type to get the encoder for. 288 - 289 Returns: 290 The encoder function for the specified type, or None if no encoder 291 is registered. ··· 294 # Return the encoder of the highest priority processor 295 return self._processors[type_name][0].encoder 296 return None 297 - 298 def has_processor(self, type_name: str) -> bool: 299 """Check if a processor is registered for a specific type. 300 - 301 Args: 302 type_name: The name of the type to check. 303 - 304 Returns: 305 True if a processor is registered for the specified type, False otherwise. 306 """ 307 return type_name in self._processors and bool(self._processors[type_name]) 308 - 309 def clear(self) -> None: 310 """Clear all registered processors.""" 311 self._processors.clear() 312 - 313 def get_registered_types(self) -> set: 314 """Get the set of all registered type names. 315 - 316 Returns: 317 A set of all registered type names. 318 """ 319 return set(self._processors.keys()) 320 - 321 def to_hook_registry(self) -> TypeHookRegistry: 322 """Convert this processor registry to a hook registry. 323 - 324 This method creates a TypeHookRegistry that uses the decoders from 325 this processor registry. 326 - 327 Returns: 328 A TypeHookRegistry with the same decoders as this processor registry. 329 """ 330 hook_registry = TypeHookRegistry() 331 - 332 for type_name, processors in self._processors.items(): 333 if processors and processors[0].decoder is not None: 334 hook_registry.register_handler(type_name, processors[0].decoder) 335 - 336 return hook_registry 337 338 ··· 341 342 343 def register_type( 344 - type_name: str, 345 - priority: int = 0 346 ) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]: 347 """Register a global type decoder function. 348 - 349 This decorator registers a function as a global decoder for a specific type 350 in the ATProto data model. 351 - 352 Args: 353 type_name: The name of the type to handle. 354 priority: The priority of this processor (higher values = higher priority). 355 - 356 Returns: 357 A decorator function that registers the decorated function as a decoder. 358 - 359 Example: 360 >>> @register_type("app.bsky.feed.post", priority=10) 361 ... def decode_post(data: Dict[str, Any]) -> Any: ··· 366 367 def get_global_processor_registry() -> TypeProcessorRegistry: 368 """Get the global type processor registry. 369 - 370 Returns: 371 The global TypeProcessorRegistry instance. 372 """ ··· 374 375 376 def register_type_encoder( 377 - type_name: str, 378 - priority: int = 0 379 ) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]: 380 """Register a global type encoder function. 381 - 382 This decorator registers a function as a global encoder for a specific type 383 in the ATProto data model. 384 - 385 Args: 386 type_name: The name of the type to handle. 387 priority: The priority of this processor (higher values = higher priority). 388 - 389 Returns: 390 A decorator function that registers the decorated function as an encoder. 391 - 392 Example: 393 >>> @register_type_encoder("app.bsky.feed.post", priority=10) 394 ... def encode_post(post: Post) -> Dict[str, Any]: ··· 398 399 400 def register_type_class( 401 - type_name: str, 402 - priority: int = 0 403 ) -> Callable[[Type[T]], Type[T]]: 404 """Register a class for both global encoding and decoding. 405 - 406 This decorator registers a class for both encoding and decoding of a specific type 407 in the ATProto data model. 408 - 409 Args: 410 type_name: The name of the type to handle. 411 priority: The priority of this processor (higher values = higher priority). 412 - 413 Returns: 414 A decorator function that registers the decorated class. 415 - 416 Example: 417 >>> @register_type_class("app.bsky.feed.post", priority=10) 418 ... class Post: 419 ... def __init__(self, text: str, created_at: str) -> None: 420 ... self.text = text 421 ... self.created_at = created_at 422 - ... 423 ... @classmethod 424 ... def from_json(cls, data: Dict[str, Any]) -> "Post": 425 ... return cls(data["text"], data["createdAt"]) 426 - ... 427 ... def to_json(self) -> Dict[str, Any]: 428 ... return {"text": self.text, "createdAt": self.created_at} 429 """ ··· 432 433 def unregister_type(type_name: str, priority: Optional[int] = None) -> None: 434 """Unregister global type processors. 435 - 436 Args: 437 type_name: The name of the type to unregister. 438 priority: If specified, only unregister processors with this priority. ··· 442 443 def get_type_decoder(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 444 """Get the global decoder function for a specific type. 445 - 446 Args: 447 type_name: The name of the type to get the decoder for. 448 - 449 Returns: 450 The decoder function for the specified type, or None if no decoder 451 is registered. ··· 455 456 def get_type_encoder(type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]: 457 """Get the global encoder function for a specific type. 458 - 459 Args: 460 type_name: The name of the type to get the encoder for. 461 - 462 Returns: 463 The encoder function for the specified type, or None if no encoder 464 is registered. ··· 468 469 def has_type_processor(type_name: str) -> bool: 470 """Check if a global processor is registered for a specific type. 471 - 472 Args: 473 type_name: The name of the type to check. 474 - 475 Returns: 476 True if a processor is registered for the specified type, False otherwise. 477 """ ··· 485 486 def get_registered_types() -> set: 487 """Get the set of all globally registered type names. 488 - 489 Returns: 490 A set of all registered type names. 491 """ ··· 494 495 def create_processor_registry() -> TypeProcessorRegistry: 496 """Create a new type processor registry. 497 - 498 This function creates a new, independent registry that can be used 499 instead of the global registry. 500 - 501 Returns: 502 A new TypeProcessorRegistry instance. 503 """ 504 - return TypeProcessorRegistry()

··· 10 from .hooks import TypeHookRegistry 11 12 # Type variable for the decorated class 13 + T = TypeVar("T") 14 15 16 class TypeProcessor: 17 """A type processor for ATProto JSON objects. 18 + 19 This class represents a processor for a specific type in the ATProto data model. 20 It contains information about how to convert JSON data to Python objects and 21 vice versa. 22 + 23 Attributes: 24 type_name: The name of the type this processor handles. 25 decoder: The function to decode JSON data to a Python object. 26 encoder: The function to encode a Python object to JSON data. 27 priority: The priority of this processor (higher values = higher priority). 28 """ 29 + 30 def __init__( 31 self, 32 type_name: str, 33 decoder: Optional[Callable[[Dict[str, Any]], Any]] = None, 34 encoder: Optional[Callable[[Any], Dict[str, Any]]] = None, 35 + priority: int = 0, 36 ) -> None: 37 """Initialize a type processor. 38 + 39 Args: 40 type_name: The name of the type this processor handles. 41 decoder: The function to decode JSON data to a Python object. ··· 46 self.decoder = decoder 47 self.encoder = encoder 48 self.priority = priority 49 + 50 def decode(self, data: Dict[str, Any]) -> Any: 51 """Decode JSON data to a Python object. 52 + 53 Args: 54 data: The JSON data to decode. 55 + 56 Returns: 57 The decoded Python object. 58 + 59 Raises: 60 ValueError: If no decoder is registered. 61 """ 62 if self.decoder is None: 63 raise ValueError(f"No decoder registered for type {self.type_name}") 64 return self.decoder(data) 65 + 66 def encode(self, obj: Any) -> Dict[str, Any]: 67 """Encode a Python object to JSON data. 68 + 69 Args: 70 obj: The Python object to encode. 71 + 72 Returns: 73 The encoded JSON data. 74 + 75 Raises: 76 ValueError: If no encoder is registered. 77 """ ··· 82 83 class TypeProcessorRegistry: 84 """Registry for type processors in the ATProto JSON decoder. 85 + 86 This class maintains a registry of type processors that can be used 87 to customize the encoding and decoding of objects with $type keys in 88 the ATProto data model. 89 + 90 Attributes: 91 _processors: Dictionary mapping type names to processor lists. 92 """ 93 + 94 def __init__(self) -> None: 95 """Initialize the type processor registry.""" 96 self._processors: Dict[str, List[TypeProcessor]] = {} 97 + 98 def register_processor(self, processor: TypeProcessor) -> None: 99 """Register a type processor. 100 + 101 Args: 102 processor: The type processor to register. 103 """ 104 if processor.type_name not in self._processors: 105 self._processors[processor.type_name] = [] 106 + 107 self._processors[processor.type_name].append(processor) 108 # Sort processors by priority (descending) 109 + self._processors[processor.type_name].sort( 110 + key=lambda p: p.priority, reverse=True 111 + ) 112 + 113 def register( 114 + self, type_name: str, priority: int = 0 0 0 115 ) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]: 116 """Register a type decoder function. 117 + 118 This method can be used as a decorator to register a function as a decoder 119 for a specific type. 120 + 121 Args: 122 type_name: The name of the type to handle. 123 priority: The priority of this processor (higher values = higher priority). 124 + 125 Returns: 126 A decorator function that registers the decorated function as a decoder. 127 + 128 Example: 129 >>> registry = TypeProcessorRegistry() 130 + >>> 131 >>> @registry.register("app.bsky.feed.post", priority=10) 132 ... def decode_post(data: Dict[str, Any]) -> Any: 133 ... return Post(**data) 134 """ 135 + 136 + def decorator( 137 + func: Callable[[Dict[str, Any]], Any], 138 + ) -> Callable[[Dict[str, Any]], Any]: 139 processor = TypeProcessor(type_name, decoder=func, priority=priority) 140 self.register_processor(processor) 141 return func 142 + 143 return decorator 144 + 145 def register_encoder( 146 + self, type_name: str, priority: int = 0 0 0 147 ) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]: 148 """Register a type encoder function. 149 + 150 This method can be used as a decorator to register a function as an encoder 151 for a specific type. 152 + 153 Args: 154 type_name: The name of the type to handle. 155 priority: The priority of this processor (higher values = higher priority). 156 + 157 Returns: 158 A decorator function that registers the decorated function as an encoder. 159 + 160 Example: 161 >>> registry = TypeProcessorRegistry() 162 + >>> 163 >>> @registry.register_encoder("app.bsky.feed.post", priority=10) 164 ... def encode_post(post: Post) -> Dict[str, Any]: 165 ... return {"text": post.text, "createdAt": post.created_at} 166 """ 167 + 168 + def decorator( 169 + func: Callable[[Any], Dict[str, Any]], 170 + ) -> Callable[[Any], Dict[str, Any]]: 171 # Check if a processor for this type already exists 172 if type_name in self._processors: 173 for processor in self._processors[type_name]: ··· 177 break 178 else: 179 # No decoder found, create a new processor 180 + processor = TypeProcessor( 181 + type_name, encoder=func, priority=priority 182 + ) 183 self.register_processor(processor) 184 else: 185 # No processor exists, create a new one 186 processor = TypeProcessor(type_name, encoder=func, priority=priority) 187 self.register_processor(processor) 188 + 189 return func 190 + 191 return decorator 192 + 193 def register_class( 194 + self, type_name: str, priority: int = 0 0 0 195 ) -> Callable[[Type[T]], Type[T]]: 196 """Register a class for both encoding and decoding. 197 + 198 This method can be used as a decorator to register a class for both 199 encoding and decoding of a specific type. 200 + 201 The class must have a class method `from_json` that takes a dictionary 202 and returns an instance of the class, and an instance method `to_json` 203 that returns a dictionary. 204 + 205 Args: 206 type_name: The name of the type to handle. 207 priority: The priority of this processor (higher values = higher priority). 208 + 209 Returns: 210 A decorator function that registers the decorated class. 211 + 212 Example: 213 >>> registry = TypeProcessorRegistry() 214 + >>> 215 >>> @registry.register_class("app.bsky.feed.post", priority=10) 216 ... class Post: 217 ... def __init__(self, text: str, created_at: str) -> None: 218 ... self.text = text 219 ... self.created_at = created_at 220 + ... 221 ... @classmethod 222 ... def from_json(cls, data: Dict[str, Any]) -> "Post": 223 ... return cls(data["text"], data["createdAt"]) 224 + ... 225 ... def to_json(self) -> Dict[str, Any]: 226 ... return {"text": self.text, "createdAt": self.created_at} 227 """ 228 + 229 def decorator(cls: Type[T]) -> Type[T]: 230 # Create decoder from class method 231 if hasattr(cls, "from_json"): ··· 237 # Create a decoder that passes the data as keyword arguments 238 decoder = lambda data: cls(**data) 239 else: 240 + raise ValueError( 241 + f"Class {cls.__name__} has no from_json method or compatible __init__" 242 + ) 243 + 244 # Create encoder from instance method 245 if hasattr(cls, "to_json"): 246 encoder = lambda obj: obj.to_json() 247 else: 248 raise ValueError(f"Class {cls.__name__} has no to_json method") 249 + 250 # Register the processor 251 + processor = TypeProcessor( 252 + type_name, decoder=decoder, encoder=encoder, priority=priority 253 + ) 254 self.register_processor(processor) 255 + 256 return cls 257 + 258 return decorator 259 + 260 def unregister(self, type_name: str, priority: Optional[int] = None) -> None: 261 """Unregister type processors. 262 + 263 Args: 264 type_name: The name of the type to unregister. 265 priority: If specified, only unregister processors with this priority. ··· 273 else: 274 # Remove all processors for this type 275 del self._processors[type_name] 276 + 277 def get_decoder(self, type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 278 """Get the decoder function for a specific type. 279 + 280 Args: 281 type_name: The name of the type to get the decoder for. 282 + 283 Returns: 284 The decoder function for the specified type, or None if no decoder 285 is registered. ··· 288 # Return the decoder of the highest priority processor 289 return self._processors[type_name][0].decoder 290 return None 291 + 292 def get_encoder(self, type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]: 293 """Get the encoder function for a specific type. 294 + 295 Args: 296 type_name: The name of the type to get the encoder for. 297 + 298 Returns: 299 The encoder function for the specified type, or None if no encoder 300 is registered. ··· 303 # Return the encoder of the highest priority processor 304 return self._processors[type_name][0].encoder 305 return None 306 + 307 def has_processor(self, type_name: str) -> bool: 308 """Check if a processor is registered for a specific type. 309 + 310 Args: 311 type_name: The name of the type to check. 312 + 313 Returns: 314 True if a processor is registered for the specified type, False otherwise. 315 """ 316 return type_name in self._processors and bool(self._processors[type_name]) 317 + 318 def clear(self) -> None: 319 """Clear all registered processors.""" 320 self._processors.clear() 321 + 322 def get_registered_types(self) -> set: 323 """Get the set of all registered type names. 324 + 325 Returns: 326 A set of all registered type names. 327 """ 328 return set(self._processors.keys()) 329 + 330 def to_hook_registry(self) -> TypeHookRegistry: 331 """Convert this processor registry to a hook registry. 332 + 333 This method creates a TypeHookRegistry that uses the decoders from 334 this processor registry. 335 + 336 Returns: 337 A TypeHookRegistry with the same decoders as this processor registry. 338 """ 339 hook_registry = TypeHookRegistry() 340 + 341 for type_name, processors in self._processors.items(): 342 if processors and processors[0].decoder is not None: 343 hook_registry.register_handler(type_name, processors[0].decoder) 344 + 345 return hook_registry 346 347 ··· 350 351 352 def register_type( 353 + type_name: str, priority: int = 0 0 354 ) -> Callable[[Callable[[Dict[str, Any]], Any]], Callable[[Dict[str, Any]], Any]]: 355 """Register a global type decoder function. 356 + 357 This decorator registers a function as a global decoder for a specific type 358 in the ATProto data model. 359 + 360 Args: 361 type_name: The name of the type to handle. 362 priority: The priority of this processor (higher values = higher priority). 363 + 364 Returns: 365 A decorator function that registers the decorated function as a decoder. 366 + 367 Example: 368 >>> @register_type("app.bsky.feed.post", priority=10) 369 ... def decode_post(data: Dict[str, Any]) -> Any: ··· 374 375 def get_global_processor_registry() -> TypeProcessorRegistry: 376 """Get the global type processor registry. 377 + 378 Returns: 379 The global TypeProcessorRegistry instance. 380 """ ··· 382 383 384 def register_type_encoder( 385 + type_name: str, priority: int = 0 0 386 ) -> Callable[[Callable[[Any], Dict[str, Any]]], Callable[[Any], Dict[str, Any]]]: 387 """Register a global type encoder function. 388 + 389 This decorator registers a function as a global encoder for a specific type 390 in the ATProto data model. 391 + 392 Args: 393 type_name: The name of the type to handle. 394 priority: The priority of this processor (higher values = higher priority). 395 + 396 Returns: 397 A decorator function that registers the decorated function as an encoder. 398 + 399 Example: 400 >>> @register_type_encoder("app.bsky.feed.post", priority=10) 401 ... def encode_post(post: Post) -> Dict[str, Any]: ··· 405 406 407 def register_type_class( 408 + type_name: str, priority: int = 0 0 409 ) -> Callable[[Type[T]], Type[T]]: 410 """Register a class for both global encoding and decoding. 411 + 412 This decorator registers a class for both encoding and decoding of a specific type 413 in the ATProto data model. 414 + 415 Args: 416 type_name: The name of the type to handle. 417 priority: The priority of this processor (higher values = higher priority). 418 + 419 Returns: 420 A decorator function that registers the decorated class. 421 + 422 Example: 423 >>> @register_type_class("app.bsky.feed.post", priority=10) 424 ... class Post: 425 ... def __init__(self, text: str, created_at: str) -> None: 426 ... self.text = text 427 ... self.created_at = created_at 428 + ... 429 ... @classmethod 430 ... def from_json(cls, data: Dict[str, Any]) -> "Post": 431 ... return cls(data["text"], data["createdAt"]) 432 + ... 433 ... def to_json(self) -> Dict[str, Any]: 434 ... return {"text": self.text, "createdAt": self.created_at} 435 """ ··· 438 439 def unregister_type(type_name: str, priority: Optional[int] = None) -> None: 440 """Unregister global type processors. 441 + 442 Args: 443 type_name: The name of the type to unregister. 444 priority: If specified, only unregister processors with this priority. ··· 448 449 def get_type_decoder(type_name: str) -> Optional[Callable[[Dict[str, Any]], Any]]: 450 """Get the global decoder function for a specific type. 451 + 452 Args: 453 type_name: The name of the type to get the decoder for. 454 + 455 Returns: 456 The decoder function for the specified type, or None if no decoder 457 is registered. ··· 461 462 def get_type_encoder(type_name: str) -> Optional[Callable[[Any], Dict[str, Any]]]: 463 """Get the global encoder function for a specific type. 464 + 465 Args: 466 type_name: The name of the type to get the encoder for. 467 + 468 Returns: 469 The encoder function for the specified type, or None if no encoder 470 is registered. ··· 474 475 def has_type_processor(type_name: str) -> bool: 476 """Check if a global processor is registered for a specific type. 477 + 478 Args: 479 type_name: The name of the type to check. 480 + 481 Returns: 482 True if a processor is registered for the specified type, False otherwise. 483 """ ··· 491 492 def get_registered_types() -> set: 493 """Get the set of all globally registered type names. 494 + 495 Returns: 496 A set of all registered type names. 497 """ ··· 500 501 def create_processor_registry() -> TypeProcessorRegistry: 502 """Create a new type processor registry. 503 + 504 This function creates a new, independent registry that can be used 505 instead of the global registry. 506 + 507 Returns: 508 A new TypeProcessorRegistry instance. 509 """ 510 + return TypeProcessorRegistry()

+42 -35

src/atpasser/data/wrapper.py

··· 29 sort_keys: bool = False, 30 encoding: str = "utf-8", 31 type_processor_registry: Optional[TypeProcessorRegistry] = None, 32 - **kwargs: Any 33 ) -> None: 34 """Serialize obj as a JSON formatted stream to fp. 35 - 36 This function is similar to json.dump() but supports ATProto-specific 37 data types, including bytes, CID links, and typed objects. 38 - 39 Args: 40 obj: The object to serialize. 41 fp: A file-like object with a write() method. ··· 70 """ 71 if cls is None: 72 cls = JsonEncoder 73 - 74 # Use the global type processor registry if none is provided 75 if type_processor_registry is None: 76 from .types import get_global_processor_registry 0 77 type_processor_registry = get_global_processor_registry() 78 - 79 # Create an encoder instance with the specified encoding and type processor registry 80 - encoder = cls(encoding=encoding, type_processor_registry=type_processor_registry, **kwargs) 81 - 0 0 82 # Use the standard json.dump with our custom encoder 83 json.dump( 84 obj, ··· 92 separators=separators, 93 default=default, 94 sort_keys=sort_keys, 95 - **kwargs 96 ) 97 98 ··· 110 sort_keys: bool = False, 111 encoding: str = "utf-8", 112 type_processor_registry: Optional[TypeProcessorRegistry] = None, 113 - **kwargs: Any 114 ) -> str: 115 """Serialize obj to a JSON formatted string. 116 - 117 This function is similar to json.dumps() but supports ATProto-specific 118 data types, including bytes, CID links, and typed objects. 119 - 120 Args: 121 obj: The object to serialize. 122 skipkeys: If True, dict keys that are not basic types (str, int, float, ··· 147 encoding: The encoding to use for string serialization. 148 type_processor_registry: Registry for type-specific processors. 149 **kwargs: Additional keyword arguments to pass to the JSON encoder. 150 - 151 Returns: 152 A JSON formatted string. 153 """ 154 if cls is None: 155 cls = JsonEncoder 156 - 157 # Create an encoder instance with the specified encoding and type processor registry 158 - encoder = cls(encoding=encoding, type_processor_registry=type_processor_registry, **kwargs) 159 - 0 0 160 # Use the standard json.dumps with our custom encoder 161 return json.dumps( 162 obj, ··· 169 separators=separators, 170 default=default, 171 sort_keys=sort_keys, 172 - **kwargs 173 ) 174 175 ··· 185 type_hook_registry: Optional[TypeHookRegistry] = None, 186 type_processor_registry: Optional[TypeProcessorRegistry] = None, 187 encoding: str = "utf-8", 188 - **kwargs: Any 189 ) -> Any: 190 """Deserialize fp (a .read()-supporting text file or binary file containing 191 a JSON document) to a Python object. 192 - 193 This function is similar to json.load() but supports ATProto-specific 194 data types, including bytes, CID links, and typed objects. 195 - 196 Args: 197 fp: A .read()-supporting text file or binary file containing a JSON document. 198 cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used. ··· 220 type_processor_registry: Registry for type-specific processors. 221 encoding: The encoding to use for string deserialization. 222 **kwargs: Additional keyword arguments to pass to the JSON decoder. 223 - 224 Returns: 225 A Python object. 226 """ 227 if cls is None: 228 cls = JsonDecoder 229 - 230 # Use the global type hook registry if none is provided 231 if type_hook_registry is None and type_processor_registry is None: 232 from .hooks import get_global_registry 0 233 type_hook_registry = get_global_registry() 234 elif type_processor_registry is not None: 235 # Convert the type processor registry to a hook registry 236 type_hook_registry = type_processor_registry.to_hook_registry() 237 - 238 # Create a decoder instance with the specified parameters 239 decoder = cls( 240 object_hook=object_hook, 241 type_hook_registry=type_hook_registry, 242 encoding=encoding, 243 - **kwargs 244 ) 245 - 246 # Use the standard json.load with our custom decoder 247 return json.load( 248 fp, ··· 252 parse_int=parse_int, 253 parse_constant=parse_constant, 254 object_pairs_hook=object_pairs_hook, 255 - **kwargs 256 ) 257 258 ··· 268 type_hook_registry: Optional[TypeHookRegistry] = None, 269 type_processor_registry: Optional[TypeProcessorRegistry] = None, 270 encoding: str = "utf-8", 271 - **kwargs: Any 272 ) -> Any: 273 """Deserialize s (a str, bytes or bytearray instance containing a JSON document) 274 to a Python object. 275 - 276 This function is similar to json.loads() but supports ATProto-specific 277 data types, including bytes, CID links, and typed objects. 278 - 279 Args: 280 s: A str, bytes or bytearray instance containing a JSON document. 281 cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used. ··· 303 type_processor_registry: Registry for type-specific processors. 304 encoding: The encoding to use for string deserialization. 305 **kwargs: Additional keyword arguments to pass to the JSON decoder. 306 - 307 Returns: 308 A Python object. 309 """ 310 if cls is None: 311 cls = JsonDecoder 312 - 313 # Use the global type hook registry if none is provided 314 if type_hook_registry is None and type_processor_registry is None: 315 from .hooks import get_global_registry 0 316 type_hook_registry = get_global_registry() 317 elif type_processor_registry is not None: 318 # Convert the type processor registry to a hook registry 319 type_hook_registry = type_processor_registry.to_hook_registry() 320 - 321 # Create a decoder instance with the specified parameters 322 decoder = cls( 323 object_hook=object_hook, 324 type_hook_registry=type_hook_registry, 325 encoding=encoding, 326 - **kwargs 327 ) 328 - 329 # Use the standard json.loads with our custom decoder 330 return json.loads( 331 s, ··· 335 parse_int=parse_int, 336 parse_constant=parse_constant, 337 object_pairs_hook=object_pairs_hook, 338 - **kwargs 339 - )

··· 29 sort_keys: bool = False, 30 encoding: str = "utf-8", 31 type_processor_registry: Optional[TypeProcessorRegistry] = None, 32 + **kwargs: Any, 33 ) -> None: 34 """Serialize obj as a JSON formatted stream to fp. 35 + 36 This function is similar to json.dump() but supports ATProto-specific 37 data types, including bytes, CID links, and typed objects. 38 + 39 Args: 40 obj: The object to serialize. 41 fp: A file-like object with a write() method. ··· 70 """ 71 if cls is None: 72 cls = JsonEncoder 73 + 74 # Use the global type processor registry if none is provided 75 if type_processor_registry is None: 76 from .types import get_global_processor_registry 77 + 78 type_processor_registry = get_global_processor_registry() 79 + 80 # Create an encoder instance with the specified encoding and type processor registry 81 + encoder = cls( 82 + encoding=encoding, type_processor_registry=type_processor_registry, **kwargs 83 + ) 84 + 85 # Use the standard json.dump with our custom encoder 86 json.dump( 87 obj, ··· 95 separators=separators, 96 default=default, 97 sort_keys=sort_keys, 98 + **kwargs, 99 ) 100 101 ··· 113 sort_keys: bool = False, 114 encoding: str = "utf-8", 115 type_processor_registry: Optional[TypeProcessorRegistry] = None, 116 + **kwargs: Any, 117 ) -> str: 118 """Serialize obj to a JSON formatted string. 119 + 120 This function is similar to json.dumps() but supports ATProto-specific 121 data types, including bytes, CID links, and typed objects. 122 + 123 Args: 124 obj: The object to serialize. 125 skipkeys: If True, dict keys that are not basic types (str, int, float, ··· 150 encoding: The encoding to use for string serialization. 151 type_processor_registry: Registry for type-specific processors. 152 **kwargs: Additional keyword arguments to pass to the JSON encoder. 153 + 154 Returns: 155 A JSON formatted string. 156 """ 157 if cls is None: 158 cls = JsonEncoder 159 + 160 # Create an encoder instance with the specified encoding and type processor registry 161 + encoder = cls( 162 + encoding=encoding, type_processor_registry=type_processor_registry, **kwargs 163 + ) 164 + 165 # Use the standard json.dumps with our custom encoder 166 return json.dumps( 167 obj, ··· 174 separators=separators, 175 default=default, 176 sort_keys=sort_keys, 177 + **kwargs, 178 ) 179 180 ··· 190 type_hook_registry: Optional[TypeHookRegistry] = None, 191 type_processor_registry: Optional[TypeProcessorRegistry] = None, 192 encoding: str = "utf-8", 193 + **kwargs: Any, 194 ) -> Any: 195 """Deserialize fp (a .read()-supporting text file or binary file containing 196 a JSON document) to a Python object. 197 + 198 This function is similar to json.load() but supports ATProto-specific 199 data types, including bytes, CID links, and typed objects. 200 + 201 Args: 202 fp: A .read()-supporting text file or binary file containing a JSON document. 203 cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used. ··· 225 type_processor_registry: Registry for type-specific processors. 226 encoding: The encoding to use for string deserialization. 227 **kwargs: Additional keyword arguments to pass to the JSON decoder. 228 + 229 Returns: 230 A Python object. 231 """ 232 if cls is None: 233 cls = JsonDecoder 234 + 235 # Use the global type hook registry if none is provided 236 if type_hook_registry is None and type_processor_registry is None: 237 from .hooks import get_global_registry 238 + 239 type_hook_registry = get_global_registry() 240 elif type_processor_registry is not None: 241 # Convert the type processor registry to a hook registry 242 type_hook_registry = type_processor_registry.to_hook_registry() 243 + 244 # Create a decoder instance with the specified parameters 245 decoder = cls( 246 object_hook=object_hook, 247 type_hook_registry=type_hook_registry, 248 encoding=encoding, 249 + **kwargs, 250 ) 251 + 252 # Use the standard json.load with our custom decoder 253 return json.load( 254 fp, ··· 258 parse_int=parse_int, 259 parse_constant=parse_constant, 260 object_pairs_hook=object_pairs_hook, 261 + **kwargs, 262 ) 263 264 ··· 274 type_hook_registry: Optional[TypeHookRegistry] = None, 275 type_processor_registry: Optional[TypeProcessorRegistry] = None, 276 encoding: str = "utf-8", 277 + **kwargs: Any, 278 ) -> Any: 279 """Deserialize s (a str, bytes or bytearray instance containing a JSON document) 280 to a Python object. 281 + 282 This function is similar to json.loads() but supports ATProto-specific 283 data types, including bytes, CID links, and typed objects. 284 + 285 Args: 286 s: A str, bytes or bytearray instance containing a JSON document. 287 cls: A custom JSONDecoder subclass. If not specified, JsonDecoder is used. ··· 309 type_processor_registry: Registry for type-specific processors. 310 encoding: The encoding to use for string deserialization. 311 **kwargs: Additional keyword arguments to pass to the JSON decoder. 312 + 313 Returns: 314 A Python object. 315 """ 316 if cls is None: 317 cls = JsonDecoder 318 + 319 # Use the global type hook registry if none is provided 320 if type_hook_registry is None and type_processor_registry is None: 321 from .hooks import get_global_registry 322 + 323 type_hook_registry = get_global_registry() 324 elif type_processor_registry is not None: 325 # Convert the type processor registry to a hook registry 326 type_hook_registry = type_processor_registry.to_hook_registry() 327 + 328 # Create a decoder instance with the specified parameters 329 decoder = cls( 330 object_hook=object_hook, 331 type_hook_registry=type_hook_registry, 332 encoding=encoding, 333 + **kwargs, 334 ) 335 + 336 # Use the standard json.loads with our custom decoder 337 return json.loads( 338 s, ··· 342 parse_int=parse_int, 343 parse_constant=parse_constant, 344 object_pairs_hook=object_pairs_hook, 345 + **kwargs, 346 + )

+1 -1

tests/__init__.py

··· 1 - """Test package for atpasser."""

··· 1 + """Test package for atpasser."""

+1 -1

tests/uri/__init__.py

··· 1 - """Test package for atpasser.uri module."""

··· 1 + """Test package for atpasser.uri module."""

+16 -16

tests/uri/test_did.py

··· 12 """Test creating a DID with a valid did:plc format.""" 13 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 14 did = DID(did_str) 15 - 16 assert str(did) == did_str 17 assert did.uri == did_str 18 ··· 20 """Test creating a DID with a valid did:web format.""" 21 did_str = "did:web:blueskyweb.xyz" 22 did = DID(did_str) 23 - 24 assert str(did) == did_str 25 assert did.uri == did_str 26 ··· 28 """Test creating a DID with various valid characters.""" 29 did_str = "did:method:val:two-with_underscores.and-dashes" 30 did = DID(did_str) 31 - 32 assert str(did) == did_str 33 assert did.uri == did_str 34 35 def test_invalid_did_wrong_format(self): 36 """Test that a DID with wrong format raises InvalidDIDError.""" 37 did_str = "not-a-did" 38 - 39 with pytest.raises(InvalidDIDError, match="invalid format"): 40 DID(did_str) 41 42 def test_invalid_did_uppercase_method(self): 43 """Test that a DID with uppercase method raises InvalidDIDError.""" 44 did_str = "did:METHOD:val" 45 - 46 with pytest.raises(InvalidDIDError, match="invalid format"): 47 DID(did_str) 48 49 def test_invalid_did_method_with_numbers(self): 50 """Test that a DID with method containing numbers raises InvalidDIDError.""" 51 did_str = "did:m123:val" 52 - 53 with pytest.raises(InvalidDIDError, match="invalid format"): 54 DID(did_str) 55 56 def test_invalid_did_empty_identifier(self): 57 """Test that a DID with empty identifier raises InvalidDIDError.""" 58 did_str = "did:method:" 59 - 60 with pytest.raises(InvalidDIDError, match="invalid format"): 61 DID(did_str) 62 63 def test_invalid_did_ends_with_colon(self): 64 """Test that a DID ending with colon raises InvalidDIDError.""" 65 did_str = "did:method:val:" 66 - 67 with pytest.raises(InvalidDIDError, match="invalid format"): 68 DID(did_str) 69 ··· 72 # Create a DID that exceeds the 2048 character limit 73 long_identifier = "a" * 2040 74 did_str = f"did:method:{long_identifier}" 75 - 76 with pytest.raises(InvalidDIDError, match="exceeds maximum length"): 77 DID(did_str) 78 ··· 81 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 82 did1 = DID(did_str) 83 did2 = DID(did_str) 84 - 85 assert did1 == did2 86 assert did1 != "not a did object" 87 ··· 89 """Test DID string representation.""" 90 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 91 did = DID(did_str) 92 - 93 assert str(did) == did_str 94 95 def test_did_fetch_plc_method(self): 96 """Test fetching a DID document for did:plc method.""" 97 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 98 did = DID(did_str) 99 - 100 # This test may fail if there's no internet connection or if the PLC directory is down 101 try: 102 document = did.fetch() ··· 110 """Test fetching a DID document for did:web method.""" 111 did_str = "did:web:blueskyweb.xyz" 112 did = DID(did_str) 113 - 114 # This test may fail if there's no internet connection or if the web server is down 115 try: 116 document = did.fetch() ··· 124 """Test that fetching a DID document with unsupported method raises InvalidDIDError.""" 125 did_str = "did:unsupported:method" 126 did = DID(did_str) 127 - 128 with pytest.raises(InvalidDIDError, match="unsupported DID method"): 129 did.fetch() 130 ··· 132 """Test that fetching a DID document with empty domain raises InvalidDIDError.""" 133 did_str = "did:web:" 134 did = DID(did_str) 135 - 136 with pytest.raises(InvalidDIDError, match="invalid format"): 137 - did.fetch()

··· 12 """Test creating a DID with a valid did:plc format.""" 13 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 14 did = DID(did_str) 15 + 16 assert str(did) == did_str 17 assert did.uri == did_str 18 ··· 20 """Test creating a DID with a valid did:web format.""" 21 did_str = "did:web:blueskyweb.xyz" 22 did = DID(did_str) 23 + 24 assert str(did) == did_str 25 assert did.uri == did_str 26 ··· 28 """Test creating a DID with various valid characters.""" 29 did_str = "did:method:val:two-with_underscores.and-dashes" 30 did = DID(did_str) 31 + 32 assert str(did) == did_str 33 assert did.uri == did_str 34 35 def test_invalid_did_wrong_format(self): 36 """Test that a DID with wrong format raises InvalidDIDError.""" 37 did_str = "not-a-did" 38 + 39 with pytest.raises(InvalidDIDError, match="invalid format"): 40 DID(did_str) 41 42 def test_invalid_did_uppercase_method(self): 43 """Test that a DID with uppercase method raises InvalidDIDError.""" 44 did_str = "did:METHOD:val" 45 + 46 with pytest.raises(InvalidDIDError, match="invalid format"): 47 DID(did_str) 48 49 def test_invalid_did_method_with_numbers(self): 50 """Test that a DID with method containing numbers raises InvalidDIDError.""" 51 did_str = "did:m123:val" 52 + 53 with pytest.raises(InvalidDIDError, match="invalid format"): 54 DID(did_str) 55 56 def test_invalid_did_empty_identifier(self): 57 """Test that a DID with empty identifier raises InvalidDIDError.""" 58 did_str = "did:method:" 59 + 60 with pytest.raises(InvalidDIDError, match="invalid format"): 61 DID(did_str) 62 63 def test_invalid_did_ends_with_colon(self): 64 """Test that a DID ending with colon raises InvalidDIDError.""" 65 did_str = "did:method:val:" 66 + 67 with pytest.raises(InvalidDIDError, match="invalid format"): 68 DID(did_str) 69 ··· 72 # Create a DID that exceeds the 2048 character limit 73 long_identifier = "a" * 2040 74 did_str = f"did:method:{long_identifier}" 75 + 76 with pytest.raises(InvalidDIDError, match="exceeds maximum length"): 77 DID(did_str) 78 ··· 81 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 82 did1 = DID(did_str) 83 did2 = DID(did_str) 84 + 85 assert did1 == did2 86 assert did1 != "not a did object" 87 ··· 89 """Test DID string representation.""" 90 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 91 did = DID(did_str) 92 + 93 assert str(did) == did_str 94 95 def test_did_fetch_plc_method(self): 96 """Test fetching a DID document for did:plc method.""" 97 did_str = "did:plc:z72i7hdynmk6r22z27h6tvur" 98 did = DID(did_str) 99 + 100 # This test may fail if there's no internet connection or if the PLC directory is down 101 try: 102 document = did.fetch() ··· 110 """Test fetching a DID document for did:web method.""" 111 did_str = "did:web:blueskyweb.xyz" 112 did = DID(did_str) 113 + 114 # This test may fail if there's no internet connection or if the web server is down 115 try: 116 document = did.fetch() ··· 124 """Test that fetching a DID document with unsupported method raises InvalidDIDError.""" 125 did_str = "did:unsupported:method" 126 did = DID(did_str) 127 + 128 with pytest.raises(InvalidDIDError, match="unsupported DID method"): 129 did.fetch() 130 ··· 132 """Test that fetching a DID document with empty domain raises InvalidDIDError.""" 133 did_str = "did:web:" 134 did = DID(did_str) 135 + 136 with pytest.raises(InvalidDIDError, match="invalid format"): 137 + did.fetch()

+22 -22

tests/uri/test_handle.py

··· 12 """Test creating a Handle with a valid simple format.""" 13 handle_str = "example.com" 14 handle = Handle(handle_str) 15 - 16 assert str(handle) == handle_str 17 assert handle.handle == handle_str 18 ··· 20 """Test creating a Handle with a valid subdomain format.""" 21 handle_str = "subdomain.example.com" 22 handle = Handle(handle_str) 23 - 24 assert str(handle) == handle_str 25 assert handle.handle == handle_str 26 ··· 28 """Test creating a Handle with a valid format containing hyphens.""" 29 handle_str = "my-example.com" 30 handle = Handle(handle_str) 31 - 32 assert str(handle) == handle_str 33 assert handle.handle == handle_str 34 ··· 36 """Test creating a Handle with a valid format containing numbers.""" 37 handle_str = "example123.com" 38 handle = Handle(handle_str) 39 - 40 assert str(handle) == handle_str 41 assert handle.handle == handle_str 42 ··· 44 """Test creating a Handle with a valid long domain name.""" 45 handle_str = "a" * 63 + "." + "b" * 63 + "." + "c" * 63 + ".com" 46 handle = Handle(handle_str) 47 - 48 assert str(handle) == handle_str 49 assert handle.handle == handle_str 50 ··· 53 # Create a handle that exceeds the 253 character limit 54 long_handle = "a" * 254 55 handle_str = f"{long_handle}.com" 56 - 57 with pytest.raises(InvalidHandleError, match="exceeds maximum length"): 58 Handle(handle_str) 59 60 def test_invalid_handle_no_dot_separator(self): 61 """Test that a Handle without a dot separator raises InvalidHandleError.""" 62 handle_str = "example" 63 - 64 with pytest.raises(InvalidHandleError, match="invalid format"): 65 Handle(handle_str) 66 67 def test_invalid_handle_starts_with_dot(self): 68 """Test that a Handle starting with a dot raises InvalidHandleError.""" 69 handle_str = ".example.com" 70 - 71 with pytest.raises(InvalidHandleError, match="invalid format"): 72 Handle(handle_str) 73 74 def test_invalid_handle_ends_with_dot(self): 75 """Test that a Handle ending with a dot raises InvalidHandleError.""" 76 handle_str = "example.com." 77 - 78 with pytest.raises(InvalidHandleError, match="invalid format"): 79 Handle(handle_str) 80 81 def test_invalid_handle_segment_too_long(self): 82 """Test that a Handle with a segment that is too long raises InvalidHandleError.""" 83 handle_str = f"{'a' * 64}.com" 84 - 85 with pytest.raises(InvalidHandleError, match="segment length error"): 86 Handle(handle_str) 87 88 def test_invalid_handle_segment_empty(self): 89 """Test that a Handle with an empty segment raises InvalidHandleError.""" 90 handle_str = "example..com" 91 - 92 with pytest.raises(InvalidHandleError, match="segment length error"): 93 Handle(handle_str) 94 95 def test_invalid_handle_invalid_characters(self): 96 """Test that a Handle with invalid characters raises InvalidHandleError.""" 97 handle_str = "ex@mple.com" 98 - 99 with pytest.raises(InvalidHandleError, match="contains invalid characters"): 100 Handle(handle_str) 101 102 def test_invalid_handle_segment_starts_with_hyphen(self): 103 """Test that a Handle with a segment starting with a hyphen raises InvalidHandleError.""" 104 handle_str = "-example.com" 105 - 106 with pytest.raises(InvalidHandleError, match="invalid format"): 107 Handle(handle_str) 108 109 def test_invalid_handle_segment_ends_with_hyphen(self): 110 """Test that a Handle with a segment ending with a hyphen raises InvalidHandleError.""" 111 handle_str = "example-.com" 112 - 113 with pytest.raises(InvalidHandleError, match="invalid format"): 114 Handle(handle_str) 115 116 def test_invalid_handle_tld_starts_with_digit(self): 117 """Test that a Handle with a TLD starting with a digit raises InvalidHandleError.""" 118 handle_str = "example.1com" 119 - 120 with pytest.raises(InvalidHandleError, match="invalid format"): 121 Handle(handle_str) 122 ··· 125 handle_str = "example.com" 126 handle1 = Handle(handle_str) 127 handle2 = Handle(handle_str) 128 - 129 assert handle1 == handle2 130 assert handle1 != "not a handle object" 131 ··· 133 """Test Handle string representation.""" 134 handle_str = "example.com" 135 handle = Handle(handle_str) 136 - 137 assert str(handle) == handle_str 138 139 def test_handle_case_insensitive_storage(self): 140 """Test that Handle stores the handle in lowercase.""" 141 handle_str = "ExAmPlE.CoM" 142 handle = Handle(handle_str) 143 - 144 # The handle should be stored in lowercase 145 assert handle.handle == "example.com" 146 # The string representation should also return the lowercase form ··· 150 """Test resolving a handle to DID using DNS method.""" 151 handle_str = "bsky.app" 152 handle = Handle(handle_str) 153 - 154 # This test may fail if there's no internet connection or if DNS resolution fails 155 try: 156 did = handle.toTID() ··· 164 """Test resolving a handle to DID using HTTP method.""" 165 handle_str = "blueskyweb.xyz" 166 handle = Handle(handle_str) 167 - 168 # This test may fail if there's no internet connection or if HTTP resolution fails 169 try: 170 did = handle.toTID() ··· 178 """Test resolving an unresolvable handle returns None.""" 179 handle_str = "nonexistent-domain-12345.com" 180 handle = Handle(handle_str) 181 - 182 # This should return None for a non-existent domain 183 did = handle.toTID() 184 - assert did is None

··· 12 """Test creating a Handle with a valid simple format.""" 13 handle_str = "example.com" 14 handle = Handle(handle_str) 15 + 16 assert str(handle) == handle_str 17 assert handle.handle == handle_str 18 ··· 20 """Test creating a Handle with a valid subdomain format.""" 21 handle_str = "subdomain.example.com" 22 handle = Handle(handle_str) 23 + 24 assert str(handle) == handle_str 25 assert handle.handle == handle_str 26 ··· 28 """Test creating a Handle with a valid format containing hyphens.""" 29 handle_str = "my-example.com" 30 handle = Handle(handle_str) 31 + 32 assert str(handle) == handle_str 33 assert handle.handle == handle_str 34 ··· 36 """Test creating a Handle with a valid format containing numbers.""" 37 handle_str = "example123.com" 38 handle = Handle(handle_str) 39 + 40 assert str(handle) == handle_str 41 assert handle.handle == handle_str 42 ··· 44 """Test creating a Handle with a valid long domain name.""" 45 handle_str = "a" * 63 + "." + "b" * 63 + "." + "c" * 63 + ".com" 46 handle = Handle(handle_str) 47 + 48 assert str(handle) == handle_str 49 assert handle.handle == handle_str 50 ··· 53 # Create a handle that exceeds the 253 character limit 54 long_handle = "a" * 254 55 handle_str = f"{long_handle}.com" 56 + 57 with pytest.raises(InvalidHandleError, match="exceeds maximum length"): 58 Handle(handle_str) 59 60 def test_invalid_handle_no_dot_separator(self): 61 """Test that a Handle without a dot separator raises InvalidHandleError.""" 62 handle_str = "example" 63 + 64 with pytest.raises(InvalidHandleError, match="invalid format"): 65 Handle(handle_str) 66 67 def test_invalid_handle_starts_with_dot(self): 68 """Test that a Handle starting with a dot raises InvalidHandleError.""" 69 handle_str = ".example.com" 70 + 71 with pytest.raises(InvalidHandleError, match="invalid format"): 72 Handle(handle_str) 73 74 def test_invalid_handle_ends_with_dot(self): 75 """Test that a Handle ending with a dot raises InvalidHandleError.""" 76 handle_str = "example.com." 77 + 78 with pytest.raises(InvalidHandleError, match="invalid format"): 79 Handle(handle_str) 80 81 def test_invalid_handle_segment_too_long(self): 82 """Test that a Handle with a segment that is too long raises InvalidHandleError.""" 83 handle_str = f"{'a' * 64}.com" 84 + 85 with pytest.raises(InvalidHandleError, match="segment length error"): 86 Handle(handle_str) 87 88 def test_invalid_handle_segment_empty(self): 89 """Test that a Handle with an empty segment raises InvalidHandleError.""" 90 handle_str = "example..com" 91 + 92 with pytest.raises(InvalidHandleError, match="segment length error"): 93 Handle(handle_str) 94 95 def test_invalid_handle_invalid_characters(self): 96 """Test that a Handle with invalid characters raises InvalidHandleError.""" 97 handle_str = "ex@mple.com" 98 + 99 with pytest.raises(InvalidHandleError, match="contains invalid characters"): 100 Handle(handle_str) 101 102 def test_invalid_handle_segment_starts_with_hyphen(self): 103 """Test that a Handle with a segment starting with a hyphen raises InvalidHandleError.""" 104 handle_str = "-example.com" 105 + 106 with pytest.raises(InvalidHandleError, match="invalid format"): 107 Handle(handle_str) 108 109 def test_invalid_handle_segment_ends_with_hyphen(self): 110 """Test that a Handle with a segment ending with a hyphen raises InvalidHandleError.""" 111 handle_str = "example-.com" 112 + 113 with pytest.raises(InvalidHandleError, match="invalid format"): 114 Handle(handle_str) 115 116 def test_invalid_handle_tld_starts_with_digit(self): 117 """Test that a Handle with a TLD starting with a digit raises InvalidHandleError.""" 118 handle_str = "example.1com" 119 + 120 with pytest.raises(InvalidHandleError, match="invalid format"): 121 Handle(handle_str) 122 ··· 125 handle_str = "example.com" 126 handle1 = Handle(handle_str) 127 handle2 = Handle(handle_str) 128 + 129 assert handle1 == handle2 130 assert handle1 != "not a handle object" 131 ··· 133 """Test Handle string representation.""" 134 handle_str = "example.com" 135 handle = Handle(handle_str) 136 + 137 assert str(handle) == handle_str 138 139 def test_handle_case_insensitive_storage(self): 140 """Test that Handle stores the handle in lowercase.""" 141 handle_str = "ExAmPlE.CoM" 142 handle = Handle(handle_str) 143 + 144 # The handle should be stored in lowercase 145 assert handle.handle == "example.com" 146 # The string representation should also return the lowercase form ··· 150 """Test resolving a handle to DID using DNS method.""" 151 handle_str = "bsky.app" 152 handle = Handle(handle_str) 153 + 154 # This test may fail if there's no internet connection or if DNS resolution fails 155 try: 156 did = handle.toTID() ··· 164 """Test resolving a handle to DID using HTTP method.""" 165 handle_str = "blueskyweb.xyz" 166 handle = Handle(handle_str) 167 + 168 # This test may fail if there's no internet connection or if HTTP resolution fails 169 try: 170 did = handle.toTID() ··· 178 """Test resolving an unresolvable handle returns None.""" 179 handle_str = "nonexistent-domain-12345.com" 180 handle = Handle(handle_str) 181 + 182 # This should return None for a non-existent domain 183 did = handle.toTID() 184 + assert did is None

+39 -33

tests/uri/test_nsid.py

··· 12 """Test creating an NSID with a valid simple format.""" 13 nsid_str = "com.example.recordName" 14 nsid = NSID(nsid_str) 15 - 16 assert str(nsid) == nsid_str 17 assert nsid.nsid == nsid_str 18 assert nsid.domainAuthority == ["com", "example"] ··· 24 """Test creating an NSID with a valid fragment.""" 25 nsid_str = "com.example.recordName#fragment" 26 nsid = NSID(nsid_str) 27 - 28 assert str(nsid) == nsid_str 29 assert nsid.nsid == nsid_str 30 assert nsid.domainAuthority == ["com", "example"] ··· 36 """Test creating an NSID with multiple domain segments.""" 37 nsid_str = "net.users.bob.ping" 38 nsid = NSID(nsid_str) 39 - 40 assert str(nsid) == nsid_str 41 assert nsid.nsid == nsid_str 42 assert nsid.domainAuthority == ["net", "users", "bob"] ··· 48 """Test creating an NSID with hyphens in domain segments.""" 49 nsid_str = "a-0.b-1.c.recordName" 50 nsid = NSID(nsid_str) 51 - 52 assert str(nsid) == nsid_str 53 assert nsid.nsid == nsid_str 54 assert nsid.domainAuthority == ["a-0", "b-1", "c"] ··· 60 """Test creating an NSID with case-sensitive name.""" 61 nsid_str = "com.example.fooBar" 62 nsid = NSID(nsid_str) 63 - 64 assert str(nsid) == nsid_str 65 assert nsid.nsid == nsid_str 66 assert nsid.domainAuthority == ["com", "example"] ··· 72 """Test creating an NSID with numbers in the name.""" 73 nsid_str = "com.example.record123" 74 nsid = NSID(nsid_str) 75 - 76 assert str(nsid) == nsid_str 77 assert nsid.nsid == nsid_str 78 assert nsid.domainAuthority == ["com", "example"] ··· 83 def test_invalid_nsid_non_ascii_characters(self): 84 """Test that an NSID with non-ASCII characters raises InvalidNSIDError.""" 85 nsid_str = "com.exa💩ple.thing" 86 - 87 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 88 NSID(nsid_str) 89 ··· 92 # Create an NSID that exceeds the 317 character limit 93 long_segment = "a" * 100 94 nsid_str = f"{long_segment}.{long_segment}.{long_segment}.recordName" 95 - 96 - with pytest.raises(InvalidNSIDError, match="domain authority length exceeds limit"): 0 0 97 NSID(nsid_str) 98 99 def test_invalid_nsid_starts_with_dot(self): 100 """Test that an NSID starting with a dot raises InvalidNSIDError.""" 101 nsid_str = ".com.example.recordName" 102 - 103 with pytest.raises(InvalidNSIDError, match="invalid format"): 104 NSID(nsid_str) 105 106 def test_invalid_nsid_ends_with_dot(self): 107 """Test that an NSID ending with a dot raises InvalidNSIDError.""" 108 nsid_str = "com.example.recordName." 109 - 110 with pytest.raises(InvalidNSIDError, match="invalid format"): 111 NSID(nsid_str) 112 113 def test_invalid_nsid_too_few_segments(self): 114 """Test that an NSID with too few segments raises InvalidNSIDError.""" 115 nsid_str = "com.example" 116 - 117 with pytest.raises(InvalidNSIDError, match="invalid format"): 118 NSID(nsid_str) 119 ··· 121 """Test that an NSID with domain authority that is too long raises InvalidNSIDError.""" 122 # Create a domain authority that exceeds the 253 character limit 123 long_segment = "a" * 63 124 - nsid_str = f"{long_segment}.{long_segment}.{long_segment}.{long_segment}.recordName" 125 - 126 - with pytest.raises(InvalidNSIDError, match="domain authority length exceeds limit"): 0 0 0 0 127 NSID(nsid_str) 128 129 def test_invalid_nsid_domain_segment_too_long(self): 130 """Test that an NSID with a domain segment that is too long raises InvalidNSIDError.""" 131 nsid_str = f"{'a' * 64}.example.recordName" 132 - 133 with pytest.raises(InvalidNSIDError, match="segment length error"): 134 NSID(nsid_str) 135 136 def test_invalid_nsid_domain_segment_empty(self): 137 """Test that an NSID with an empty domain segment raises InvalidNSIDError.""" 138 nsid_str = "com..example.recordName" 139 - 140 with pytest.raises(InvalidNSIDError, match="segment length error"): 141 NSID(nsid_str) 142 143 def test_invalid_nsid_domain_invalid_characters(self): 144 """Test that an NSID with invalid characters in domain raises InvalidNSIDError.""" 145 nsid_str = "com.ex@mple.recordName" 146 - 147 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 148 NSID(nsid_str) 149 150 def test_invalid_nsid_domain_segment_starts_with_hyphen(self): 151 """Test that an NSID with a domain segment starting with a hyphen raises InvalidNSIDError.""" 152 nsid_str = "com.-example.recordName" 153 - 154 with pytest.raises(InvalidNSIDError, match="invalid format"): 155 NSID(nsid_str) 156 157 def test_invalid_nsid_domain_segment_ends_with_hyphen(self): 158 """Test that an NSID with a domain segment ending with a hyphen raises InvalidNSIDError.""" 159 nsid_str = "com.example-.recordName" 160 - 161 with pytest.raises(InvalidNSIDError, match="invalid format"): 162 NSID(nsid_str) 163 164 def test_invalid_nsid_tld_starts_with_digit(self): 165 """Test that an NSID with a TLD starting with a digit raises InvalidNSIDError.""" 166 nsid_str = "1com.example.recordName" 167 - 168 with pytest.raises(InvalidNSIDError, match="invalid format"): 169 NSID(nsid_str) 170 171 def test_invalid_nsid_name_empty(self): 172 """Test that an NSID with an empty name raises InvalidNSIDError.""" 173 nsid_str = "com.example." 174 - 175 with pytest.raises(InvalidNSIDError, match="invalid format"): 176 NSID(nsid_str) 177 178 def test_invalid_nsid_name_too_long(self): 179 """Test that an NSID with a name that is too long raises InvalidNSIDError.""" 180 nsid_str = f"com.example.{'a' * 64}" 181 - 182 with pytest.raises(InvalidNSIDError, match="name length error"): 183 NSID(nsid_str) 184 185 def test_invalid_nsid_name_invalid_characters(self): 186 """Test that an NSID with invalid characters in name raises InvalidNSIDError.""" 187 nsid_str = "com.example.record-name" 188 - 189 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 190 NSID(nsid_str) 191 192 def test_invalid_nsid_name_starts_with_digit(self): 193 """Test that an NSID with a name starting with a digit raises InvalidNSIDError.""" 194 nsid_str = "com.example.1record" 195 - 196 with pytest.raises(InvalidNSIDError, match="invalid format"): 197 NSID(nsid_str) 198 199 def test_invalid_nsid_fragment_empty(self): 200 """Test that an NSID with an empty fragment raises InvalidNSIDError.""" 201 nsid_str = "com.example.recordName#" 202 - 203 with pytest.raises(InvalidNSIDError, match="fragment length error"): 204 NSID(nsid_str) 205 206 def test_invalid_nsid_fragment_too_long(self): 207 """Test that an NSID with a fragment that is too long raises InvalidNSIDError.""" 208 nsid_str = f"com.example.recordName#{'a' * 64}" 209 - 210 with pytest.raises(InvalidNSIDError, match="fragment length error"): 211 NSID(nsid_str) 212 213 def test_invalid_nsid_fragment_invalid_characters(self): 214 """Test that an NSID with invalid characters in fragment raises InvalidNSIDError.""" 215 nsid_str = "com.example.recordName#fragment-with-hyphen" 216 - 217 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 218 NSID(nsid_str) 219 220 def test_invalid_nsid_fragment_starts_with_digit(self): 221 """Test that an NSID with a fragment starting with a digit raises InvalidNSIDError.""" 222 nsid_str = "com.example.recordName#1fragment" 223 - 224 with pytest.raises(InvalidNSIDError, match="invalid format"): 225 NSID(nsid_str) 226 ··· 229 nsid_str = "com.example.recordName" 230 nsid1 = NSID(nsid_str) 231 nsid2 = NSID(nsid_str) 232 - 233 assert nsid1 == nsid2 234 assert nsid1 != "not an nsid object" 235 ··· 237 """Test NSID string representation.""" 238 nsid_str = "com.example.recordName" 239 nsid = NSID(nsid_str) 240 - 241 assert str(nsid) == nsid_str 242 243 def test_nsid_string_representation_with_fragment(self): 244 """Test NSID string representation with fragment.""" 245 nsid_str = "com.example.recordName#fragment" 246 nsid = NSID(nsid_str) 247 - 248 - assert str(nsid) == nsid_str

··· 12 """Test creating an NSID with a valid simple format.""" 13 nsid_str = "com.example.recordName" 14 nsid = NSID(nsid_str) 15 + 16 assert str(nsid) == nsid_str 17 assert nsid.nsid == nsid_str 18 assert nsid.domainAuthority == ["com", "example"] ··· 24 """Test creating an NSID with a valid fragment.""" 25 nsid_str = "com.example.recordName#fragment" 26 nsid = NSID(nsid_str) 27 + 28 assert str(nsid) == nsid_str 29 assert nsid.nsid == nsid_str 30 assert nsid.domainAuthority == ["com", "example"] ··· 36 """Test creating an NSID with multiple domain segments.""" 37 nsid_str = "net.users.bob.ping" 38 nsid = NSID(nsid_str) 39 + 40 assert str(nsid) == nsid_str 41 assert nsid.nsid == nsid_str 42 assert nsid.domainAuthority == ["net", "users", "bob"] ··· 48 """Test creating an NSID with hyphens in domain segments.""" 49 nsid_str = "a-0.b-1.c.recordName" 50 nsid = NSID(nsid_str) 51 + 52 assert str(nsid) == nsid_str 53 assert nsid.nsid == nsid_str 54 assert nsid.domainAuthority == ["a-0", "b-1", "c"] ··· 60 """Test creating an NSID with case-sensitive name.""" 61 nsid_str = "com.example.fooBar" 62 nsid = NSID(nsid_str) 63 + 64 assert str(nsid) == nsid_str 65 assert nsid.nsid == nsid_str 66 assert nsid.domainAuthority == ["com", "example"] ··· 72 """Test creating an NSID with numbers in the name.""" 73 nsid_str = "com.example.record123" 74 nsid = NSID(nsid_str) 75 + 76 assert str(nsid) == nsid_str 77 assert nsid.nsid == nsid_str 78 assert nsid.domainAuthority == ["com", "example"] ··· 83 def test_invalid_nsid_non_ascii_characters(self): 84 """Test that an NSID with non-ASCII characters raises InvalidNSIDError.""" 85 nsid_str = "com.exa💩ple.thing" 86 + 87 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 88 NSID(nsid_str) 89 ··· 92 # Create an NSID that exceeds the 317 character limit 93 long_segment = "a" * 100 94 nsid_str = f"{long_segment}.{long_segment}.{long_segment}.recordName" 95 + 96 + with pytest.raises( 97 + InvalidNSIDError, match="domain authority length exceeds limit" 98 + ): 99 NSID(nsid_str) 100 101 def test_invalid_nsid_starts_with_dot(self): 102 """Test that an NSID starting with a dot raises InvalidNSIDError.""" 103 nsid_str = ".com.example.recordName" 104 + 105 with pytest.raises(InvalidNSIDError, match="invalid format"): 106 NSID(nsid_str) 107 108 def test_invalid_nsid_ends_with_dot(self): 109 """Test that an NSID ending with a dot raises InvalidNSIDError.""" 110 nsid_str = "com.example.recordName." 111 + 112 with pytest.raises(InvalidNSIDError, match="invalid format"): 113 NSID(nsid_str) 114 115 def test_invalid_nsid_too_few_segments(self): 116 """Test that an NSID with too few segments raises InvalidNSIDError.""" 117 nsid_str = "com.example" 118 + 119 with pytest.raises(InvalidNSIDError, match="invalid format"): 120 NSID(nsid_str) 121 ··· 123 """Test that an NSID with domain authority that is too long raises InvalidNSIDError.""" 124 # Create a domain authority that exceeds the 253 character limit 125 long_segment = "a" * 63 126 + nsid_str = ( 127 + f"{long_segment}.{long_segment}.{long_segment}.{long_segment}.recordName" 128 + ) 129 + 130 + with pytest.raises( 131 + InvalidNSIDError, match="domain authority length exceeds limit" 132 + ): 133 NSID(nsid_str) 134 135 def test_invalid_nsid_domain_segment_too_long(self): 136 """Test that an NSID with a domain segment that is too long raises InvalidNSIDError.""" 137 nsid_str = f"{'a' * 64}.example.recordName" 138 + 139 with pytest.raises(InvalidNSIDError, match="segment length error"): 140 NSID(nsid_str) 141 142 def test_invalid_nsid_domain_segment_empty(self): 143 """Test that an NSID with an empty domain segment raises InvalidNSIDError.""" 144 nsid_str = "com..example.recordName" 145 + 146 with pytest.raises(InvalidNSIDError, match="segment length error"): 147 NSID(nsid_str) 148 149 def test_invalid_nsid_domain_invalid_characters(self): 150 """Test that an NSID with invalid characters in domain raises InvalidNSIDError.""" 151 nsid_str = "com.ex@mple.recordName" 152 + 153 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 154 NSID(nsid_str) 155 156 def test_invalid_nsid_domain_segment_starts_with_hyphen(self): 157 """Test that an NSID with a domain segment starting with a hyphen raises InvalidNSIDError.""" 158 nsid_str = "com.-example.recordName" 159 + 160 with pytest.raises(InvalidNSIDError, match="invalid format"): 161 NSID(nsid_str) 162 163 def test_invalid_nsid_domain_segment_ends_with_hyphen(self): 164 """Test that an NSID with a domain segment ending with a hyphen raises InvalidNSIDError.""" 165 nsid_str = "com.example-.recordName" 166 + 167 with pytest.raises(InvalidNSIDError, match="invalid format"): 168 NSID(nsid_str) 169 170 def test_invalid_nsid_tld_starts_with_digit(self): 171 """Test that an NSID with a TLD starting with a digit raises InvalidNSIDError.""" 172 nsid_str = "1com.example.recordName" 173 + 174 with pytest.raises(InvalidNSIDError, match="invalid format"): 175 NSID(nsid_str) 176 177 def test_invalid_nsid_name_empty(self): 178 """Test that an NSID with an empty name raises InvalidNSIDError.""" 179 nsid_str = "com.example." 180 + 181 with pytest.raises(InvalidNSIDError, match="invalid format"): 182 NSID(nsid_str) 183 184 def test_invalid_nsid_name_too_long(self): 185 """Test that an NSID with a name that is too long raises InvalidNSIDError.""" 186 nsid_str = f"com.example.{'a' * 64}" 187 + 188 with pytest.raises(InvalidNSIDError, match="name length error"): 189 NSID(nsid_str) 190 191 def test_invalid_nsid_name_invalid_characters(self): 192 """Test that an NSID with invalid characters in name raises InvalidNSIDError.""" 193 nsid_str = "com.example.record-name" 194 + 195 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 196 NSID(nsid_str) 197 198 def test_invalid_nsid_name_starts_with_digit(self): 199 """Test that an NSID with a name starting with a digit raises InvalidNSIDError.""" 200 nsid_str = "com.example.1record" 201 + 202 with pytest.raises(InvalidNSIDError, match="invalid format"): 203 NSID(nsid_str) 204 205 def test_invalid_nsid_fragment_empty(self): 206 """Test that an NSID with an empty fragment raises InvalidNSIDError.""" 207 nsid_str = "com.example.recordName#" 208 + 209 with pytest.raises(InvalidNSIDError, match="fragment length error"): 210 NSID(nsid_str) 211 212 def test_invalid_nsid_fragment_too_long(self): 213 """Test that an NSID with a fragment that is too long raises InvalidNSIDError.""" 214 nsid_str = f"com.example.recordName#{'a' * 64}" 215 + 216 with pytest.raises(InvalidNSIDError, match="fragment length error"): 217 NSID(nsid_str) 218 219 def test_invalid_nsid_fragment_invalid_characters(self): 220 """Test that an NSID with invalid characters in fragment raises InvalidNSIDError.""" 221 nsid_str = "com.example.recordName#fragment-with-hyphen" 222 + 223 with pytest.raises(InvalidNSIDError, match="contains invalid characters"): 224 NSID(nsid_str) 225 226 def test_invalid_nsid_fragment_starts_with_digit(self): 227 """Test that an NSID with a fragment starting with a digit raises InvalidNSIDError.""" 228 nsid_str = "com.example.recordName#1fragment" 229 + 230 with pytest.raises(InvalidNSIDError, match="invalid format"): 231 NSID(nsid_str) 232 ··· 235 nsid_str = "com.example.recordName" 236 nsid1 = NSID(nsid_str) 237 nsid2 = NSID(nsid_str) 238 + 239 assert nsid1 == nsid2 240 assert nsid1 != "not an nsid object" 241 ··· 243 """Test NSID string representation.""" 244 nsid_str = "com.example.recordName" 245 nsid = NSID(nsid_str) 246 + 247 assert str(nsid) == nsid_str 248 249 def test_nsid_string_representation_with_fragment(self): 250 """Test NSID string representation with fragment.""" 251 nsid_str = "com.example.recordName#fragment" 252 nsid = NSID(nsid_str) 253 + 254 + assert str(nsid) == nsid_str

+27 -17

tests/uri/test_restricted_uri.py

··· 10 11 def test_valid_restricted_uri_with_did_collection_and_rkey(self): 12 """Test creating a RestrictedURI with a valid DID, collection, and rkey.""" 13 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 14 uri = RestrictedURI(uri_str) 15 - 16 assert str(uri) == uri_str 17 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 18 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 26 """Test creating a RestrictedURI with a valid handle, collection, and rkey.""" 27 uri_str = "at://bnewbold.bsky.team/app.bsky.feed.post/3jwdwj2ctlk26" 28 uri = RestrictedURI(uri_str) 29 - 30 assert str(uri) == uri_str 31 assert uri.authorityAsText == "bnewbold.bsky.team" 32 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 39 """Test creating a RestrictedURI with only a collection.""" 40 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 41 uri = RestrictedURI(uri_str) 42 - 43 assert str(uri) == uri_str 44 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 45 assert uri.path == ["app.bsky.feed.post"] ··· 51 """Test creating a RestrictedURI with only an authority.""" 52 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur" 53 uri = RestrictedURI(uri_str) 54 - 55 assert str(uri) == uri_str 56 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 57 assert uri.path == [] ··· 60 61 def test_invalid_restricted_uri_with_query(self): 62 """Test that a RestrictedURI with query parameters raises InvalidRestrictedURIError.""" 63 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post?param1=value1" 64 - 65 - with pytest.raises(InvalidRestrictedURIError, match="query parameters not supported"): 0 0 0 0 66 RestrictedURI(uri_str) 67 68 def test_invalid_restricted_uri_with_fragment(self): 69 """Test that a RestrictedURI with a fragment raises InvalidRestrictedURIError.""" 70 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26#$.some.json.path" 71 - 72 with pytest.raises(InvalidRestrictedURIError, match="fragments not supported"): 73 RestrictedURI(uri_str) 74 75 def test_invalid_restricted_uri_with_invalid_authority(self): 76 """Test that a RestrictedURI with invalid authority raises InvalidRestrictedURIError.""" 77 uri_str = "at://invalid_authority/app.bsky.feed.post/3jwdwj2ctlk26" 78 - 79 with pytest.raises(InvalidRestrictedURIError, match="invalid authority"): 80 RestrictedURI(uri_str) 81 82 def test_invalid_restricted_uri_too_many_path_segments(self): 83 """Test that a RestrictedURI with too many path segments raises InvalidRestrictedURIError.""" 84 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26/extra" 85 - 86 with pytest.raises(InvalidRestrictedURIError, match="too many path segments"): 87 RestrictedURI(uri_str) 88 89 def test_invalid_restricted_uri_base_uri_validation_failure(self): 90 """Test that a RestrictedURI with invalid base URI raises InvalidURIError.""" 91 uri_str = "https://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 92 - 93 with pytest.raises(InvalidURIError, match="invalid format"): 94 RestrictedURI(uri_str) 95 96 def test_restricted_uri_equality(self): 97 """Test RestrictedURI equality comparison.""" 98 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 99 uri1 = RestrictedURI(uri_str) 100 uri2 = RestrictedURI(uri_str) 101 - 102 assert uri1 == uri2 103 assert uri1 != "not a uri object" 104 105 def test_restricted_uri_string_representation(self): 106 """Test RestrictedURI string representation.""" 107 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 108 uri = RestrictedURI(uri_str) 109 - 110 - assert str(uri) == uri_str

··· 10 11 def test_valid_restricted_uri_with_did_collection_and_rkey(self): 12 """Test creating a RestrictedURI with a valid DID, collection, and rkey.""" 13 + uri_str = ( 14 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 15 + ) 16 uri = RestrictedURI(uri_str) 17 + 18 assert str(uri) == uri_str 19 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 20 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 28 """Test creating a RestrictedURI with a valid handle, collection, and rkey.""" 29 uri_str = "at://bnewbold.bsky.team/app.bsky.feed.post/3jwdwj2ctlk26" 30 uri = RestrictedURI(uri_str) 31 + 32 assert str(uri) == uri_str 33 assert uri.authorityAsText == "bnewbold.bsky.team" 34 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 41 """Test creating a RestrictedURI with only a collection.""" 42 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 43 uri = RestrictedURI(uri_str) 44 + 45 assert str(uri) == uri_str 46 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 47 assert uri.path == ["app.bsky.feed.post"] ··· 53 """Test creating a RestrictedURI with only an authority.""" 54 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur" 55 uri = RestrictedURI(uri_str) 56 + 57 assert str(uri) == uri_str 58 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 59 assert uri.path == [] ··· 62 63 def test_invalid_restricted_uri_with_query(self): 64 """Test that a RestrictedURI with query parameters raises InvalidRestrictedURIError.""" 65 + uri_str = ( 66 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post?param1=value1" 67 + ) 68 + 69 + with pytest.raises( 70 + InvalidRestrictedURIError, match="query parameters not supported" 71 + ): 72 RestrictedURI(uri_str) 73 74 def test_invalid_restricted_uri_with_fragment(self): 75 """Test that a RestrictedURI with a fragment raises InvalidRestrictedURIError.""" 76 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26#$.some.json.path" 77 + 78 with pytest.raises(InvalidRestrictedURIError, match="fragments not supported"): 79 RestrictedURI(uri_str) 80 81 def test_invalid_restricted_uri_with_invalid_authority(self): 82 """Test that a RestrictedURI with invalid authority raises InvalidRestrictedURIError.""" 83 uri_str = "at://invalid_authority/app.bsky.feed.post/3jwdwj2ctlk26" 84 + 85 with pytest.raises(InvalidRestrictedURIError, match="invalid authority"): 86 RestrictedURI(uri_str) 87 88 def test_invalid_restricted_uri_too_many_path_segments(self): 89 """Test that a RestrictedURI with too many path segments raises InvalidRestrictedURIError.""" 90 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26/extra" 91 + 92 with pytest.raises(InvalidRestrictedURIError, match="too many path segments"): 93 RestrictedURI(uri_str) 94 95 def test_invalid_restricted_uri_base_uri_validation_failure(self): 96 """Test that a RestrictedURI with invalid base URI raises InvalidURIError.""" 97 uri_str = "https://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 98 + 99 with pytest.raises(InvalidURIError, match="invalid format"): 100 RestrictedURI(uri_str) 101 102 def test_restricted_uri_equality(self): 103 """Test RestrictedURI equality comparison.""" 104 + uri_str = ( 105 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 106 + ) 107 uri1 = RestrictedURI(uri_str) 108 uri2 = RestrictedURI(uri_str) 109 + 110 assert uri1 == uri2 111 assert uri1 != "not a uri object" 112 113 def test_restricted_uri_string_representation(self): 114 """Test RestrictedURI string representation.""" 115 + uri_str = ( 116 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 117 + ) 118 uri = RestrictedURI(uri_str) 119 + 120 + assert str(uri) == uri_str

+39 -39

tests/uri/test_rkey.py

··· 13 """Test creating an RKey with a valid simple format.""" 14 rkey_str = "3jui7kd54zh2y" 15 rkey = RKey(rkey_str) 16 - 17 assert str(rkey) == rkey_str 18 assert rkey.recordKey == rkey_str 19 ··· 21 """Test creating an RKey with various valid characters.""" 22 rkey_str = "example.com" 23 rkey = RKey(rkey_str) 24 - 25 assert str(rkey) == rkey_str 26 assert rkey.recordKey == rkey_str 27 ··· 29 """Test creating an RKey with valid special characters.""" 30 rkey_str = "~1.2-3_" 31 rkey = RKey(rkey_str) 32 - 33 assert str(rkey) == rkey_str 34 assert rkey.recordKey == rkey_str 35 ··· 37 """Test creating an RKey with a colon.""" 38 rkey_str = "pre:fix" 39 rkey = RKey(rkey_str) 40 - 41 assert str(rkey) == rkey_str 42 assert rkey.recordKey == rkey_str 43 ··· 45 """Test creating an RKey with just an underscore.""" 46 rkey_str = "_" 47 rkey = RKey(rkey_str) 48 - 49 assert str(rkey) == rkey_str 50 assert rkey.recordKey == rkey_str 51 52 def test_invalid_rkey_empty(self): 53 """Test that an empty RKey raises InvalidRKeyError.""" 54 rkey_str = "" 55 - 56 with pytest.raises(InvalidRKeyError, match="record key is empty"): 57 RKey(rkey_str) 58 ··· 60 """Test that an RKey that is too long raises InvalidRKeyError.""" 61 # Create an RKey that exceeds the 512 character limit 62 rkey_str = "a" * 513 63 - 64 with pytest.raises(InvalidRKeyError, match="exceeds maximum length"): 65 RKey(rkey_str) 66 67 def test_invalid_rkey_reserved_double_dot(self): 68 """Test that an RKey with '..' raises InvalidRKeyError.""" 69 rkey_str = ".." 70 - 71 with pytest.raises(InvalidRKeyError, match="reserved value"): 72 RKey(rkey_str) 73 74 def test_invalid_rkey_reserved_single_dot(self): 75 """Test that an RKey with '.' raises InvalidRKeyError.""" 76 rkey_str = "." 77 - 78 with pytest.raises(InvalidRKeyError, match="reserved value"): 79 RKey(rkey_str) 80 81 def test_invalid_rkey_invalid_characters(self): 82 """Test that an RKey with invalid characters raises InvalidRKeyError.""" 83 rkey_str = "alpha/beta" 84 - 85 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 86 RKey(rkey_str) 87 88 def test_invalid_rkey_hash_character(self): 89 """Test that an RKey with a hash character raises InvalidRKeyError.""" 90 rkey_str = "#extra" 91 - 92 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 93 RKey(rkey_str) 94 95 def test_invalid_rkey_at_character(self): 96 """Test that an RKey with an at character raises InvalidRKeyError.""" 97 rkey_str = "@handle" 98 - 99 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 100 RKey(rkey_str) 101 102 def test_invalid_rkey_space(self): 103 """Test that an RKey with a space raises InvalidRKeyError.""" 104 rkey_str = "any space" 105 - 106 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 107 RKey(rkey_str) 108 109 def test_invalid_rkey_plus_character(self): 110 """Test that an RKey with a plus character raises InvalidRKeyError.""" 111 rkey_str = "any+space" 112 - 113 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 114 RKey(rkey_str) 115 116 def test_invalid_rkey_brackets(self): 117 """Test that an RKey with brackets raises InvalidRKeyError.""" 118 rkey_str = "number[3]" 119 - 120 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 121 RKey(rkey_str) 122 123 def test_invalid_rkey_parentheses(self): 124 """Test that an RKey with parentheses raises InvalidRKeyError.""" 125 rkey_str = "number(3)" 126 - 127 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 128 RKey(rkey_str) 129 130 def test_invalid_rkey_quotes(self): 131 """Test that an RKey with quotes raises InvalidRKeyError.""" 132 rkey_str = '"quote"' 133 - 134 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 135 RKey(rkey_str) 136 137 def test_invalid_rkey_base64_padding(self): 138 """Test that an RKey with base64 padding raises InvalidRKeyError.""" 139 rkey_str = "dHJ1ZQ==" 140 - 141 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 142 RKey(rkey_str) 143 ··· 146 rkey_str = "3jui7kd54zh2y" 147 rkey1 = RKey(rkey_str) 148 rkey2 = RKey(rkey_str) 149 - 150 assert rkey1 == rkey2 151 assert rkey1 != "not an rkey object" 152 ··· 154 """Test RKey string representation.""" 155 rkey_str = "3jui7kd54zh2y" 156 rkey = RKey(rkey_str) 157 - 158 assert str(rkey) == rkey_str 159 160 ··· 164 def test_tid_creation_default(self): 165 """Test creating a TID with default parameters.""" 166 tid = TID() 167 - 168 assert isinstance(tid, TID) 169 assert isinstance(tid, RKey) 170 assert isinstance(tid.timestamp, datetime.datetime) ··· 176 """Test creating a TID with a specific timestamp.""" 177 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 178 tid = TID(time=timestamp) 179 - 180 assert tid.timestamp == timestamp 181 assert isinstance(tid.clockIdentifier, int) 182 assert 0 <= tid.clockIdentifier < 1024 ··· 185 """Test creating a TID with a specific clock identifier.""" 186 clock_id = 42 187 tid = TID(clockIdentifier=clock_id) 188 - 189 assert tid.clockIdentifier == clock_id 190 assert isinstance(tid.timestamp, datetime.datetime) 191 ··· 194 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 195 clock_id = 42 196 tid = TID(time=timestamp, clockIdentifier=clock_id) 197 - 198 assert tid.timestamp == timestamp 199 assert tid.clockIdentifier == clock_id 200 ··· 203 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 204 clock_id = 42 205 tid = TID(time=timestamp, clockIdentifier=clock_id) 206 - 207 int_value = int(tid) 208 expected_value = int(timestamp.timestamp() * 1000000) * 1024 + clock_id 209 - 210 assert int_value == expected_value 211 212 def test_tid_string_representation(self): 213 """Test TID string representation.""" 214 tid = TID() 215 - 216 str_value = str(tid) 217 assert len(str_value) == 13 218 assert all(c in "234567abcdefghijklmnopqrstuvwxyz" for c in str_value) ··· 223 clock_id = 42 224 tid1 = TID(time=timestamp, clockIdentifier=clock_id) 225 tid2 = TID(time=timestamp, clockIdentifier=clock_id) 226 - 227 assert tid1 == tid2 228 229 def test_tid_equality_with_rkey(self): ··· 232 clock_id = 42 233 tid = TID(time=timestamp, clockIdentifier=clock_id) 234 rkey = RKey(str(tid)) 235 - 236 assert tid == rkey 237 238 def test_tid_inequality_with_different_object(self): 239 """Test TID inequality comparison with a different object type.""" 240 tid = TID() 241 - 242 assert tid != "not a tid object" 243 244 def test_tid_inequality_with_different_timestamp(self): ··· 248 clock_id = 42 249 tid1 = TID(time=timestamp1, clockIdentifier=clock_id) 250 tid2 = TID(time=timestamp2, clockIdentifier=clock_id) 251 - 252 assert tid1 != tid2 253 254 def test_tid_inequality_with_different_clock_id(self): ··· 258 clock_id2 = 43 259 tid1 = TID(time=timestamp, clockIdentifier=clock_id1) 260 tid2 = TID(time=timestamp, clockIdentifier=clock_id2) 261 - 262 assert tid1 != tid2 263 264 ··· 268 def test_import_tid_from_integer_default(self): 269 """Test importing a TID from integer with default value.""" 270 tid = importTIDfromInteger() 271 - 272 assert isinstance(tid, TID) 273 assert isinstance(tid.timestamp, datetime.datetime) 274 assert isinstance(tid.clockIdentifier, int) ··· 280 clock_id = 42 281 original_tid = TID(time=timestamp, clockIdentifier=clock_id) 282 int_value = int(original_tid) 283 - 284 imported_tid = importTIDfromInteger(int_value) 285 - 286 assert imported_tid.timestamp == timestamp 287 assert imported_tid.clockIdentifier == clock_id 288 289 def test_import_tid_from_base32_default(self): 290 """Test importing a TID from base32 with default value.""" 291 tid = importTIDfromBase32() 292 - 293 assert isinstance(tid, TID) 294 assert isinstance(tid.timestamp, datetime.datetime) 295 assert isinstance(tid.clockIdentifier, int) ··· 299 """Test importing a TID from base32 with a specific value.""" 300 original_tid = TID() 301 str_value = str(original_tid) 302 - 303 imported_tid = importTIDfromBase32(str_value) 304 - 305 - assert int(imported_tid) == int(original_tid)

··· 13 """Test creating an RKey with a valid simple format.""" 14 rkey_str = "3jui7kd54zh2y" 15 rkey = RKey(rkey_str) 16 + 17 assert str(rkey) == rkey_str 18 assert rkey.recordKey == rkey_str 19 ··· 21 """Test creating an RKey with various valid characters.""" 22 rkey_str = "example.com" 23 rkey = RKey(rkey_str) 24 + 25 assert str(rkey) == rkey_str 26 assert rkey.recordKey == rkey_str 27 ··· 29 """Test creating an RKey with valid special characters.""" 30 rkey_str = "~1.2-3_" 31 rkey = RKey(rkey_str) 32 + 33 assert str(rkey) == rkey_str 34 assert rkey.recordKey == rkey_str 35 ··· 37 """Test creating an RKey with a colon.""" 38 rkey_str = "pre:fix" 39 rkey = RKey(rkey_str) 40 + 41 assert str(rkey) == rkey_str 42 assert rkey.recordKey == rkey_str 43 ··· 45 """Test creating an RKey with just an underscore.""" 46 rkey_str = "_" 47 rkey = RKey(rkey_str) 48 + 49 assert str(rkey) == rkey_str 50 assert rkey.recordKey == rkey_str 51 52 def test_invalid_rkey_empty(self): 53 """Test that an empty RKey raises InvalidRKeyError.""" 54 rkey_str = "" 55 + 56 with pytest.raises(InvalidRKeyError, match="record key is empty"): 57 RKey(rkey_str) 58 ··· 60 """Test that an RKey that is too long raises InvalidRKeyError.""" 61 # Create an RKey that exceeds the 512 character limit 62 rkey_str = "a" * 513 63 + 64 with pytest.raises(InvalidRKeyError, match="exceeds maximum length"): 65 RKey(rkey_str) 66 67 def test_invalid_rkey_reserved_double_dot(self): 68 """Test that an RKey with '..' raises InvalidRKeyError.""" 69 rkey_str = ".." 70 + 71 with pytest.raises(InvalidRKeyError, match="reserved value"): 72 RKey(rkey_str) 73 74 def test_invalid_rkey_reserved_single_dot(self): 75 """Test that an RKey with '.' raises InvalidRKeyError.""" 76 rkey_str = "." 77 + 78 with pytest.raises(InvalidRKeyError, match="reserved value"): 79 RKey(rkey_str) 80 81 def test_invalid_rkey_invalid_characters(self): 82 """Test that an RKey with invalid characters raises InvalidRKeyError.""" 83 rkey_str = "alpha/beta" 84 + 85 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 86 RKey(rkey_str) 87 88 def test_invalid_rkey_hash_character(self): 89 """Test that an RKey with a hash character raises InvalidRKeyError.""" 90 rkey_str = "#extra" 91 + 92 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 93 RKey(rkey_str) 94 95 def test_invalid_rkey_at_character(self): 96 """Test that an RKey with an at character raises InvalidRKeyError.""" 97 rkey_str = "@handle" 98 + 99 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 100 RKey(rkey_str) 101 102 def test_invalid_rkey_space(self): 103 """Test that an RKey with a space raises InvalidRKeyError.""" 104 rkey_str = "any space" 105 + 106 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 107 RKey(rkey_str) 108 109 def test_invalid_rkey_plus_character(self): 110 """Test that an RKey with a plus character raises InvalidRKeyError.""" 111 rkey_str = "any+space" 112 + 113 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 114 RKey(rkey_str) 115 116 def test_invalid_rkey_brackets(self): 117 """Test that an RKey with brackets raises InvalidRKeyError.""" 118 rkey_str = "number[3]" 119 + 120 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 121 RKey(rkey_str) 122 123 def test_invalid_rkey_parentheses(self): 124 """Test that an RKey with parentheses raises InvalidRKeyError.""" 125 rkey_str = "number(3)" 126 + 127 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 128 RKey(rkey_str) 129 130 def test_invalid_rkey_quotes(self): 131 """Test that an RKey with quotes raises InvalidRKeyError.""" 132 rkey_str = '"quote"' 133 + 134 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 135 RKey(rkey_str) 136 137 def test_invalid_rkey_base64_padding(self): 138 """Test that an RKey with base64 padding raises InvalidRKeyError.""" 139 rkey_str = "dHJ1ZQ==" 140 + 141 with pytest.raises(InvalidRKeyError, match="contains invalid characters"): 142 RKey(rkey_str) 143 ··· 146 rkey_str = "3jui7kd54zh2y" 147 rkey1 = RKey(rkey_str) 148 rkey2 = RKey(rkey_str) 149 + 150 assert rkey1 == rkey2 151 assert rkey1 != "not an rkey object" 152 ··· 154 """Test RKey string representation.""" 155 rkey_str = "3jui7kd54zh2y" 156 rkey = RKey(rkey_str) 157 + 158 assert str(rkey) == rkey_str 159 160 ··· 164 def test_tid_creation_default(self): 165 """Test creating a TID with default parameters.""" 166 tid = TID() 167 + 168 assert isinstance(tid, TID) 169 assert isinstance(tid, RKey) 170 assert isinstance(tid.timestamp, datetime.datetime) ··· 176 """Test creating a TID with a specific timestamp.""" 177 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 178 tid = TID(time=timestamp) 179 + 180 assert tid.timestamp == timestamp 181 assert isinstance(tid.clockIdentifier, int) 182 assert 0 <= tid.clockIdentifier < 1024 ··· 185 """Test creating a TID with a specific clock identifier.""" 186 clock_id = 42 187 tid = TID(clockIdentifier=clock_id) 188 + 189 assert tid.clockIdentifier == clock_id 190 assert isinstance(tid.timestamp, datetime.datetime) 191 ··· 194 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 195 clock_id = 42 196 tid = TID(time=timestamp, clockIdentifier=clock_id) 197 + 198 assert tid.timestamp == timestamp 199 assert tid.clockIdentifier == clock_id 200 ··· 203 timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0) 204 clock_id = 42 205 tid = TID(time=timestamp, clockIdentifier=clock_id) 206 + 207 int_value = int(tid) 208 expected_value = int(timestamp.timestamp() * 1000000) * 1024 + clock_id 209 + 210 assert int_value == expected_value 211 212 def test_tid_string_representation(self): 213 """Test TID string representation.""" 214 tid = TID() 215 + 216 str_value = str(tid) 217 assert len(str_value) == 13 218 assert all(c in "234567abcdefghijklmnopqrstuvwxyz" for c in str_value) ··· 223 clock_id = 42 224 tid1 = TID(time=timestamp, clockIdentifier=clock_id) 225 tid2 = TID(time=timestamp, clockIdentifier=clock_id) 226 + 227 assert tid1 == tid2 228 229 def test_tid_equality_with_rkey(self): ··· 232 clock_id = 42 233 tid = TID(time=timestamp, clockIdentifier=clock_id) 234 rkey = RKey(str(tid)) 235 + 236 assert tid == rkey 237 238 def test_tid_inequality_with_different_object(self): 239 """Test TID inequality comparison with a different object type.""" 240 tid = TID() 241 + 242 assert tid != "not a tid object" 243 244 def test_tid_inequality_with_different_timestamp(self): ··· 248 clock_id = 42 249 tid1 = TID(time=timestamp1, clockIdentifier=clock_id) 250 tid2 = TID(time=timestamp2, clockIdentifier=clock_id) 251 + 252 assert tid1 != tid2 253 254 def test_tid_inequality_with_different_clock_id(self): ··· 258 clock_id2 = 43 259 tid1 = TID(time=timestamp, clockIdentifier=clock_id1) 260 tid2 = TID(time=timestamp, clockIdentifier=clock_id2) 261 + 262 assert tid1 != tid2 263 264 ··· 268 def test_import_tid_from_integer_default(self): 269 """Test importing a TID from integer with default value.""" 270 tid = importTIDfromInteger() 271 + 272 assert isinstance(tid, TID) 273 assert isinstance(tid.timestamp, datetime.datetime) 274 assert isinstance(tid.clockIdentifier, int) ··· 280 clock_id = 42 281 original_tid = TID(time=timestamp, clockIdentifier=clock_id) 282 int_value = int(original_tid) 283 + 284 imported_tid = importTIDfromInteger(int_value) 285 + 286 assert imported_tid.timestamp == timestamp 287 assert imported_tid.clockIdentifier == clock_id 288 289 def test_import_tid_from_base32_default(self): 290 """Test importing a TID from base32 with default value.""" 291 tid = importTIDfromBase32() 292 + 293 assert isinstance(tid, TID) 294 assert isinstance(tid.timestamp, datetime.datetime) 295 assert isinstance(tid.clockIdentifier, int) ··· 299 """Test importing a TID from base32 with a specific value.""" 300 original_tid = TID() 301 str_value = str(original_tid) 302 + 303 imported_tid = importTIDfromBase32(str_value) 304 + 305 + assert int(imported_tid) == int(original_tid)

+26 -18

tests/uri/test_uri.py

··· 10 11 def test_valid_uri_with_did(self): 12 """Test creating a URI with a valid DID.""" 13 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 14 uri = URI(uri_str) 15 - 16 assert str(uri) == uri_str 17 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 18 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 26 """Test creating a URI with a valid handle.""" 27 uri_str = "at://bnewbold.bsky.team/app.bsky.feed.post/3jwdwj2ctlk26" 28 uri = URI(uri_str) 29 - 30 assert str(uri) == uri_str 31 assert uri.authorityAsText == "bnewbold.bsky.team" 32 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 36 """Test creating a URI with only a collection.""" 37 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 38 uri = URI(uri_str) 39 - 40 assert str(uri) == uri_str 41 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 42 assert uri.path == ["app.bsky.feed.post"] ··· 46 """Test creating a URI with only an authority.""" 47 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur" 48 uri = URI(uri_str) 49 - 50 assert str(uri) == uri_str 51 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 52 assert uri.path == [] ··· 56 """Test creating a URI with query parameters.""" 57 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post?param1=value1&param2=value2" 58 uri = URI(uri_str) 59 - 60 assert uri.query == {"param1": ["value1"], "param2": ["value2"]} 61 assert uri.queryAsText == "param1%3Dvalue1%26param2%3Dvalue2" 62 ··· 64 """Test creating a URI with a fragment.""" 65 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26#$.some.json.path" 66 uri = URI(uri_str) 67 - 68 assert uri.fragment is not None 69 assert uri.fragmentAsText == "%24.some.json.path" 70 71 def test_invalid_uri_non_ascii_characters(self): 72 """Test that non-ASCII characters in URI raise InvalidURIError.""" 73 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/💩" 74 - 75 with pytest.raises(InvalidURIError, match="contains invalid characters"): 76 URI(uri_str) 77 ··· 80 # Create a URI that exceeds the 8000 character limit 81 long_path = "a" * 8000 82 uri_str = f"at://did:plc:z72i7hdynmk6r22z27h6tvur/{long_path}" 83 - 84 with pytest.raises(InvalidURIError, match="exceeds maximum length"): 85 URI(uri_str) 86 87 def test_invalid_uri_wrong_scheme(self): 88 """Test that a URI with wrong scheme raises InvalidURIError.""" 89 - uri_str = "https://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 90 - 0 0 91 with pytest.raises(InvalidURIError, match="invalid format"): 92 URI(uri_str) 93 94 def test_invalid_uri_trailing_slash(self): 95 """Test that a URI with trailing slash raises InvalidURIError.""" 96 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/" 97 - 98 with pytest.raises(InvalidURIError, match="cannot end with a slash"): 99 URI(uri_str) 100 101 def test_invalid_uri_with_userinfo(self): 102 """Test that a URI with userinfo raises InvalidURIError.""" 103 uri_str = "at://user:pass@did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 104 - 105 with pytest.raises(InvalidURIError, match="does not support user information"): 106 URI(uri_str) 107 108 def test_uri_equality(self): 109 """Test URI equality comparison.""" 110 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 111 uri1 = URI(uri_str) 112 uri2 = URI(uri_str) 113 - 114 assert uri1 == uri2 115 assert uri1 != "not a uri object" 116 117 def test_uri_string_representation(self): 118 """Test URI string representation.""" 119 - uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 0 0 120 uri = URI(uri_str) 121 - 122 - assert str(uri) == uri_str

··· 10 11 def test_valid_uri_with_did(self): 12 """Test creating a URI with a valid DID.""" 13 + uri_str = ( 14 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 15 + ) 16 uri = URI(uri_str) 17 + 18 assert str(uri) == uri_str 19 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 20 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 28 """Test creating a URI with a valid handle.""" 29 uri_str = "at://bnewbold.bsky.team/app.bsky.feed.post/3jwdwj2ctlk26" 30 uri = URI(uri_str) 31 + 32 assert str(uri) == uri_str 33 assert uri.authorityAsText == "bnewbold.bsky.team" 34 assert uri.path == ["app.bsky.feed.post", "3jwdwj2ctlk26"] ··· 38 """Test creating a URI with only a collection.""" 39 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 40 uri = URI(uri_str) 41 + 42 assert str(uri) == uri_str 43 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 44 assert uri.path == ["app.bsky.feed.post"] ··· 48 """Test creating a URI with only an authority.""" 49 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur" 50 uri = URI(uri_str) 51 + 52 assert str(uri) == uri_str 53 assert uri.authorityAsText == "did:plc:z72i7hdynmk6r22z27h6tvur" 54 assert uri.path == [] ··· 58 """Test creating a URI with query parameters.""" 59 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post?param1=value1&param2=value2" 60 uri = URI(uri_str) 61 + 62 assert uri.query == {"param1": ["value1"], "param2": ["value2"]} 63 assert uri.queryAsText == "param1%3Dvalue1%26param2%3Dvalue2" 64 ··· 66 """Test creating a URI with a fragment.""" 67 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26#$.some.json.path" 68 uri = URI(uri_str) 69 + 70 assert uri.fragment is not None 71 assert uri.fragmentAsText == "%24.some.json.path" 72 73 def test_invalid_uri_non_ascii_characters(self): 74 """Test that non-ASCII characters in URI raise InvalidURIError.""" 75 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/💩" 76 + 77 with pytest.raises(InvalidURIError, match="contains invalid characters"): 78 URI(uri_str) 79 ··· 82 # Create a URI that exceeds the 8000 character limit 83 long_path = "a" * 8000 84 uri_str = f"at://did:plc:z72i7hdynmk6r22z27h6tvur/{long_path}" 85 + 86 with pytest.raises(InvalidURIError, match="exceeds maximum length"): 87 URI(uri_str) 88 89 def test_invalid_uri_wrong_scheme(self): 90 """Test that a URI with wrong scheme raises InvalidURIError.""" 91 + uri_str = ( 92 + "https://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 93 + ) 94 + 95 with pytest.raises(InvalidURIError, match="invalid format"): 96 URI(uri_str) 97 98 def test_invalid_uri_trailing_slash(self): 99 """Test that a URI with trailing slash raises InvalidURIError.""" 100 uri_str = "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/" 101 + 102 with pytest.raises(InvalidURIError, match="cannot end with a slash"): 103 URI(uri_str) 104 105 def test_invalid_uri_with_userinfo(self): 106 """Test that a URI with userinfo raises InvalidURIError.""" 107 uri_str = "at://user:pass@did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post" 108 + 109 with pytest.raises(InvalidURIError, match="does not support user information"): 110 URI(uri_str) 111 112 def test_uri_equality(self): 113 """Test URI equality comparison.""" 114 + uri_str = ( 115 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 116 + ) 117 uri1 = URI(uri_str) 118 uri2 = URI(uri_str) 119 + 120 assert uri1 == uri2 121 assert uri1 != "not a uri object" 122 123 def test_uri_string_representation(self): 124 """Test URI string representation.""" 125 + uri_str = ( 126 + "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3jwdwj2ctlk26" 127 + ) 128 uri = URI(uri_str) 129 + 130 + assert str(uri) == uri_str