forked from jacobcr/python-gcm-dummy
-
Notifications
You must be signed in to change notification settings - Fork 0
/
android_mock.py
105 lines (82 loc) · 2.61 KB
/
android_mock.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
#!/usr/bin/python2.7
#-*- coding: utf-8 -*-
#
# Copyright 2013 Telefonica Digital. All rights reserved.
#
# Authors:
# Antonio Robres < [email protected]> - 2013
#
from bottle import run, post, request, error, route
import ujson
from optparse import OptionParser
from collections import deque
response_template = {'failure': 0,
'canonical_ids': 0,
'success': 0,
'multicast_id': 6616353152392206975,
'results': []
}
responses_saved = deque()
requests_handled = deque()
@error(404)
def error404(error):
return 'Nothing here, sorry'
@post("/")
def return_response():
'''
Response with a GCM response mocked. If there are saved responses in the dict "responses_saved", send the first
response and deletes it. If there isn't responses saved before, send a GCM response with all the receivers as
success.
:return: json body with response saved or with all the receivers success.
'''
data = request.body.readline()
body_received = ujson.loads(data)
requests_handled.append(body_received)
response = response_template
if not responses_saved:
response['success'] = len(body_received['registration_ids'])
for x in range(len(body_received['registration_ids'])):
response['results'].append(({u'message_id': u'0:1370674827295849'}))
else:
response = responses_saved.popleft()
return ujson.dumps(response)
@post("/save_response")
def save_response():
'''
Store the body provided in a dict to be used later.
:return: 200 OK
'''
data = request.body.readline()
body_received = ujson.loads(data)
responses_saved.append(body_received)
return 'OK'
@route("/reset_responses")
def reset_responses():
responses_saved.clear()
return 'Responses deleted'
@route("/reset_stats")
def reset_stats():
requests_handled.clear()
return 'Stats reset'
@route("/stats")
def stats():
body = {'num_requests': len(requests_handled),
'requests': list(requests_handled)}
return ujson.dumps(body)
def main():
parser = OptionParser()
parser.add_option(
"-p", "--port",
dest="port",
help="Server port [%default]",
default=8082)
parser.add_option(
"-b", "--bind_address",
dest="bind",
help="Bind addreess [%default]",
default="0.0.0.0")
(options, args) = parser.parse_args()
print "Starting server on %s:%s" % (options.bind, int(options.port))
run(host=options.bind, port=options.port, debug=True)
if __name__ == "__main__":
main()