Skip to content

Commit

Permalink
PB-974: separate test files for opentransapi and stationboard
Browse files Browse the repository at this point in the history 
Originally there were separate test files for opentransapi and the stationboard.
This is now again the case, also several test cases got lost, which were re-added
again.
  • Loading branch information
@hansmannj
hansmannj committed Dec 2, 2024
1 parent d21aa78 commit 5de9dbf
Show file tree
Hide file tree
Showing 2 changed files with 175 additions and 110 deletions.
95 changes: 95 additions & 0 deletions tests/integration/test_opentransapi.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
from datetime import datetime
from unittest.mock import Mock
from unittest.mock import patch

from pytz import timezone
from pyramid.httpexceptions import HTTPBadRequest
import requests_mock

from chsdi.lib.opentransapi import opentransapi
from chsdi.lib.opentransapi.opentransapi import format_time
from chsdi.views.stationboard import TransportView
from tests.integration import TestsBase
from tests.integration.helpers import generate_mock_response


class TestOpenTransApi(TestsBase):
def setUp(self):
self.mock_api_key = "dummy_api_key"
self.mock_url = "https://dummy-url.com"

@requests_mock.Mocker()
def test_stationboard(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
mock_departures = [
{
"id": "ch:1:sloid:30813::1",
"label": "Hogwarts Express",
"currentDate": now,
"departureDate": "2024-11-19T08:52:00Z",
"estimatedDate": "2024-11-19T08:52:00Z",
"destinationName": "Hogwarts",
"destinationId": "ch:1:sloid:91178::3",
}
]
mock_response = generate_mock_response(mock_departures, now)

mock_requests.post(
self.mock_url,
text=mock_response,
status_code=200
)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
results = api.get_departures(8501120, number_results=1)

self.assertEqual(len(results), 1)
self.assertEqual(results[0]["id"], "ch:1:sloid:30813::1")
self.assertEqual(results[0]["label"], "Hogwarts Express")
self.assertEqual(results[0]["currentDate"], format_time(now))
self.assertEqual(results[0]["departureDate"], format_time("2024-11-19T08:52:00Z"))
self.assertEqual(results[0]["estimatedDate"], format_time("2024-11-19T08:52:00Z"))
self.assertEqual(results[0]["destinationName"], "Hogwarts")
self.assertEqual(results[0]["destinationId"], "ch:1:sloid:91178::3")

@requests_mock.Mocker()
def test_stationboard_nonexisting_station(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
# empty response in order to trigger the OpenTransNoStationException
mock_response = generate_mock_response([], now)
mock_requests.post(self.mock_url, text=mock_response, status_code=200)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
with self.assertRaises(opentransapi.OpenTransNoStationException):
api.get_departures(999999)

@requests_mock.Mocker()
def test_stationboard_invalid_id(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
mock_response = generate_mock_response([], now)
mock_requests.post(self.mock_url, text=mock_response, status_code=200)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
with self.assertRaises(opentransapi.OpenTransNoStationException):
api.get_departures("invalid_id")

@patch('chsdi.views.stationboard.get_current_registry')
def test_invalid_limit_param(self, mock_get_current_registry):

mock_request = Mock()
mock_request.matched_route.name = 'stationboard'
mock_request.params = {}
mock_request.matchdict = {'id': '8501120'}

mock_registry = Mock()
mock_registry.settings = {
'opentrans_api_key': 'dummy_api_key',
'opentrans_url': 'https://dummy-url.com',
}
mock_get_current_registry.return_value = mock_registry

mock_request.params['limit'] = 'lalala'

# Expect an HTTPBadRequest exception
with self.assertRaises(HTTPBadRequest):
TransportView(mock_request)
190 changes: 80 additions & 110 deletions tests/integration/test_stationboard.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,139 +4,109 @@

from pytz import timezone
from pyramid.httpexceptions import HTTPBadRequest
from pyramid.httpexceptions import HTTPNotFound

import requests_mock

from chsdi.lib.opentransapi import opentransapi
from chsdi.lib.opentransapi.opentransapi import format_time
from chsdi.views.stationboard import TransportView
from tests.integration import TestsBase


def generate_mock_response(departures, now):
"""Generate a dynamic mock OJP response with correct structure."""
departure_events = "".join(
f"""
<StopEventResult>
<StopEvent>
<ThisCall>
<CallAtStop>
<siri:StopPointRef>{dep['id']}</siri:StopPointRef>
<StopPointName>
<Text>{dep['label']}</Text>
</StopPointName>
<ServiceDeparture>
<TimetabledTime>{dep['departureDate']}</TimetabledTime>
<EstimatedTime>{dep.get('estimatedDate', dep['departureDate'])}</EstimatedTime>
</ServiceDeparture>
</CallAtStop>
</ThisCall>
<Service>
<PublishedServiceName>
<Text>{dep['label']}</Text>
</PublishedServiceName>
<DestinationText>
<Text>{dep['destinationName']}</Text>
</DestinationText>
<DestinationStopPointRef>{dep['destinationId']}</DestinationStopPointRef>
</Service>
</StopEvent>
</StopEventResult>
"""
for dep in departures
)

mock_response = f"""<?xml version="1.0" ?>
<OJP xmlns:siri="http://www.siri.org.uk/siri" xmlns="http://www.vdv.de/ojp" version="2.0">
<OJPResponse>
<siri:ServiceDelivery>
<siri:ResponseTimestamp>{now}</siri:ResponseTimestamp>
<OJPStopEventDelivery>
<siri:ResponseTimestamp>{now}</siri:ResponseTimestamp>
{departure_events}
</OJPStopEventDelivery>
</siri:ServiceDelivery>
</OJPResponse>
</OJP>"""

return mock_response
from tests.integration.helpers import generate_mock_response


class TestStationboard(TestsBase):
def setUp(self):
self.mock_api_key = "dummy_api_key"
self.mock_url = "https://dummy-url.com"
self.mock_request = Mock()
self.mock_request.matched_route.name = 'stationboard'
self.mock_request.params = {}
self.mock_request.matchdict = {}

@requests_mock.Mocker()
def test_stationboard(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
self.mock_registry = Mock()
self.mock_registry.settings = {
'opentrans_api_key': 'dummy_api_key',
'opentrans_url': 'https://dummy-url.com',
}

@patch('chsdi.views.stationboard.get_current_registry')
def test_stationboard(self, mock_get_current_registry):
mock_get_current_registry.return_value = self.mock_registry
now = datetime.now(timezone('UTC')).strftime('%Y-%m-%dT%H:%M:%S.%f%z')
mock_departures = [
{
"id": "ch:1:sloid:30813::1",
"label": "Hogwarts Express",
"currentDate": now,
"departureDate": "2024-11-19T08:52:00Z",
"estimatedDate": "2024-11-19T08:52:00Z",
"departureDate": now,
"estimatedDate": now,
"destinationName": "Hogwarts",
"destinationId": "ch:1:sloid:91178::3",
}
]
mock_response = generate_mock_response(mock_departures, now)

mock_requests.post(
self.mock_url,
text=mock_response,
status_code=200
)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
results = api.get_departures(8501120, number_results=1)

self.assertEqual(len(results), 1)
self.assertEqual(results[0]["id"], "ch:1:sloid:30813::1")
self.assertEqual(results[0]["label"], "Hogwarts Express")
self.assertEqual(results[0]["currentDate"], format_time(now))
self.assertEqual(results[0]["departureDate"], format_time("2024-11-19T08:52:00Z"))
self.assertEqual(results[0]["estimatedDate"], format_time("2024-11-19T08:52:00Z"))
self.assertEqual(results[0]["destinationName"], "Hogwarts")
self.assertEqual(results[0]["destinationId"], "ch:1:sloid:91178::3")

@requests_mock.Mocker()
def test_stationboard_missing_station(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
mock_response = generate_mock_response([], now)
mock_requests.post(self.mock_url, text=mock_response, status_code=200)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
with self.assertRaises(opentransapi.OpenTransNoStationException):
api.get_departures(999999)

@requests_mock.Mocker()
def test_stationboard_invalid_id(self, mock_requests):
now = datetime.now(timezone('Europe/Zurich')).isoformat(timespec="microseconds")
mock_response = generate_mock_response([], now)
mock_requests.post(self.mock_url, text=mock_response, status_code=200)

api = opentransapi.OpenTrans(self.mock_api_key, self.mock_url)
with self.assertRaises(opentransapi.OpenTransNoStationException):
api.get_departures("invalid_id")
with requests_mock.Mocker() as m:
m.post(self.mock_registry.settings['opentrans_url'], text=mock_response)

self.mock_request.matchdict = {"id": "99999"}
view = TransportView(self.mock_request)
resp = view.get_departures()

self.assertEqual(len(resp), 1)
self.assertEqual(resp[0]["id"], "ch:1:sloid:30813::1")
self.assertEqual(resp[0]["label"], "Hogwarts Express")
self.assertEqual(resp[0]["departureDate"], format_time(now))
self.assertEqual(resp[0]["estimatedDate"], format_time(now))
self.assertEqual(resp[0]["destinationName"], "Hogwarts")
self.assertEqual(resp[0]["destinationId"], "ch:1:sloid:91178::3")

@patch('chsdi.views.stationboard.get_current_registry')
def test_invalid_limit_param(self, mock_get_current_registry):
def test_stationboard_nonexisting_id(self, mock_get_current_registry):
mock_get_current_registry.return_value = self.mock_registry
now = datetime.now(timezone('UTC')).strftime('%Y-%m-%dT%H:%M:%S.%f%z')
nonexisting_id = "99999"

# This will simulate the response of the OJP 2.0 API, when no station is found
mock_response = f"""<?xml version="1.0" ?>
<OJP xmlns:siri="http://www.siri.org.uk/siri" xmlns="http://www.vdv.de/ojp" version="2.0">
<OJPResponse>
<siri:ServiceDelivery>
<siri:ResponseTimestamp>{now}</siri:ResponseTimestamp>
<OJPStopEventDelivery>
<siri:ResponseTimestamp>{now}</siri:ResponseTimestamp>
<!-- No StopEventResult entries -->
</OJPStopEventDelivery>
</siri:ServiceDelivery>
</OJPResponse>
</OJP>"""

with requests_mock.Mocker() as m:
m.post(self.mock_registry.settings['opentrans_url'], text=mock_response)
self.mock_request.matchdict = {"id": nonexisting_id}
view = TransportView(self.mock_request)

with self.assertRaises(HTTPNotFound) as e:
view.get_departures()
self.assertEqual(str(e.exception), f"No data available for the station {nonexisting_id}.")

mock_request = Mock()
mock_request.matched_route.name = 'stationboard'
mock_request.params = {}
mock_request.matchdict = {'id': '8501120'}
@patch('chsdi.views.stationboard.get_current_registry')
def test_stationboard_wrong_id(self, mock_get_current_registry):
mock_get_current_registry.return_value = self.mock_registry
wrong_id = "lalala"
self.mock_request.matchdict = {"id": wrong_id}

mock_registry = Mock()
mock_registry.settings = {
'opentrans_api_key': 'dummy_api_key',
'opentrans_url': 'https://dummy-url.com',
}
mock_get_current_registry.return_value = mock_registry
with self.assertRaises(HTTPBadRequest) as e:
_ = TransportView(self.mock_request)
self.assertEqual(str(e.exception), "The id must be an integer.")

mock_request.params['limit'] = 'lalala'
@patch('chsdi.views.stationboard.get_current_registry')
def test_stationboard_wrong_limit(self, mock_get_current_registry):
mock_get_current_registry.return_value = self.mock_registry
self.mock_request.params["limit"] = "lalala"
self.mock_request.matchdict = {"id": "123"}

# Expect an HTTPBadRequest exception
with self.assertRaises(HTTPBadRequest):
TransportView(mock_request)
with self.assertRaises(HTTPBadRequest) as e:
_ = TransportView(self.mock_request)
self.assertEqual(str(e.exception), "The limit parameter must be an integer.")

@patch('chsdi.views.stationboard.get_current_registry')
def test_stationboard_with_cb(self, mock_get_current_registry):
pass

0 comments on commit 5de9dbf

Please sign in to comment.