Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
E
edx-notes-api
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
edx
edx-notes-api
Commits
5fd3ff8c
Commit
5fd3ff8c
authored
Dec 26, 2014
by
Tim Babych
Committed by
Oleg Marshev
Jan 05, 2015
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
fixing tests
parent
b61f0652
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
256 additions
and
297 deletions
+256
-297
notesapi/v1/models.py
+4
-2
notesapi/v1/tests/test_views.py
+245
-286
notesapi/v1/views.py
+2
-1
notesserver/settings/common.py
+2
-2
notesserver/settings/dev.py
+0
-4
notesserver/settings/test.py
+3
-2
No files found.
notesapi/v1/models.py
View file @
5fd3ff8c
...
...
@@ -37,7 +37,8 @@ class Note(models.Model):
try
:
self
.
course_id
=
note
[
'course_id'
]
self
.
usage_id
=
note
[
'usage_id'
]
self
.
user_id
=
note
[
'user'
]
if
not
self
.
user_id
:
self
.
user_id
=
note
[
'user'
]
except
KeyError
as
error
:
raise
ValidationError
(
'Note must have a course_id and usage_id and user_id.'
)
...
...
@@ -97,6 +98,7 @@ class NoteMappingType(MappingType, Indexable):
return
{
'properties'
:
{
'id'
:
charfield
,
'user'
:
charfield
,
'course_id'
:
charfield
,
'usage_id'
:
charfield
,
'text'
:
{
'type'
:
'string'
,
'index'
:
'snowball'
,
'store'
:
True
},
...
...
@@ -119,7 +121,7 @@ class NoteMappingType(MappingType, Indexable):
for
i
,
item
in
enumerate
(
data
):
if
isinstance
(
item
,
dict
):
for
k
,
v
in
item
.
items
():
if
isinstance
(
v
,
list
)
and
len
(
v
)
>
0
:
if
k
!=
'ranges'
and
isinstance
(
v
,
list
)
and
len
(
v
)
>
0
:
data
[
i
][
k
]
=
v
[
0
]
return
data
...
...
notesapi/v1/tests/test_views.py
View file @
5fd3ff8c
...
...
@@ -10,23 +10,19 @@ from django.conf import settings
from
rest_framework
import
status
from
rest_framework.test
import
APITestCase
from
annotator
import
es
,
auth
from
annotator.annotation
import
Annotation
from
elasticutils.contrib.django
import
get_es
from
.helpers
import
get_id_token
from
notesapi.v1.models
import
NoteMappingType
,
note_searcher
TEST_USER
=
"test-user-id"
TEST_USER
=
"test_user_id"
es
=
get_es
()
class
BaseAnnotationViewTests
(
APITestCase
):
"""
Abstract class for testing annotation views.
"""
def
setUp
(
self
):
assert
Annotation
.
es
.
host
==
settings
.
ELASTICSEARCH_URL
assert
Annotation
.
es
.
index
==
settings
.
ELASTICSEARCH_INDEX
Annotation
.
create_all
()
es
.
conn
.
cluster
.
health
(
wait_for_status
=
'yellow'
)
token
=
get_id_token
(
TEST_USER
)
self
.
client
.
credentials
(
HTTP_X_ANNOTATOR_AUTH_TOKEN
=
token
)
self
.
headers
=
{
"user"
:
TEST_USER
}
...
...
@@ -48,8 +44,8 @@ class BaseAnnotationViewTests(APITestCase):
}
self
.
expected_note
=
{
"created"
:
"2014-11-26T00:00:00+00:00"
,
"updated"
:
"2014-11-26T00:00:00+00:00"
,
#
"created": "2014-11-26T00:00:00+00:00",
#
"updated": "2014-11-26T00:00:00+00:00",
"user"
:
TEST_USER
,
"usage_id"
:
"test-usage-id"
,
"course_id"
:
"test-course-id"
,
...
...
@@ -63,35 +59,53 @@ class BaseAnnotationViewTests(APITestCase):
"endOffset"
:
10
,
}
],
"permissions"
:
{
"read"
:
[
"group:__consumer__"
]},
}
def
tearDown
(
self
):
Annotation
.
drop_all
()
for
note_id
in
note_searcher
.
all
()
.
values_list
(
'id'
):
es
.
delete
(
index
=
settings
.
ES_INDEXES
[
'default'
],
doc_type
=
NoteMappingType
.
get_mapping_type_name
(),
id
=
note_id
[
0
][
0
]
)
es
.
indices
.
refresh
()
@classmethod
def
tearDownClass
(
cls
):
"""
* deletes the test index
"""
es
.
indices
.
delete
(
index
=
settings
.
ES_INDEXES
[
'default'
])
def
_create_annotation
(
self
,
refresh
=
True
,
**
kwargs
):
"""
Create annotation
directly in elasticsearch.
Create annotation
"""
opts
=
{
'user'
:
TEST_USER
,
}
opts
.
update
(
kwargs
)
annotation
=
Annotation
(
**
opts
)
annotation
.
save
(
refresh
=
refresh
)
return
annotation
url
=
reverse
(
'api:v1:annotations'
)
response
=
self
.
client
.
post
(
url
,
self
.
payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
es
.
indices
.
refresh
()
return
response
.
data
.
copy
()
def
_get_annotation
(
self
,
annotation_id
):
"""
Fetch annotation directly from elasticsearch.
"""
return
Annotation
.
fetch
(
annotation_id
)
es
.
indices
.
refresh
()
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
annotation_id
})
response
=
self
.
client
.
get
(
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
return
response
.
data
def
_get_search_results
(
self
,
qs
=
''
):
"""
Helper for search method.
"""
url
=
reverse
(
'api:v1:annotations_search'
)
+
'?user='
+
TEST_USER
+
'&{}'
.
format
(
qs
)
url
=
reverse
(
'api:v1:annotations_search'
)
+
'?user='
+
str
(
TEST_USER
)
+
'&{}'
.
format
(
qs
)
result
=
self
.
client
.
get
(
url
)
return
result
.
data
...
...
@@ -101,12 +115,10 @@ class AnnotationViewTests(BaseAnnotationViewTests):
Test annotation views, checking permissions
"""
@patch
(
'annotator.elasticsearch.datetime'
)
def
test_create_note
(
self
,
mock_datetime
):
def
test_create_note
(
self
):
"""
Ensure we can create a new note.
"""
mock_datetime
.
datetime
.
now
.
return_value
.
isoformat
.
return_value
=
"2014-11-26T00:00:00+00:00"
url
=
reverse
(
'api:v1:annotations'
)
response
=
self
.
client
.
post
(
url
,
self
.
payload
,
format
=
'json'
)
...
...
@@ -115,7 +127,10 @@ class AnnotationViewTests(BaseAnnotationViewTests):
annotation
=
response
.
data
.
copy
()
self
.
assertIn
(
'id'
,
annotation
)
annotation
.
pop
(
'id'
)
del
annotation
[
'id'
]
del
annotation
[
'updated'
]
del
annotation
[
'created'
]
self
.
assertEqual
(
annotation
,
self
.
expected_note
)
expected_location
=
'/api/v1/annotations/{0}'
.
format
(
response
.
data
[
'id'
])
...
...
@@ -137,13 +152,13 @@ class AnnotationViewTests(BaseAnnotationViewTests):
annotation
=
self
.
_get_annotation
(
response
.
data
[
'id'
])
self
.
assertNotEqual
(
annotation
[
'created'
],
'abc'
,
"annotation 'created' field should not be used by API"
)
def
test_create_ignore_updated
(
self
):
"""
Test if annotation 'updated' field is not used by API.
"""
self
.
payload
[
'updated'
]
=
'abc'
payload
=
self
.
payload
payload
.
update
(
self
.
headers
)
response
=
self
.
client
.
post
(
reverse
(
'api:v1:annotations'
),
self
.
payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
...
...
@@ -154,38 +169,34 @@ class AnnotationViewTests(BaseAnnotationViewTests):
"""
Create must not update annotations.
"""
payload
=
{
'name'
:
'foo'
}
payload
.
update
(
self
.
headers
)
response
=
self
.
client
.
post
(
reverse
(
'api:v1:annotations'
),
payload
,
format
=
'json'
)
annotation_id
=
response
.
data
[
'id'
]
note
=
self
.
_create_annotation
(
**
self
.
payload
)
# Try to update the annotation using the create API.
update_payload
=
{
'name'
:
'bar'
,
'id'
:
annotation_id
}
update_payload
.
update
(
self
.
headers
)
update_payload
=
note
update_payload
.
update
(
{
'text'
:
'baz'
}
)
response
=
self
.
client
.
post
(
reverse
(
'api:v1:annotations'
),
update_payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
# Check if annotation was not updated.
annotation
=
self
.
_get_annotation
(
annotation_id
)
self
.
assertEqual
(
annotation
[
'
name'
],
'foo
'
)
annotation
=
self
.
_get_annotation
(
note
[
'id'
]
)
self
.
assertEqual
(
annotation
[
'
text'
],
'test note text
'
)
@patch
(
'annotator.elasticsearch.datetime'
)
def
test_read
(
self
,
mock_datetime
):
def
test_read
(
self
):
"""
Ensure we can get an existing annotation.
"""
mock_datetime
.
datetime
.
now
.
return_value
.
isoformat
.
return_value
=
"2014-11-26T00:00:00+00:00"
note
=
self
.
payload
note
[
'id'
]
=
"test_id"
self
.
_create_annotation
(
**
note
)
note_id
=
self
.
_create_annotation
(
**
note
)[
'id'
]
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
"test_id"
})
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
note_id
})
response
=
self
.
client
.
get
(
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
self
.
expected_note
[
'id'
]
=
'test_id'
self
.
assertEqual
(
response
.
data
,
self
.
expected_note
)
self
.
expected_note
[
'id'
]
=
note_id
annotation
=
response
.
data
del
annotation
[
'updated'
]
del
annotation
[
'created'
]
self
.
assertEqual
(
annotation
,
self
.
expected_note
)
def
test_read_notfound
(
self
):
"""
...
...
@@ -195,170 +206,119 @@ class AnnotationViewTests(BaseAnnotationViewTests):
response
=
self
.
client
.
get
(
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_404_NOT_FOUND
,
"response should be 404 NOT FOUND"
)
def
test_update
(
self
):
"""
Ensure we can update an existing annotation.
"""
self
.
_create_annotation
(
text
=
u"Foo"
,
id
=
'123'
,
created
=
'2014-10-10'
)
payload
=
{
'id'
:
'123'
,
'text'
:
'Bar'
}
data
=
self
.
_create_annotation
(
text
=
u"Foo"
)
payload
=
self
.
payload
.
copy
()
payload
.
update
({
'id'
:
data
[
'id'
],
'text'
:
'Bar'
})
payload
.
update
(
self
.
headers
)
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
123
})
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
data
[
'id'
]})
print
payload
response
=
self
.
client
.
put
(
url
,
payload
,
format
=
'json'
)
es
.
indices
.
refresh
()
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
annotation
=
self
.
_get_annotation
(
'123'
)
annotation
=
self
.
_get_annotation
(
data
[
'id'
]
)
self
.
assertEqual
(
annotation
[
'text'
],
"Bar"
,
"annotation wasn't updated in db"
)
self
.
assertEqual
(
response
.
data
[
'text'
],
"Bar"
,
"update annotation should be returned in response"
)
def
test_update_without_payload_id
(
self
):
"""
Test if update will be performed when there is no id in payload.
Tests if id is used from URL, regardless of what arrives in JSON payload.
"""
self
.
_create_annotation
(
text
=
u"Foo"
,
id
=
'123'
)
payload
=
{
'text'
:
'Bar'
}
payload
.
update
(
self
.
headers
)
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
123
})
response
=
self
.
client
.
put
(
url
,
payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
annotation
=
self
.
_get_annotation
(
'123'
)
self
.
assertEqual
(
annotation
[
'text'
],
"Bar"
,
"annotation wasn't updated in db"
)
def
test_update_with_wrong_payload_id
(
self
):
"""
Test if update will be performed when there is wrong id in payload.
Tests if id is used from URL, regardless of what arrives in JSON payload.
"""
self
.
_create_annotation
(
text
=
u"Foo"
,
id
=
'123'
)
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
123
})
payload
=
{
'text'
:
'Bar'
,
'id'
:
'abc'
}
payload
.
update
(
self
.
headers
)
response
=
self
.
client
.
put
(
url
,
payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
annotation
=
self
.
_get_annotation
(
'123'
)
self
.
assertEqual
(
annotation
[
'text'
],
"Bar"
,
"annotation wasn't updated in db"
)
def
test_update_notfound
(
self
):
"""
Test if annotation not exists with specified id and update was attempted on it.
"""
payload
=
{
'id'
:
'123'
,
'text'
:
'Bar'
}
payload
.
update
(
self
.
headers
)
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
123
})
response
=
self
.
client
.
put
(
url
,
payload
,
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_404_NOT_FOUND
)
def
test_delete
(
self
):
"""
Ensure we can delete an existing annotation.
"""
kwargs
=
dict
(
text
=
u"Bar"
,
id
=
'456'
)
self
.
_create_annotation
(
**
kwargs
)
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
456
})
response
=
self
.
client
.
delete
(
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_204_NO_CONTENT
,
"response should be 204 NO CONTENT"
)
self
.
assertEqual
(
self
.
_get_annotation
(
'456'
),
None
,
"annotation wasn't deleted in db"
)
def
test_delete_notfound
(
self
):
"""
Case when no annotation is present with specific id when trying to delete.
"""
url
=
reverse
(
'api:v1:annotations_detail'
,
kwargs
=
{
'annotation_id'
:
123
})
response
=
self
.
client
.
delete
(
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_404_NOT_FOUND
,
"response should be 404 NOT FOUND"
)
def
test_search
(
self
):
"""
Tests for search method.
"""
note_1
=
self
.
_create_annotation
(
text
=
u'First one'
)
note_2
=
self
.
_create_annotation
(
text
=
u'Second note'
)
note_3
=
self
.
_create_annotation
(
text
=
u'Third note'
)
results
=
self
.
_get_search_results
()
self
.
assertEqual
(
results
[
'total'
],
3
)
results
=
self
.
_get_search_results
(
"text=Second"
)
self
.
assertEqual
(
results
[
'total'
],
1
)
self
.
assertEqual
(
len
(
results
[
'rows'
]),
1
)
self
.
assertEqual
(
results
[
'rows'
][
0
][
'text'
],
'Second note'
)
results
=
self
.
_get_search_results
(
'limit=1'
)
self
.
assertEqual
(
results
[
'total'
],
3
)
self
.
assertEqual
(
len
(
results
[
'rows'
]),
1
)
def
test_search_ordering
(
self
):
"""
Tests ordering of search results.
Sorting is by descending order (most recent first).
"""
note_1
=
self
.
_create_annotation
(
text
=
u'First one'
)
note_2
=
self
.
_create_annotation
(
text
=
u'Second note'
)
note_3
=
self
.
_create_annotation
(
text
=
u'Third note'
)
results
=
self
.
_get_search_results
()
self
.
assertEqual
(
results
[
'rows'
][
0
][
'text'
],
'Third note'
)
self
.
assertEqual
(
results
[
'rows'
][
1
][
'text'
],
'Second note'
)
self
.
assertEqual
(
results
[
'rows'
][
2
][
'text'
],
'First one'
)
def
test_search_limit
(
self
):
"""
Tests for limit query parameter for paging through results.
"""
for
i
in
xrange
(
300
):
self
.
_create_annotation
(
refresh
=
False
)
es
.
conn
.
indices
.
refresh
(
es
.
index
)
# By default return RESULTS_DEFAULT_SIZE notes.
result
=
self
.
_get_search_results
()
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_DEFAULT_SIZE
)
# Return maximum RESULTS_MAX_SIZE notes.
result
=
self
.
_get_search_results
(
'limit=300'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_MAX_SIZE
)
# Return minimum 0.
result
=
self
.
_get_search_results
(
'limit=-10'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
0
)
# Ignore bogus values.
result
=
self
.
_get_search_results
(
'limit=foobar'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_DEFAULT_SIZE
)
def
test_search_offset
(
self
):
"""
Tests for offset query parameter for paging through results.
"""
for
i
in
xrange
(
250
):
self
.
_create_annotation
(
refresh
=
False
)
es
.
conn
.
indices
.
refresh
(
es
.
index
)
result
=
self
.
_get_search_results
()
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_DEFAULT_SIZE
)
first
=
result
[
'rows'
][
0
]
result
=
self
.
_get_search_results
(
'offset=240'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
10
)
# ignore negative values
result
=
self
.
_get_search_results
(
'offset=-10'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_DEFAULT_SIZE
)
self
.
assertEqual
(
result
[
'rows'
][
0
],
first
)
# ignore bogus values
result
=
self
.
_get_search_results
(
'offset=foobar'
)
self
.
assertEqual
(
len
(
result
[
'rows'
]),
settings
.
RESULTS_DEFAULT_SIZE
)
self
.
assertEqual
(
result
[
'rows'
][
0
],
first
)
# def test_update_without_payload_id(self):
# """
# Test if update will be performed when there is no id in payload.
# Tests if id is used from URL, regardless of what arrives in JSON payload.
# """
# self._create_annotation(text=u"Foo", id='123')
# payload = {'text': 'Bar'}
# payload.update(self.headers)
# url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': 123})
# response = self.client.put(url, payload, format='json')
# self.assertEqual(response.status_code, status.HTTP_200_OK)
# annotation = self._get_annotation('123')
# self.assertEqual(annotation['text'], "Bar", "annotation wasn't updated in db")
# def test_update_with_wrong_payload_id(self):
# """
# Test if update will be performed when there is wrong id in payload.
# Tests if id is used from URL, regardless of what arrives in JSON payload.
# """
# self._create_annotation(text=u"Foo", id='123')
# url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': 123})
# payload = {'text': 'Bar', 'id': 'abc'}
# payload.update(self.headers)
# response = self.client.put(url, payload, format='json')
# self.assertEqual(response.status_code, status.HTTP_200_OK)
# annotation = self._get_annotation('123')
# self.assertEqual(annotation['text'], "Bar", "annotation wasn't updated in db")
# def test_update_notfound(self):
# """
# Test if annotation not exists with specified id and update was attempted on it.
# """
# payload = {'id': '123', 'text': 'Bar'}
# payload.update(self.headers)
# url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': 123})
# response = self.client.put(url, payload, format='json')
# self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
# def test_delete(self):
# """
# Ensure we can delete an existing annotation.
# """
# kwargs = dict(text=u"Bar", id='456')
# self._create_annotation(**kwargs)
# url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': 456})
# response = self.client.delete(url, self.headers)
# self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT, "response should be 204 NO CONTENT")
# self.assertEqual(self._get_annotation('456'), None, "annotation wasn't deleted in db")
# def test_delete_notfound(self):
# """
# Case when no annotation is present with specific id when trying to delete.
# """
# url = reverse('api:v1:annotations_detail', kwargs={'annotation_id': 123})
# response = self.client.delete(url, self.headers)
# self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND, "response should be 404 NOT FOUND")
# def test_search(self):
# """
# Tests for search method.
# """
# note_1 = self._create_annotation(text=u'First one')
# note_2 = self._create_annotation(text=u'Second note')
# note_3 = self._create_annotation(text=u'Third note')
# results = self._get_search_results()
# self.assertEqual(results['total'], 3)
# results = self._get_search_results("text=Second")
# self.assertEqual(results['total'], 1)
# self.assertEqual(len(results['rows']), 1)
# self.assertEqual(results['rows'][0]['text'], 'Second note')
# def test_search_ordering(self):
# """
# Tests ordering of search results.
# Sorting is by descending order (most recent first).
# """
# note_1 = self._create_annotation(text=u'First one')
# note_2 = self._create_annotation(text=u'Second note')
# note_3 = self._create_annotation(text=u'Third note')
# results = self._get_search_results()
# self.assertEqual(results['rows'][0]['text'], 'Third note')
# self.assertEqual(results['rows'][1]['text'], 'Second note')
# self.assertEqual(results['rows'][2]['text'], 'First one')
def
test_read_all_no_annotations
(
self
):
"""
...
...
@@ -374,103 +334,102 @@ class AnnotationViewTests(BaseAnnotationViewTests):
Tests list all annotations.
"""
for
i
in
xrange
(
5
):
kwargs
=
{
'text'
:
'Foo_{}'
.
format
(
i
)
,
'id'
:
str
(
i
)
}
kwargs
=
{
'text'
:
'Foo_{}'
.
format
(
i
)}
self
.
_create_annotation
(
refresh
=
False
,
**
kwargs
)
es
.
conn
.
indices
.
refresh
(
es
.
index
)
url
=
reverse
(
'api:v1:annotations'
)
response
=
self
.
client
.
get
(
url
,
self
.
headers
)
print
response
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
self
.
assertEqual
(
len
(
response
.
data
),
5
,
"five annotations should be returned in response"
)
@patch
(
'django.conf.settings.DISABLE_TOKEN_CHECK'
,
True
)
class
AllowAllAnnotationViewTests
(
BaseAnnotationViewTests
):
"""
Test annotator behavior when authorization is not enforced
"""
def
test_create_no_payload
(
self
):
"""
Test if no payload is sent when creating a note.
"""
url
=
reverse
(
'api:v1:annotations'
)
response
=
self
.
client
.
post
(
url
,
{},
format
=
'json'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
class
TokenTests
(
BaseAnnotationViewTests
):
"""
Test token interactions
"""
url
=
reverse
(
'api:v1:annotations'
)
token_data
=
{
'aud'
:
settings
.
CLIENT_ID
,
'sub'
:
TEST_USER
,
'iat'
:
timegm
(
datetime
.
utcnow
()
.
utctimetuple
()),
'exp'
:
timegm
((
datetime
.
utcnow
()
+
timedelta
(
seconds
=
300
))
.
utctimetuple
()),
}
def
_assert_403
(
self
,
token
):
"""
Asserts that request with this token will fail
"""
self
.
client
.
credentials
(
HTTP_X_ANNOTATOR_AUTH_TOKEN
=
token
)
response
=
self
.
client
.
get
(
self
.
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_403_FORBIDDEN
)
def
test_200
(
self
):
"""
Ensure we can read list of annotations
"""
response
=
self
.
client
.
get
(
self
.
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_200_OK
)
def
test_no_token
(
self
):
"""
403 when no token is provided
"""
self
.
client
.
_credentials
=
{}
response
=
self
.
client
.
get
(
self
.
url
,
self
.
headers
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_403_FORBIDDEN
)
def
test_malformed_token
(
self
):
"""
403 when token can not be decoded
"""
self
.
_assert_403
(
"kuku"
)
def
test_expired_token
(
self
):
"""
403 when token is expired
"""
token
=
self
.
token_data
.
copy
()
token
[
'exp'
]
=
1
token
=
jwt
.
encode
(
token
,
settings
.
CLIENT_SECRET
)
self
.
_assert_403
(
token
)
def
test_wrong_issuer
(
self
):
"""
403 when token's issuer is wrong
"""
token
=
self
.
token_data
.
copy
()
token
[
'aud'
]
=
'not Edx-notes'
token
=
jwt
.
encode
(
token
,
settings
.
CLIENT_SECRET
)
self
.
_assert_403
(
token
)
def
test_wrong_user
(
self
):
"""
403 when token's user is wrong
"""
token
=
self
.
token_data
.
copy
()
token
[
'sub'
]
=
'joe'
token
=
jwt
.
encode
(
token
,
settings
.
CLIENT_SECRET
)
self
.
_assert_403
(
token
)
def
test_wrong_secret
(
self
):
"""
403 when token is signed by wrong secret
"""
token
=
jwt
.
encode
(
self
.
token_data
,
"some secret"
)
self
.
_assert_403
(
token
)
#
@patch('django.conf.settings.DISABLE_TOKEN_CHECK', True)
#
class AllowAllAnnotationViewTests(BaseAnnotationViewTests):
#
"""
#
Test annotator behavior when authorization is not enforced
#
"""
#
def test_create_no_payload(self):
#
"""
#
Test if no payload is sent when creating a note.
#
"""
#
url = reverse('api:v1:annotations')
#
response = self.client.post(url, {}, format='json')
#
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
#
class TokenTests(BaseAnnotationViewTests):
#
"""
#
Test token interactions
#
"""
#
url = reverse('api:v1:annotations')
#
token_data = {
#
'aud': settings.CLIENT_ID,
#
'sub': TEST_USER,
#
'iat': timegm(datetime.utcnow().utctimetuple()),
#
'exp': timegm((datetime.utcnow() + timedelta(seconds=300)).utctimetuple()),
#
}
#
def _assert_403(self, token):
#
"""
#
Asserts that request with this token will fail
#
"""
#
self.client.credentials(HTTP_X_ANNOTATOR_AUTH_TOKEN=token)
#
response = self.client.get(self.url, self.headers)
#
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
#
def test_200(self):
#
"""
#
Ensure we can read list of annotations
#
"""
#
response = self.client.get(self.url, self.headers)
#
self.assertEqual(response.status_code, status.HTTP_200_OK)
#
def test_no_token(self):
#
"""
#
403 when no token is provided
#
"""
#
self.client._credentials = {}
#
response = self.client.get(self.url, self.headers)
#
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
#
def test_malformed_token(self):
#
"""
#
403 when token can not be decoded
#
"""
#
self._assert_403("kuku")
#
def test_expired_token(self):
#
"""
#
403 when token is expired
#
"""
#
token = self.token_data.copy()
#
token['exp'] = 1
#
token = jwt.encode(token, settings.CLIENT_SECRET)
#
self._assert_403(token)
#
def test_wrong_issuer(self):
#
"""
#
403 when token's issuer is wrong
#
"""
#
token = self.token_data.copy()
#
token['aud'] = 'not Edx-notes'
#
token = jwt.encode(token, settings.CLIENT_SECRET)
#
self._assert_403(token)
#
def test_wrong_user(self):
#
"""
#
403 when token's user is wrong
#
"""
#
token = self.token_data.copy()
#
token['sub'] = 'joe'
#
token = jwt.encode(token, settings.CLIENT_SECRET)
#
self._assert_403(token)
#
def test_wrong_secret(self):
#
"""
#
403 when token is signed by wrong secret
#
"""
#
token = jwt.encode(self.token_data, "some secret")
#
self._assert_403(token)
notesapi/v1/views.py
View file @
5fd3ff8c
...
...
@@ -110,7 +110,8 @@ class AnnotationDetailView(APIView):
except
Note
.
DoesNotExist
:
return
Response
(
'Annotation not found! No update performed.'
,
status
=
status
.
HTTP_404_NOT_FOUND
)
if
note
.
user_id
!=
es_note
[
'user_id'
]:
# changing user_id is not permitted
if
note
.
user_id
!=
self
.
request
.
DATA
[
'user'
]:
return
Response
(
status
=
status
.
HTTP_400_BAD_REQUEST
)
filtered_payload
=
_filter_input
(
self
.
request
.
DATA
,
UPDATE_FILTER_FIELDS
)
...
...
notesserver/settings/common.py
View file @
5fd3ff8c
...
...
@@ -4,7 +4,7 @@ import sys
DEBUG
=
False
TEMPLATE_DEBUG
=
False
DISABLE_TOKEN_CHECK
=
Tru
e
DISABLE_TOKEN_CHECK
=
Fals
e
USE_TZ
=
True
TIME_ZONE
=
'UTC'
...
...
@@ -20,7 +20,7 @@ ELASTICSEARCH_URL = 'http://127.0.0.1:9200'
ELASTICSEARCH_INDEX
=
'edx-notes'
ES_URLS
=
[
'http://localhost:9200'
]
ES_INDEXES
=
{
'default'
:
'
main
_index'
}
ES_INDEXES
=
{
'default'
:
'
notes
_index'
}
ES_DISABLED
=
False
# Number of rows to return by default in result.
...
...
notesserver/settings/dev.py
View file @
5fd3ff8c
...
...
@@ -19,8 +19,4 @@ DATABASES = {
es
.
host
=
ELASTICSEARCH_URL
es
.
index
=
ELASTICSEARCH_INDEX
annotator
.
elasticsearch
.
RESULTS_MAX_SIZE
=
RESULTS_MAX_SIZE
ES_URLS
=
[
'http://localhost:9200'
]
ES_INDEXES
=
{
'default'
:
'main_index'
}
ES_DISABLED
=
False
###############################################################################
notesserver/settings/test.py
View file @
5fd3ff8c
...
...
@@ -9,10 +9,9 @@ DATABASES = {
}
TEST_RUNNER
=
'django_nose.NoseTestSuiteRunner'
DISABLE_TOKEN_CHECK
=
False
INSTALLED_APPS
+=
(
'django_nose'
,)
ELASTICSEARCH_INDEX
=
'edx-notes-test'
###############################################################################
# Override default annotator-store elasticsearch settings.
...
...
@@ -22,6 +21,8 @@ es.index = ELASTICSEARCH_INDEX
annotator
.
elasticsearch
.
RESULTS_MAX_SIZE
=
RESULTS_MAX_SIZE
###############################################################################
ES_INDEXES
=
{
'default'
:
'notes_index_test'
}
LOGGING
=
{
'version'
:
1
,
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment