notif_report #3
@@ -260,6 +260,25 @@ class ServerCapabilities:
|
||||
|
||||
self.capability_dict = {"versions": list(versions_dict.values())}
|
||||
|
||||
|
||||
# filter duplicates in actions for each version
|
||||
for version in self.capability_dict["versions"]:
|
||||
seen_actions = set()
|
||||
unique_actions = []
|
||||
for action in version["actions"]:
|
||||
if action["action"] not in seen_actions:
|
||||
seen_actions.add(action["action"])
|
||||
unique_actions.append(action)
|
||||
version["actions"] = unique_actions
|
||||
|
||||
# remove action_OTA_Ping from version 2024-10
|
||||
for version in self.capability_dict["versions"]:
|
||||
if version["version"] == "2024-10":
|
||||
version["actions"] = [
|
||||
action for action in version["actions"]
|
||||
if action.get("action") != "action_OTA_Ping"
|
||||
]
|
||||
|
||||
return None
|
||||
|
||||
def get_capabilities_dict(self) -> Dict:
|
||||
@@ -379,12 +398,17 @@ class PingAction(AlpineBitsAction):
|
||||
|
||||
warning_response = OtaPingRs.Warnings(warning=[warning])
|
||||
|
||||
all_capabilities = server_capabilities.get_capabilities_json()
|
||||
|
||||
# remove action_OTA_Ping from version 2024-10
|
||||
all_capabilities = capabilities_dict
|
||||
|
||||
|
||||
all_capabilities_json = json.dumps(all_capabilities, indent=2)
|
||||
|
||||
response_ota_ping = OtaPingRs(
|
||||
version="7.000",
|
||||
warnings=warning_response,
|
||||
echo_data=all_capabilities,
|
||||
echo_data=all_capabilities_json,
|
||||
success="",
|
||||
)
|
||||
|
||||
|
||||
@@ -1,6 +1,47 @@
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
from alpine_bits_python.alpinebits_server import AlpineBitsServer, AlpineBitsClientInfo
|
||||
import re
|
||||
from xsdata_pydantic.bindings import XmlParser
|
||||
from alpine_bits_python.generated.alpinebits import OtaPingRs
|
||||
|
||||
|
||||
|
||||
|
||||
def extract_relevant_sections(xml_string):
|
||||
# Remove version attribute value, keep only presence
|
||||
# Use the same XmlParser as AlpineBitsServer
|
||||
parser = XmlParser()
|
||||
obj = parser.from_string(xml_string, OtaPingRs)
|
||||
return obj
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ping_action_response_matches_expected():
|
||||
|
||||
with open("test/test_data/Handshake-OTA_PingRQ.xml", "r", encoding="utf-8") as f:
|
||||
server = AlpineBitsServer()
|
||||
with open("test/test_data/Handshake-OTA_PingRQ.xml", "r", encoding="utf-8") as f:
|
||||
request_xml = f.read()
|
||||
with open("test/test_data/Handshake-OTA_PingRS.xml", "r", encoding="utf-8") as f:
|
||||
expected_xml = f.read()
|
||||
client_info = AlpineBitsClientInfo(username="irrelevant", password="irrelevant")
|
||||
response = await server.handle_request(
|
||||
request_action_name="OTA_Ping:Handshaking",
|
||||
request_xml=request_xml,
|
||||
client_info=client_info,
|
||||
version="2024-10"
|
||||
)
|
||||
actual_obj = extract_relevant_sections(response.xml_content)
|
||||
expected_obj = extract_relevant_sections(expected_xml)
|
||||
# log failures to xml files in test_output for easier debugging
|
||||
if actual_obj != expected_obj:
|
||||
with open("test/test_output/actual_ping_response.xml", "w", encoding="utf-8") as f:
|
||||
f.write(response.xml_content)
|
||||
with open("test/test_output/expected_ping_response.xml", "w", encoding="utf-8") as f:
|
||||
f.write(expected_xml)
|
||||
|
||||
assert actual_obj == expected_obj
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ping_action_response_success():
|
||||
|
||||
@@ -59,9 +59,6 @@
|
||||
},
|
||||
{
|
||||
"action": "action_OTA_HotelResNotif_GuestRequests"
|
||||
},
|
||||
{
|
||||
"action": "action_OTA_Read"
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -76,9 +73,6 @@
|
||||
},
|
||||
{
|
||||
"action": "action_OTA_HotelResNotif_GuestRequests"
|
||||
},
|
||||
{
|
||||
"action": "action_OTA_Read"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user