-
Notifications
You must be signed in to change notification settings - Fork 2
/
test311.py
168 lines (149 loc) · 7.36 KB
/
test311.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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
import random
import unittest
import urllib
import types
import re
from xml.dom import minidom
SERVICE_ELEMENT_NAMES = ('service_code', 'metadata', 'type', 'keywords', 'group', 'service_name', 'description')
ATTRIBUTE_ELEMENT_NAMES = ('variable', 'code', 'datatype', 'required', 'datatype_description', 'order', 'description', 'description')
SERVICE_REQUEST_ELEMENT_NAMES = ('service_request_id', 'status', 'status_notes', 'service_name', 'service_code', 'description', 'agency_responsible', 'service_notice', 'requested_datetime', 'updated_datetime', 'expected_datetime', 'address', 'address_id', 'zipcode', 'lat', 'long', 'georss:point')
JURISDICTION_ID = 'dc.gov'
#URL_OPEN_311 = 'http://311.test.in.dc.gov/csr/Open311'
URL_OPEN_311 = 'http://311.dc.gov/csr/Open311'
API_KEY = '7d7859704a'
#JURISDICTION_ID = 'sfgov.org'
#URL_OPEN_311 = 'https://open311.sfgov.org/dev/V2'
#API_KEY = 'DT'
def parseServices(dom):
services = []
for serviceEl in dom.getElementsByTagName('service'):
services.append(parseService(serviceEl))
return services
def parseService(serviceEl):
service = {}
for elName in SERVICE_ELEMENT_NAMES:
print elName
service[elName] = serviceEl.getElementsByTagName(elName)[0].childNodes[0].data
return service
def parseServiceDefinition(dom):
sd = {}
sdEl = dom.getElementsByTagName('service_definition')[0]
sd['service_code'] = sdEl.getElementsByTagName('service_code')[0].childNodes[0].data
sd['attributes'] = []
for attrEl in sdEl.getElementsByTagName('attribute'):
attribute = {}
for attrElName in ATTRIBUTE_ELEMENT_NAMES:
attribute[attrElName] = \
attrEl.getElementsByTagName(attrElName)[0].childNodes[0].data
values = {}
for valEl in attrEl.getElementsByTagName('value'):
values[valEl.getElementsByTagName('key')[0].childNodes[0].data] = \
valEl.getElementsByTagName('name')[0].childNodes[0].data
attribute['values'] = values
sd['attributes'].append(attribute)
return sd
def parseServiceRequests(dom):
srs = []
for srEl in dom.getElementsByTagName('request'):
srs.append(parseServiceRequest(srEl))
return srs
def parseServiceRequest(srEl):
sr = {}
for elName in SERVICE_REQUEST_ELEMENT_NAMES:
els = srEl.getElementsByTagName(elName)
if len(els):
nodes = els[0].childNodes
if len(nodes):
sr[elName] = nodes[0].data
return sr
def parseSubmitResult(dom):
result = {}
return result
def getServices(opener):
url = "%s/services.xml?%s" % (URL_OPEN_311, urllib.urlencode({'jurisdiction_id':JURISDICTION_ID}))
dom = minidom.parse(opener.open(url))
f = open('services.xml', 'w')
f.write(dom.toxml())
f.close()
return parseServices(dom)
def getServiceDefinition(opener, serviceCode):
url = "%s/service_definition.xml?%s" % (URL_OPEN_311, urllib.urlencode({'jurisdiction_id':JURISDICTION_ID, 'service_code':serviceCode}))
dom = minidom.parse(opener.open(url))
f = open('service_definition_%s.xml' % serviceCode, 'w')
f.write(dom.toxml())
f.close()
return parseServiceDefinition(dom)
def getServiceRequests(opener, serviceCode):
url = "%s/requests.xml?%s" % (URL_OPEN_311, urllib.urlencode({'jurisdiction_id':JURISDICTION_ID, 'service_code':serviceCode}))
dom = minidom.parse(opener.open(url))
f = open('requests_%s.xml' % serviceCode, 'w')
f.write(dom.toxml())
f.close()
return parseServiceRequests(dom)
def submitServiceRequest(opener, serviceRequestDict):
serviceRequestDict.update({'jurisdiction_id':JURISDICTION_ID, 'api_key':API_KEY})
url = "%s/requests.xml" % (URL_OPEN_311)
data = opener.open(url, urllib.urlencode(serviceRequestDict))
dom = minidom.parse(data)
print 'CODE: %s' % data.getcode()
f = open('requests_submit.xml', 'w')
f.write(dom.toxml())
f.close()
return parseSubmitResult(dom)
class TestOpen311(unittest.TestCase):
def setUp(self):
self.opener = urllib.FancyURLopener({})
def testServices(self):
for s in getServices(self.opener):
self.assertTrue(re.search('.+', s.get('service_code', '')), 'service_code is required')
self.assertTrue(s['metadata'] in ('true', 'false'), 'metadata must be a boolean string')
self.assertTrue(s['type'] in ('realtime', 'batch', 'blackbox'), 'type is not a valid type')
self.assertTrue(re.search('.+', s.get('description', '')), 'description is required')
def testServiceDefinitions(self):
for s in getServices(self.opener):
if s['metadata'] == 'true' or s['metadata'] == 'True' or s['metadata'] is True:
sd = getServiceDefinition(self.opener, s['service_code'])
print 'Found service definition for service code: %s' % s['service_code']
self.assertEqual(sd['service_code'], s['service_code'], 'Service codes did not match.')
for attr in sd['attributes']:
self.assertTrue(attr['variable'] in ('true', 'false'), 'variable must be a boolean string')
self.assertTrue(re.search('.+', attr.get('code', '')), 'code is required')
self.assertTrue(attr['datatype'] in ('string', 'number', 'datetime', 'text', 'singlevaluelist', 'multivaluelist'), 'datatype is not a valid type')
self.assertTrue(attr['required'] in ('true', 'false'), 'required must be a boolean string')
self.assertTrue(type(eval(attr['order'])) in (types.IntType, types.FloatType), 'order = %s is not valid' % attr['order'])
def testServiceRequests(self):
for s in getServices(self.opener):
srs = getServiceRequests(self.opener, s['service_code'])
print 'Found %s service requests for service code: %s' % (len(srs), s['service_code'])
for sr in srs:
print '\tFound service request id: %s' % sr['service_request_id']
self._testServiceRequest(sr)
def _testServiceRequest(self, sr):
print sr
self.assertTrue(re.search('.+', sr.get('service_request_id', '')), 'service_request_id is required')
self.assertTrue(sr['status'] in ('open', 'closed'), 'status is not a valid type')
if sr.get('service_name', False):
self.assertTrue(re.search('.+', sr.get('service_name', '')), 'service_name is required %s' % sr)
if sr.get('service_code', False):
self.assertTrue(re.search('.+', sr.get('service_code', '')), 'service_code is required')
self.assertTrue(re.search('^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)Z?$', sr['requested_datetime']), 'requested_datetime is not a valid datetime')
if sr.get('updated_datetime', False):
self.assertTrue(re.search('^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)Z?$', sr['updated_datetime']), 'updated_datetime is not a valid datetime')
if sr.get('expected_datetime', False):
self.assertTrue(re.search('^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)Z?$', sr['expected_datetime']), 'expected_datetime is not a valid datetime')
self.assertTrue(re.search('.+', sr['address']), 'address is required')
if sr.get('zipcode', False):
self.assertEqual(type(eval(sr['zipcode'])), types.IntegerType, 'zipcode is not valid')
if sr.get('lat', False):
self.assertEqual(type(eval(sr['lat'])), types.FloatType, 'lat is not valid')
if sr.get('long', False):
self.assertEqual(type(eval(sr['long'])), types.FloatType, 'long is not valid')
if sr.get('georss:point', False):
lat, lon = sr['georss:point'].split(' ')
self.assertEqual(type(eval(lat)), types.FloatType, 'georss:point lat is not valid')
self.assertEqual(type(eval(lon)), types.FloatType, 'georss:point lon is not valid')
def testSubmitServiceRequest(self):
print submitServiceRequest(self.opener, {'service_code':'DMV66'}).read()
if __name__ == '__main__':
suite = unittest.TestLoader().loadTestsFromTestCase(TestOpen311)
unittest.TextTestRunner(verbosity=2).run(suite)