105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
import json
|
|
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)
|
|
|
|
actual_matches = json.loads(actual_obj.warnings.warning[0].content[0])
|
|
|
|
expected_matches = json.loads(expected_obj.warnings.warning[0].content[0])
|
|
|
|
assert actual_matches == expected_matches, (
|
|
f"Expected warnings {expected_matches}, got {actual_matches}"
|
|
)
|
|
|
|
actual_capabilities = json.loads(actual_obj.echo_data)
|
|
expected_capabilities = json.loads(expected_obj.echo_data)
|
|
|
|
assert actual_capabilities == expected_capabilities, (
|
|
f"Expected echo data {expected_capabilities}, got {actual_capabilities}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_action_response_success():
|
|
server = AlpineBitsServer()
|
|
with open("test/test_data/Handshake-OTA_PingRQ.xml", "r", encoding="utf-8") as f:
|
|
request_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",
|
|
)
|
|
assert response.status_code == 200
|
|
assert "<OTA_PingRS" in response.xml_content
|
|
assert "<Success" in response.xml_content
|
|
assert "Version=" in response.xml_content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_action_response_version_arbitrary():
|
|
server = AlpineBitsServer()
|
|
with open("test/test_data/Handshake-OTA_PingRQ.xml", "r", encoding="utf-8") as f:
|
|
request_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="2022-10",
|
|
)
|
|
assert response.status_code == 200
|
|
assert "<OTA_PingRS" in response.xml_content
|
|
assert "Version=" in response.xml_content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_action_response_invalid_action():
|
|
server = AlpineBitsServer()
|
|
with open("test/test_data/Handshake-OTA_PingRQ.xml", "r", encoding="utf-8") as f:
|
|
request_xml = f.read()
|
|
client_info = AlpineBitsClientInfo(username="irrelevant", password="irrelevant")
|
|
response = await server.handle_request(
|
|
request_action_name="InvalidAction",
|
|
request_xml=request_xml,
|
|
client_info=client_info,
|
|
version="2024-10",
|
|
)
|
|
assert response.status_code == 400
|
|
assert "Error" in response.xml_content
|