From 1f9c969e69e8bfc28bd04f27e582e00ddcc208fb Mon Sep 17 00:00:00 2001 From: Jonas Linter Date: Tue, 7 Oct 2025 09:00:01 +0200 Subject: [PATCH] formatting changes --- src/alpine_bits_python/alpinebits_server.py | 90 +++------------------ 1 file changed, 12 insertions(+), 78 deletions(-) diff --git a/src/alpine_bits_python/alpinebits_server.py b/src/alpine_bits_python/alpinebits_server.py index c68ccb0..fcaa85d 100644 --- a/src/alpine_bits_python/alpinebits_server.py +++ b/src/alpine_bits_python/alpinebits_server.py @@ -46,6 +46,14 @@ class HttpStatusCode(IntEnum): INTERNAL_SERVER_ERROR = 500 +def dump_json_for_xml(json_content: Any) -> str: + """Dump JSON content as a pretty-printed string for embedding in XML. + + Adds newlines before and after the JSON block for better readability in XML. + """ + return f"\n{json.dumps(json_content, indent=4)}\n" + + class AlpineBitsActionName(Enum): """Enum for AlpineBits action names with capability and request name mappings.""" @@ -290,9 +298,6 @@ class ServerCapabilities: self.create_capabilities_dict() return self.capability_dict - def get_capabilities_json(self) -> str: - """Get capabilities as formatted JSON string.""" - return json.dumps(self.get_capabilities_dict(), indent=2) def get_supported_actions(self) -> List[str]: """Get list of all supported action names.""" @@ -388,22 +393,20 @@ class PingAction(AlpineBitsAction): # Debug print to see what we matched # Create successful ping response with matched capabilities - capabilities_json = json.dumps(matching_capabilities, indent=4) + capabilities_json_str = dump_json_for_xml(matching_capabilities) + warning = OtaPingRs.Warnings.Warning( status=WarningStatus.ALPINEBITS_HANDSHAKE, type_value="11", - content=[capabilities_json], + content=[capabilities_json_str], ) warning_response = OtaPingRs.Warnings(warning=[warning]) - # remove action_OTA_Ping from version 2024-10 - all_capabilities = capabilities_dict - - client_response_echo_data = json.dumps(echo_data_client, indent=4) + client_response_echo_data = dump_json_for_xml(echo_data_client) response_ota_ping = OtaPingRs( version="7.000", @@ -707,10 +710,6 @@ class AlpineBitsServer: """Get server capabilities.""" return self.capabilities.get_capabilities_dict() - def get_capabilities_json(self) -> str: - """Get server capabilities as JSON.""" - return self.capabilities.get_capabilities_json() - async def handle_request( self, request_action_name: str, @@ -851,68 +850,3 @@ class AlpineBitsServer: return True -async def main(): - """Demonstrate the automatic capabilities discovery and request handling.""" - print("๐Ÿš€ AlpineBits Server Capabilities Discovery & Request Handling Demo") - print("=" * 70) - - # Create server instance - server = AlpineBitsServer() - - print("\n๐Ÿ“‹ Discovered Action Classes:") - print("-" * 30) - for capability_name, action_class in server.capabilities.action_registry.items(): - action_enum = AlpineBitsActionName.get_by_capability_name(capability_name) - request_name = action_enum.request_name if action_enum else "unknown" - print(f"โœ… {capability_name} -> {action_class.__name__}") - print(f" Request name: {request_name}") - - print( - f"\n๐Ÿ“Š Total Implemented Actions: {len(server.capabilities.get_supported_actions())}" - ) - - print("\n๐Ÿ” Generated Capabilities JSON:") - print("-" * 30) - capabilities_json = server.get_capabilities_json() - print(capabilities_json) - - print("\n๐ŸŽฏ Supported Request Names:") - print("-" * 30) - for request_name in server.get_supported_request_names(): - print(f" โ€ข {request_name}") - - print("\n๐Ÿงช Testing Request Handling:") - print("-" * 30) - - test_xml = "sample request" - - # Test different request formats - test_cases = [ - ("OTA_Ping:Handshaking", "2024-10"), - ("OTA_Read:GuestRequests", "2024-10"), - ("OTA_Read:GuestRequests", "2022-10"), - ("OTA_HotelAvailNotif", "2024-10"), - ("UnknownAction", "2024-10"), - ("OTA_Ping:Handshaking", "unsupported-version"), - ] - - for request_name, version in test_cases: - print(f"\n๏ฟฝ Testing: {request_name} (v{version})") - - # Check if supported first - is_supported = server.is_action_supported(request_name, version) - print(f" Supported: {is_supported}") - - # Handle the request - response = await server.handle_request(request_name, test_xml, version) - print(f" Status: {response.status_code}") - if len(response.xml_content) > 100: - print(f" Response: {response.xml_content[:100]}...") - else: - print(f" Response: {response.xml_content}") - - print("\nโœ… Demo completed successfully!") - - -if __name__ == "__main__": - asyncio.run(main())