-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_network_flow.py
143 lines (121 loc) · 6.71 KB
/
test_network_flow.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#!/usr/bin/env python3
import json
import requests
import urllib3
from xsdata.formats.dataclass.client import Client
from xsdata.formats.dataclass.parsers import XmlParser
from xsdata.formats.dataclass.parsers.config import ParserConfig
from xsdata.formats.dataclass.serializers import XmlSerializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig
from configuration import *
from test_create_ojp_request import *
from map_nova_to_ojp import test_nova_to_ojp
from map_ojp_to_nova import test_ojp_fare_request_to_nova_request
from map_ojp_to_ojp import parse_ojp, map_ojp_trip_result_to_ojp_fare_request #, map_ojp_trip_result_to_ojp_refine_request
from nova import PreisAuskunftServicePortTypeSoapv14ErstellePreisAuskunft
from ojp import Ojp
from logger import log
ns_map = {'': 'http://www.siri.org.uk/siri', 'ojp': 'http://www.vdv.de/ojp'}
def call_ojp_2000(request_body):
try:
access_token = OJP_TOKEN
headers = {'Authorization': 'Bearer ' + access_token, 'Content-Type': 'application/xml; charset=utf-8' }
api_call_response = requests.post(OJP_URL_API, data=request_body.encode('utf-8'), headers=headers, verify=False)
api_call_response.encoding = api_call_response.apparent_encoding
return api_call_response.text
except Exception as e:
message = 'Failed: {:s}.'.format(str(e))
raise IOError(message, e)
class OAuth2Helper:
# Credits: https://developer.byu.edu/docs/consume-api/use-api/oauth-20/oauth-20-python-sample-code
def __init__(self, client_id, client_secret):
urllib3.disable_warnings()
self.client_id = client_id
self.client_secret = client_secret
self.current_token = None
def get_token(self, new_token=False):
if new_token or not self.current_token:
data = {'grant_type': 'client_credentials', 'scope': 'api://e1710a9f-d3e8-4751-b662-42f242e79f20/.default'}
access_token_response = requests.post(NOVA_URL_TOKEN, data=data, verify=False, allow_redirects=False,
auth=(self.client_id, self.client_secret))
self.current_token = json.loads(access_token_response.text)
return self.current_token['access_token']
def get_nova_client():
# TODO: It might be more elegant to cache this client in some way, but we would need to handle the expiry of the token
oauth_helper = OAuth2Helper(client_id=NOVA_CLIENT_ID, client_secret=NOVA_CLIENT_SECRET)
access_token = oauth_helper.get_token()
headers = {'Authorization': 'Bearer ' + access_token, "User-Agent": "OJP2NOVA/0.2" }
config = ParserConfig(
base_url=None,
process_xinclude=False,
fail_on_unknown_properties=False,
fail_on_unknown_attributes=False,
)
parser = XmlParser(config)
client = Client.from_service(PreisAuskunftServicePortTypeSoapv14ErstellePreisAuskunft, location=NOVA_URL_API, encoding="utf-8")
client.parser = parser
return client
def test_nova_request_reply(ojp: Ojp):
oauth_helper = OAuth2Helper(client_id=NOVA_CLIENT_ID, client_secret=NOVA_CLIENT_SECRET)
access_token = oauth_helper.get_token()
headers = {'Authorization': 'Bearer ' + access_token, "User-Agent": "OJP2NOVA/0.2"}
nova_request = test_ojp_fare_request_to_nova_request(ojp)
nova_client = get_nova_client()
nova_response = nova_client.send(nova_request, headers=headers)
if nova_response:
serializer_config = SerializerConfig(ignore_default_attributes=True, pretty_print=True)
serializer = XmlSerializer(serializer_config)
nova_response_xml = serializer.render(nova_response)
log('generated/nova_response.xml',nova_response_xml)
return nova_response
def check_configuration():
if (len(NOVA_CLIENT_SECRET)==0):
print("Secrets not set in the configuration")
exit(1)
if __name__ == '__main__':
#check configuration
ojp_trip_request_xml=''
check_configuration()
serializer_config = SerializerConfig(ignore_default_attributes=True, pretty_print=True)
serializer = XmlSerializer(serializer_config)
for rf in READFILE:
if (not READTRIPREQUESTFILE):
ojp_trip_request = test_create_ojp_trip_request_simple_1()
ojp_trip_request_xml = serializer.render(ojp_trip_request, ns_map=ns_map)
else:
inputfile = open(rf, 'r', encoding='utf-8')
ojp_trip_request_xml = inputfile.read()
inputfile.close()
log('generated/ojp_trip_request.xml', ojp_trip_request_xml)
try:
print (f"\n********************************************\n{rf}\n********************************************\n")
r = call_ojp_2000(ojp_trip_request_xml)
ojp_trip_result = parse_ojp(r)
# TODO ojp_trip_result.ojpresponse.service_delivery.ojptrip_delivery.status== false => error. However, I do only get to ojptrip_delivery.
ojp_trip_result_xml = serializer.render(ojp_trip_result, ns_map=ns_map)
log('generated/ojp_trip_reply.xml', ojp_trip_result_xml)
# TODO: This would only work in OJP v1.1
# ojp_refine_request = map_ojp_trip_result_to_ojp_refine_request(ojp_trip_result)
# ojp_refine_request_xml = serializer.render(ojp_refine_request, ns_map=ns_map)
# open('ojp_trip_refine_request.xml', 'w').write(ojp_refine_request_xml)
# r = call_ojp_2000(ojp_refine_request_xml)
ojp_trip_result = parse_ojp(r)
ojp_trip_result_xml = serializer.render(ojp_trip_result, ns_map=ns_map)
open('generated/ojp_trip_refine_reply.xml', 'w', encoding='utf-8').write(ojp_trip_result_xml)
ojp_fare_request = map_ojp_trip_result_to_ojp_fare_request(ojp_trip_result)
ojp_fare_request_xml = serializer.render(ojp_fare_request, ns_map=ns_map)
log('generated/ojp_fare_request.xml', ojp_fare_request_xml)
nova_response = test_nova_request_reply(ojp_fare_request)
if nova_response:
ojp_fare_result = test_nova_to_ojp(nova_response)
ojp_fare_result_xml = serializer.render(ojp_fare_result, ns_map=ns_map)
for fr1 in ojp_fare_result.fare_result:
for fr in fr1.trip_fare_result:
print("Legs: " + str(fr.from_trip_leg_id_ref) + "-" + str(fr.to_trip_leg_id_ref))
print(fr.fare_product)
print("\n")
log('generated/ojp_fare_result.xml', ojp_fare_result_xml)
except Exception as e:
# not yet really sophisticated handling of all other errors during the work (should be regular OJPDeliveries with OtherError set
log('generated/error_file.xml', str(e))
print (str(e))