127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
import datetime
|
|
import json
|
|
import logging.config
|
|
import os
|
|
import time
|
|
|
|
import requests
|
|
import zeep
|
|
from django.core.serializers.json import DjangoJSONEncoder
|
|
from lxml import etree
|
|
from zeep.transports import Transport
|
|
from zeep.wsse.username import UsernameToken
|
|
|
|
|
|
class SoapFilter(logging.Filter):
|
|
def filter(elf, record):
|
|
# ignore get WSDL record
|
|
if 'HTTP Post to' in record.message:
|
|
return True
|
|
if 'HTTP Response from' in record.message:
|
|
return True
|
|
return False
|
|
|
|
|
|
class SoapHandler(logging.Handler):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
for purpose in 'sent', 'received':
|
|
try:
|
|
os.remove('soap_' + purpose + '.xml')
|
|
except OSError:
|
|
pass
|
|
|
|
def format(self, record):
|
|
# cannot use formatter to do that
|
|
start = record.message.find('<soap')
|
|
soap = record.message[start:]
|
|
return soap
|
|
|
|
def emit(self, record):
|
|
msg = self.format(record)
|
|
xml = etree.fromstring(msg)
|
|
if 'HTTP Post to' in record.message:
|
|
path = 'soap_sent.xml'
|
|
elif 'HTTP Response from' in record.message:
|
|
path = 'soap_received.xml'
|
|
with open(path, 'ab') as soap_file:
|
|
soap_file.write(('---8<---' + '-' * 50 + '\n').encode())
|
|
soap_file.write(etree.tostring(xml, pretty_print=True, encoding='UTF-8'))
|
|
|
|
# keep all traces
|
|
path = 'all_' + path
|
|
with open(path, 'ab') as soap_file:
|
|
soap_file.write(('---8<---' + '-' * 50 + '\n').encode())
|
|
soap_file.write(etree.tostring(xml, pretty_print=True, encoding='UTF-8'))
|
|
|
|
|
|
def configure_logging(verbose):
|
|
if verbose > 1:
|
|
level = 'DEBUG'
|
|
else:
|
|
level = 'INFO'
|
|
logging.config.dictConfig(
|
|
{
|
|
'version': 1,
|
|
'filters': {
|
|
'soap_filter': {
|
|
'()': 'utils.SoapFilter',
|
|
},
|
|
},
|
|
'formatters': {'verbose': {'format': '%(name)s: %(message)s'}},
|
|
'handlers': {
|
|
'console': {
|
|
'level': level,
|
|
'class': 'logging.StreamHandler',
|
|
'formatter': 'verbose',
|
|
},
|
|
'soap': {
|
|
'level': level,
|
|
'class': 'utils.SoapHandler',
|
|
'filters': ['soap_filter'],
|
|
# no formatter used
|
|
},
|
|
},
|
|
'loggers': {
|
|
'zeep.transports': {
|
|
'level': 'DEBUG',
|
|
'propagate': True,
|
|
'handlers': ['console', 'soap'],
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def load_config(env):
|
|
with open('config.json') as conf_file:
|
|
config = json.load(conf_file)
|
|
return config[env.lower()]
|
|
|
|
|
|
def get_wsdl_url(env, service):
|
|
'''VPN is required to access test and integ environments'''
|
|
config = load_config(env)
|
|
return config['url'] + 'services/' + service.title() + 'Service?wsdl'
|
|
|
|
|
|
def get_client(env, service):
|
|
config = load_config(env)
|
|
settings = zeep.Settings(strict=False, xsd_ignore_sequence_order=True)
|
|
wsse = UsernameToken(config['login'], config['password'])
|
|
session = requests.Session()
|
|
session.verify = config['session_verify']
|
|
wsdl_url = get_wsdl_url(env, service)
|
|
print(wsdl_url)
|
|
|
|
transport = Transport(session=session)
|
|
client = zeep.Client(wsdl_url, transport=transport, wsse=wsse, settings=settings)
|
|
return client
|
|
|
|
|
|
class JSONEncoder(DjangoJSONEncoder):
|
|
def default(self, o):
|
|
if isinstance(o, time.struct_time):
|
|
o = datetime.datetime(*tuple(o)[:6])
|
|
return super().default(o)
|