Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F81962976
test_receivers.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Sun, Sep 8, 20:36
Size
5 KB
Mime Type
text/x-python
Expires
Tue, Sep 10, 20:36 (2 d)
Engine
blob
Format
Raw Data
Handle
20639264
Attached To
R3600 invenio-infoscience
test_receivers.py
View Options
# -*- coding: utf-8 -*-
##
## This file is part of Invenio.
## Copyright (C) 2014 CERN.
##
## Invenio is free software; you can redistribute it and/or
## modify it under the terms of the GNU General Public License as
## published by the Free Software Foundation; either version 2 of the
## License, or (at your option) any later version.
##
## Invenio is distributed in the hope that it will be useful, but
## WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
## General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with Invenio; if not, write to the Free Software Foundation, Inc.,
## 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
from
__future__
import
absolute_import
import
json
from
flask_registry
import
RegistryError
from
invenio.testsuite
import
InvenioTestCase
from
..models
import
Event
,
Receiver
,
InvalidPayload
,
CeleryReceiver
,
\
ReceiverDoesNotExists
,
InvalidSignature
from
..signatures
import
get_hmac
from
invenio.celery
import
celery
class
ReceiverTestCase
(
InvenioTestCase
):
def
setUp
(
self
):
self
.
called
=
0
self
.
payload
=
None
self
.
user_id
=
None
# Force synchronously task running
celery
.
conf
[
'CELERY_ALWAYS_EAGER'
]
=
True
@celery.task
(
ignore_result
=
True
)
def
test_task
(
event_state
):
e
=
Event
()
e
.
__setstate__
(
event_state
)
self
.
called
+=
1
self
.
payload
=
e
.
payload
self
.
user_id
=
e
.
user_id
self
.
task_callable
=
test_task
def
tearDown
(
self
):
self
.
called
=
None
self
.
payload
=
None
self
.
user_id
=
None
def
callable
(
self
,
event
):
self
.
called
+=
1
self
.
payload
=
event
.
payload
self
.
user_id
=
event
.
user_id
def
callable_wrong_signature
(
self
):
self
.
called
+=
1
def
test_receiver_registration
(
self
):
r
=
Receiver
(
self
.
callable
)
r_invalid
=
Receiver
(
self
.
callable_wrong_signature
)
Receiver
.
register
(
'test-receiver'
,
r
)
Receiver
.
register
(
'test-invalid'
,
r_invalid
)
assert
'test-receiver'
in
Receiver
.
all
()
assert
Receiver
.
get
(
'test-receiver'
)
==
r
# Double registration
self
.
assertRaises
(
RegistryError
,
Receiver
.
register
,
'test-receiver'
,
r
)
Receiver
.
unregister
(
'test-receiver'
)
assert
'test-receiver'
not
in
Receiver
.
all
()
Receiver
.
register
(
'test-receiver'
,
r
)
# JSON payload parsing
payload
=
json
.
dumps
(
dict
(
somekey
=
'somevalue'
))
headers
=
[(
'Content-Type'
,
'application/json'
)]
with
self
.
app
.
test_request_context
(
headers
=
headers
,
data
=
payload
):
r
.
consume_event
(
2
)
assert
self
.
called
==
1
assert
self
.
payload
==
json
.
loads
(
payload
)
assert
self
.
user_id
==
2
self
.
assertRaises
(
TypeError
,
r_invalid
.
consume_event
,
2
)
assert
self
.
called
==
1
# Form encoded values payload parsing
payload
=
dict
(
somekey
=
'somevalue'
)
with
self
.
app
.
test_request_context
(
method
=
'POST'
,
data
=
payload
):
r
.
consume_event
(
2
)
assert
self
.
called
==
2
assert
self
.
payload
==
dict
(
somekey
=
[
'somevalue'
])
# Test invalid post data
with
self
.
app
.
test_request_context
(
method
=
'POST'
,
data
=
"invaliddata"
):
self
.
assertRaises
(
InvalidPayload
,
r
.
consume_event
,
2
)
# Test Celery Receiver
rcelery
=
CeleryReceiver
(
self
.
task_callable
)
CeleryReceiver
.
register
(
'celery-receiver'
,
rcelery
)
# Form encoded values payload parsing
payload
=
dict
(
somekey
=
'somevalue'
)
with
self
.
app
.
test_request_context
(
method
=
'POST'
,
data
=
payload
):
rcelery
.
consume_event
(
1
)
assert
self
.
called
==
3
assert
self
.
payload
==
dict
(
somekey
=
[
'somevalue'
])
assert
self
.
user_id
==
1
def
test_unknown_receiver
(
self
):
self
.
assertRaises
(
ReceiverDoesNotExists
,
Receiver
.
get
,
'unknown'
)
def
test_hookurl
(
self
):
r
=
Receiver
(
self
.
callable
)
Receiver
.
register
(
'test-receiver'
,
r
)
assert
Receiver
.
get_hook_url
(
'test-receiver'
,
'token'
)
!=
\
'http://test.local/?access_token=token'
self
.
app
.
config
[
'WEBHOOKS_DEBUG_RECEIVER_URLS'
]
=
{
'test-receiver'
:
'http://test.local/?access_token=
%(token)s
'
}
assert
Receiver
.
get_hook_url
(
'test-receiver'
,
'token'
)
==
\
'http://test.local/?access_token=token'
def
test_signature_checking
(
self
):
"""
webhooks - checks signatures for payload
"""
r
=
Receiver
(
self
.
callable
,
signature
=
'X-Hub-Signature'
)
Receiver
.
register
(
'test-receiver-sign'
,
r
)
# check correct signature
payload
=
json
.
dumps
(
dict
(
somekey
=
'somevalue'
))
headers
=
[(
'Content-Type'
,
'application/json'
),
(
'X-Hub-Signature'
,
get_hmac
(
payload
))]
with
self
.
app
.
test_request_context
(
headers
=
headers
,
data
=
payload
):
r
.
consume_event
(
2
)
assert
self
.
payload
==
json
.
loads
(
payload
)
# check signature with prefix
headers
=
[(
'Content-Type'
,
'application/json'
),
(
'X-Hub-Signature'
,
'sha1='
+
get_hmac
(
payload
))]
with
self
.
app
.
test_request_context
(
headers
=
headers
,
data
=
payload
):
r
.
consume_event
(
2
)
assert
self
.
payload
==
json
.
loads
(
payload
)
# check incorrect signature
headers
=
[(
'Content-Type'
,
'application/json'
),
(
'X-Hub-Signature'
,
get_hmac
(
"somevalue"
))]
with
self
.
app
.
test_request_context
(
headers
=
headers
,
data
=
payload
):
self
.
assertRaises
(
InvalidSignature
,
r
.
consume_event
,
2
)
Event Timeline
Log In to Comment