This commit is contained in:
Rehan Dalal 2017-04-28 13:39:16 -04:00
Родитель 71792095e0
Коммит 335f125ac8
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 410D198EEF339E0B
3 изменённых файлов: 1046 добавлений и 18 удалений

Просмотреть файл

@ -0,0 +1,217 @@
from django.conf import settings
from django.db.models import Q
import django_filters
from rest_framework import generics, permissions, status, viewsets
from rest_framework.decorators import detail_route, list_route
from rest_framework.exceptions import NotFound
from rest_framework.response import Response
from normandy.base.api import UpdateOrCreateModelViewSet
from normandy.base.api.filters import CaseInsensitiveBooleanFilter
from normandy.base.api.mixins import CachingViewsetMixin
from normandy.base.api.permissions import AdminEnabledOrReadOnly
from normandy.base.api.renderers import JavaScriptRenderer
from normandy.base.decorators import api_cache_control, reversion_transaction
from normandy.recipes.models import (
Action,
ApprovalRequest,
Recipe,
RecipeRevision
)
from normandy.recipes.api.v1.serializers import (
SignedRecipeSerializer,
)
from normandy.recipes.api.v2.serializers import (
ActionSerializer,
ApprovalRequestSerializer,
RecipeRevisionSerializer,
RecipeSerializer,
)
class ActionViewSet(CachingViewsetMixin, viewsets.ReadOnlyModelViewSet):
"""Viewset for viewing recipe actions."""
queryset = Action.objects.all()
serializer_class = ActionSerializer
class ActionImplementationView(generics.RetrieveAPIView):
"""
Retrieves the implementation code for an action. Raises a 404 if the
given hash doesn't match the hash we've stored.
"""
queryset = Action.objects.all()
lookup_field = 'id'
permission_classes = []
renderer_classes = [JavaScriptRenderer]
@api_cache_control(max_age=settings.ACTION_IMPLEMENTATION_CACHE_TIME)
def retrieve(self, request, id, impl_hash):
action = self.get_object()
if impl_hash != action.implementation_hash:
raise NotFound('Hash does not match current stored action.')
return Response(action.implementation)
class RecipeFilters(django_filters.FilterSet):
enabled = CaseInsensitiveBooleanFilter(name='enabled', lookup_expr='eq')
class Meta:
model = Recipe
fields = ['latest_revision__action', 'enabled']
class RecipeViewSet(CachingViewsetMixin, UpdateOrCreateModelViewSet):
"""Viewset for viewing and uploading recipes."""
queryset = Recipe.objects.all()
serializer_class = RecipeSerializer
filter_class = RecipeFilters
permission_classes = [
permissions.DjangoModelPermissionsOrAnonReadOnly,
AdminEnabledOrReadOnly,
]
def get_queryset(self):
queryset = self.queryset
if self.request.GET.get('status') == 'enabled':
queryset = queryset.filter(enabled=True)
elif self.request.GET.get('status') == 'disabled':
queryset = queryset.filter(enabled=False)
if 'channels' in self.request.GET:
channels = self.request.GET.get('channels').split(',')
queryset = queryset.filter(latest_revision__channels__slug__in=channels)
if 'countries' in self.request.GET:
countries = self.request.GET.get('countries').split(',')
queryset = queryset.filter(latest_revision__countries__code__in=countries)
if 'locales' in self.request.GET:
locales = self.request.GET.get('locales').split(',')
queryset = queryset.filter(latest_revision__locales__code__in=locales)
if 'text' in self.request.GET:
text = self.request.GET.get('text')
queryset = queryset.filter(Q(latest_revision__name__contains=text) |
Q(latest_revision__extra_filter_expression__contains=text))
return queryset
@list_route(methods=['GET'])
@api_cache_control()
def signed(self, request, pk=None):
recipes = self.filter_queryset(self.get_queryset()).exclude(signature=None)
serializer = SignedRecipeSerializer(recipes, many=True)
return Response(serializer.data)
@detail_route(methods=['GET'])
@api_cache_control()
def history(self, request, pk=None):
recipe = self.get_object()
serializer = RecipeRevisionSerializer(recipe.revisions.all(), many=True,
context={'request': request})
return Response(serializer.data)
@reversion_transaction
@detail_route(methods=['POST'])
def enable(self, request, pk=None):
recipe = self.get_object()
recipe.enabled = True
try:
recipe.save()
except Recipe.NotApproved as e:
return Response({'enabled': str(e)}, status=status.HTTP_409_CONFLICT)
return Response(RecipeSerializer(recipe).data)
@reversion_transaction
@detail_route(methods=['POST'])
def disable(self, request, pk=None):
recipe = self.get_object()
recipe.enabled = False
recipe.save()
return Response(RecipeSerializer(recipe).data)
class RecipeRevisionViewSet(viewsets.ReadOnlyModelViewSet):
queryset = RecipeRevision.objects.all()
serializer_class = RecipeRevisionSerializer
permission_classes = [
AdminEnabledOrReadOnly,
permissions.DjangoModelPermissionsOrAnonReadOnly,
]
@detail_route(methods=['POST'])
def request_approval(self, request, pk=None):
revision = self.get_object()
if revision.approval_status is not None:
return Response({'error': 'This revision already has an approval request.'},
status=status.HTTP_400_BAD_REQUEST)
approval_request = revision.request_approval(creator=request.user)
return Response(ApprovalRequestSerializer(approval_request).data,
status=status.HTTP_201_CREATED)
class ApprovalRequestViewSet(viewsets.ReadOnlyModelViewSet):
queryset = ApprovalRequest.objects.all()
serializer_class = ApprovalRequestSerializer
permission_classes = [
AdminEnabledOrReadOnly,
permissions.DjangoModelPermissionsOrAnonReadOnly,
]
@detail_route(methods=['POST'])
def approve(self, request, pk=None):
approval_request = self.get_object()
if 'comment' not in request.data:
return Response({'comment': 'This field is required.'},
status=status.HTTP_400_BAD_REQUEST)
try:
approval_request.approve(approver=request.user, comment=request.data.get('comment'))
except ApprovalRequest.NotActionable:
return Response(
{'error': 'This approval request has already been approved or rejected.'},
status=status.HTTP_400_BAD_REQUEST)
except ApprovalRequest.CannotActOnOwnRequest:
return Response(
{'error': 'You cannot approve your own approval request.'},
status=status.HTTP_403_FORBIDDEN)
return Response(ApprovalRequestSerializer(approval_request).data)
@detail_route(methods=['POST'])
def reject(self, request, pk=None):
approval_request = self.get_object()
if 'comment' not in request.data:
return Response({'comment': 'This field is required.'},
status=status.HTTP_400_BAD_REQUEST)
try:
approval_request.reject(approver=request.user, comment=request.data.get('comment'))
except ApprovalRequest.NotActionable:
return Response(
{'error': 'This approval request has already been approved or rejected.'},
status=status.HTTP_400_BAD_REQUEST)
except ApprovalRequest.CannotActOnOwnRequest:
return Response(
{'error': 'You cannot reject your own approval request.'},
status=status.HTTP_403_FORBIDDEN)
return Response(ApprovalRequestSerializer(approval_request).data)
@detail_route(methods=['POST'])
def close(self, request, pk=None):
approval_request = self.get_object()
approval_request.close()
return Response(status=status.HTTP_204_NO_CONTENT)

Просмотреть файл

@ -0,0 +1,805 @@
import hashlib
from unittest.mock import patch
from django.db import connection
from django.test.utils import CaptureQueriesContext
import pytest
from rest_framework.reverse import reverse
from normandy.base.api.permissions import AdminEnabledOrReadOnly
from normandy.base.tests import UserFactory, Whatever
from normandy.base.utils import aware_datetime, canonical_json_dumps
from normandy.recipes.models import ApprovalRequest, Recipe
from normandy.recipes.tests import (
ActionFactory,
ApprovalRequestFactory,
ChannelFactory,
CountryFactory,
LocaleFactory,
RecipeFactory,
fake_sign,
)
@pytest.mark.django_db
class TestActionAPI(object):
def test_it_works(self, api_client):
res = api_client.get('/api/v2/action/')
assert res.status_code == 200
assert res.data == []
def test_it_serves_actions(self, api_client):
action = ActionFactory(
name='foo',
implementation='foobar',
arguments_schema={'type': 'object'}
)
res = api_client.get('/api/v2/action/')
action_url = reverse('recipes:action-implementation', kwargs={
'name': action.name,
'impl_hash': action.implementation_hash,
})
assert res.status_code == 200
assert res.data == [
{
'id': action.id,
'name': 'foo',
'implementation_url': Whatever.endswith(action_url),
'arguments_schema': {'type': 'object'}
}
]
def test_list_view_includes_cache_headers(self, api_client):
res = api_client.get('/api/v2/action/')
assert res.status_code == 200
# It isn't important to assert a particular value for max-age
assert 'max-age=' in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_detail_view_includes_cache_headers(self, api_client):
action = ActionFactory()
res = api_client.get('/api/v2/action/{id}/'.format(id=action.id))
assert res.status_code == 200
# It isn't important to assert a particular value for max-age
assert 'max-age=' in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_list_sets_no_cookies(self, api_client):
res = api_client.get('/api/v2/action/')
assert res.status_code == 200
assert 'Cookies' not in res
def test_detail_sets_no_cookies(self, api_client):
action = ActionFactory()
res = api_client.get('/api/v2/action/{id}/'.format(id=action.id))
assert res.status_code == 200
assert res.client.cookies == {}
@pytest.mark.django_db
class TestImplementationAPI(object):
def test_it_serves_implementations(self, api_client):
action = ActionFactory()
res = api_client.get('/api/v2/action/{id}/implementation/{hash}/'.format(
id=action.id,
hash=action.implementation_hash,
))
assert res.status_code == 200
assert res.content.decode() == action.implementation
assert res['Content-Type'] == 'application/javascript; charset=utf-8'
def test_it_404s_if_hash_doesnt_match(self, api_client):
action = ActionFactory(implementation='asdf')
bad_hash = hashlib.sha1('nomatch'.encode()).hexdigest()
res = api_client.get('/api/v2/action/{id}/implementation/{hash}/'.format(
id=action.id,
hash=bad_hash,
))
assert res.status_code == 404
assert res.content.decode() == '/* Hash does not match current stored action. */'
assert res['Content-Type'] == 'application/javascript; charset=utf-8'
def test_it_includes_cache_headers(self, api_client, settings):
# Note: Can't override the cache time setting, since it is read
# when invoking the decorator at import time. Changing it would
# require mocking, and that isn't worth it.
action = ActionFactory()
res = api_client.get('/api/v2/action/{id}/implementation/{hash}/'.format(
id=action.id,
hash=action.implementation_hash,
))
assert res.status_code == 200
max_age = 'max-age={}'.format(settings.ACTION_IMPLEMENTATION_CACHE_TIME)
assert max_age in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_sets_no_cookies(self, api_client):
action = ActionFactory()
res = api_client.get('/api/v2/action/{id}/implementation/{hash}/'.format(
id=action.id,
hash=action.implementation_hash,
))
assert res.status_code == 200
assert res.client.cookies == {}
@pytest.mark.django_db
class TestRecipeAPI(object):
def test_it_works(self, api_client):
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
assert res.data == []
def test_it_serves_recipes(self, api_client):
recipe = RecipeFactory()
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
assert res.data[0]['name'] == recipe.name
def test_it_can_create_recipes(self, api_client):
action = ActionFactory()
# Enabled recipe
res = api_client.post('/api/v2/recipe/', {
'name': 'Test Recipe',
'action_id': action.id,
'arguments': {},
'extra_filter_expression': 'whatever',
'enabled': True
})
assert res.status_code == 201
recipes = Recipe.objects.all()
assert recipes.count() == 1
def test_it_can_create_disabled_recipes(self, api_client):
action = ActionFactory()
# Disabled recipe
res = api_client.post('/api/v2/recipe/', {
'name': 'Test Recipe',
'action_id': action.id,
'arguments': {},
'extra_filter_expression': 'whatever',
'enabled': False
})
assert res.status_code == 201
recipes = Recipe.objects.all()
assert recipes.count() == 1
def test_it_can_edit_recipes(self, api_client):
recipe = RecipeFactory(name='unchanged', extra_filter_expression='true')
old_revision_id = recipe.revision_id
res = api_client.patch('/api/v2/recipe/%s/' % recipe.id, {
'name': 'changed',
'extra_filter_expression': 'false',
})
assert res.status_code == 200
recipe = Recipe.objects.all()[0]
assert recipe.name == 'changed'
assert recipe.filter_expression == 'false'
assert recipe.revision_id != old_revision_id
def test_creation_when_action_does_not_exist(self, api_client):
res = api_client.post('/api/v2/recipe/', {'name': 'Test Recipe',
'action_id': 1234,
'arguments': '{}'})
assert res.status_code == 400
recipes = Recipe.objects.all()
assert recipes.count() == 0
def test_creation_when_arguments_are_invalid(self, api_client):
action = ActionFactory(
name='foobarbaz',
arguments_schema={
'type': 'object',
'properties': {'message': {'type': 'string'}},
'required': ['message']
}
)
res = api_client.post('/api/v2/recipe/', {'name': 'Test Recipe',
'enabled': True,
'extra_filter_expression': 'true',
'action_id': action.id,
'arguments': {'message': ''}})
assert res.status_code == 400
recipes = Recipe.objects.all()
assert recipes.count() == 0
def test_it_can_change_action_for_recipes(self, api_client):
recipe = RecipeFactory()
action = ActionFactory()
res = api_client.patch('/api/v2/recipe/%s/' % recipe.id, {'action_id': action.id})
assert res.status_code == 200
recipe = Recipe.objects.get(pk=recipe.id)
assert recipe.action == action
def test_it_can_change_arguments_for_recipes(self, api_client):
recipe = RecipeFactory(arguments_json='{}')
action = ActionFactory(
name='foobarbaz',
arguments_schema={
'type': 'object',
'properties': {'message': {'type': 'string'}},
'required': ['message']
}
)
arguments = {'message': 'test message'}
res = api_client.patch('/api/v2/recipe/%s/' % recipe.id, {
'action_id': action.id, 'arguments': arguments})
assert res.status_code == 200
recipe = Recipe.objects.get(pk=recipe.id)
assert recipe.arguments == arguments
def test_it_can_delete_recipes(self, api_client):
recipe = RecipeFactory()
res = api_client.delete('/api/v2/recipe/%s/' % recipe.id)
assert res.status_code == 204
recipes = Recipe.objects.all()
assert recipes.count() == 0
def test_available_if_admin_enabled(self, api_client, settings):
settings.ADMIN_ENABLED = True
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
assert res.data == []
def test_readonly_if_admin_disabled(self, api_client, settings):
settings.ADMIN_ENABLED = False
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
recipe = RecipeFactory(name='unchanged')
res = api_client.patch('/api/v2/recipe/%s/' % recipe.id, {'name': 'changed'})
assert res.status_code == 403
assert res.data['detail'] == AdminEnabledOrReadOnly.message
def test_history(self, api_client):
recipe = RecipeFactory(name='version 1')
recipe.revise(name='version 2')
recipe.revise(name='version 3')
res = api_client.get('/api/v2/recipe/%s/history/' % recipe.id)
assert res.data[0]['recipe']['name'] == 'version 3'
assert res.data[1]['recipe']['name'] == 'version 2'
assert res.data[2]['recipe']['name'] == 'version 1'
def test_it_can_enable_recipes(self, api_client):
recipe = RecipeFactory(enabled=False, approver=UserFactory())
res = api_client.post('/api/v2/recipe/%s/enable/' % recipe.id)
assert res.status_code == 200
assert res.data['enabled'] is True
recipe = Recipe.objects.all()[0]
assert recipe.enabled
def test_cannot_enable_unapproved_recipes(self, api_client):
recipe = RecipeFactory(enabled=False)
res = api_client.post('/api/v2/recipe/%s/enable/' % recipe.id)
assert res.status_code == 409
assert res.data['enabled'] == 'Cannot enable a recipe that is not approved.'
def test_it_can_disable_recipes(self, api_client):
recipe = RecipeFactory(approver=UserFactory(), enabled=True)
res = api_client.post('/api/v2/recipe/%s/disable/' % recipe.id)
assert res.status_code == 200
assert res.data['enabled'] is False
recipe = Recipe.objects.all()[0]
assert not recipe.is_approved
assert not recipe.enabled
def test_filtering_by_enabled_lowercase(self, api_client):
r1 = RecipeFactory(approver=UserFactory(), enabled=True)
RecipeFactory(enabled=False)
res = api_client.get('/api/v2/recipe/?enabled=true')
assert res.status_code == 200
assert [r['id'] for r in res.data] == [r1.id]
def test_filtering_by_enabled_fuzz(self, api_client):
"""
Test that we don't return 500 responses when we get unexpected boolean filters.
This was a real case that showed up in our error logging.
"""
url = "/api/v2/recipe/?enabled=javascript%3a%2f*<%2fscript><svg%2fonload%3d'%2b%2f'%2f%2b"
res = api_client.get(url)
assert res.status_code == 400
assert res.data == {
'messages': [
"'javascript:/*</script><svg/onload='+/'/+' value must be either True or False.",
],
}
def test_list_view_includes_cache_headers(self, api_client):
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
# It isn't important to assert a particular value for max_age
assert 'max-age=' in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_signed_view_includes_cache_headers(self, api_client):
res = api_client.get('/api/v2/recipe/signed/')
assert res.status_code == 200
# It isn't important to assert a particular value for max-age
assert 'max-age=' in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_detail_view_includes_cache_headers(self, api_client):
recipe = RecipeFactory()
res = api_client.get(f'/api/v2/recipe/{recipe.id}/')
assert res.status_code == 200
# It isn't important to assert a particular value for max-age
assert 'max-age=' in res['Cache-Control']
assert 'public' in res['Cache-Control']
def test_signed_listing_works(self, api_client):
r1 = RecipeFactory(signed=True)
res = api_client.get('/api/v2/recipe/signed/')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['recipe']['id'] == r1.id
assert res.data[0]['signature']['signature'] == r1.signature.signature
def test_signed_only_lists_signed_recipes(self, api_client):
r1 = RecipeFactory(signed=True)
r2 = RecipeFactory(signed=True)
RecipeFactory(signed=False)
res = api_client.get('/api/v2/recipe/signed/')
assert res.status_code == 200
assert len(res.data) == 2
res.data.sort(key=lambda r: r['recipe']['id'])
assert res.data[0]['recipe']['id'] == r1.id
assert res.data[0]['signature']['signature'] == r1.signature.signature
assert res.data[1]['recipe']['id'] == r2.id
assert res.data[1]['signature']['signature'] == r2.signature.signature
def test_signed_listing_filters_by_enabled(Self, api_client):
enabled_recipe = RecipeFactory(signed=True, approver=UserFactory(), enabled=True)
disabled_recipe = RecipeFactory(signed=True, enabled=False)
res = api_client.get('/api/v2/recipe/signed/?enabled=1')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['recipe']['id'] == enabled_recipe.id
res = api_client.get('/api/v2/recipe/signed/?enabled=0')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['recipe']['id'] == disabled_recipe.id
def test_list_sets_no_cookies(self, api_client):
res = api_client.get('/api/v2/recipe/')
assert res.status_code == 200
assert 'Cookies' not in res
def test_detail_sets_no_cookies(self, api_client):
recipe = RecipeFactory()
res = api_client.get('/api/v2/recipe/{id}/'.format(id=recipe.id))
assert res.status_code == 200
assert res.client.cookies == {}
def test_list_filter_status(self, api_client):
r1 = RecipeFactory(enabled=False)
r2 = RecipeFactory(approver=UserFactory(), enabled=True)
res = api_client.get('/api/v2/recipe/?status=enabled')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r2.id
res = api_client.get('/api/v2/recipe/?status=disabled')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r1.id
def test_list_filter_channels(self, api_client):
r1 = RecipeFactory(channels=[ChannelFactory(slug='beta')])
r2 = RecipeFactory(channels=[ChannelFactory(slug='release')])
res = api_client.get('/api/v2/recipe/?channels=beta')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r1.id
res = api_client.get('/api/v2/recipe/?channels=beta,release')
assert res.status_code == 200
assert len(res.data) == 2
for recipe in res.data:
assert recipe['id'] in [r1.id, r2.id]
def test_list_filter_countries(self, api_client):
r1 = RecipeFactory(countries=[CountryFactory(code='US')])
r2 = RecipeFactory(countries=[CountryFactory(code='CA')])
res = api_client.get('/api/v2/recipe/?countries=US')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r1.id
res = api_client.get('/api/v2/recipe/?countries=US,CA')
assert res.status_code == 200
assert len(res.data) == 2
for recipe in res.data:
assert recipe['id'] in [r1.id, r2.id]
def test_list_filter_locales(self, api_client):
r1 = RecipeFactory(locales=[LocaleFactory(code='en-US')])
r2 = RecipeFactory(locales=[LocaleFactory(code='fr-CA')])
res = api_client.get('/api/v2/recipe/?locales=en-US')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r1.id
res = api_client.get('/api/v2/recipe/?locales=en-US,fr-CA')
assert res.status_code == 200
assert len(res.data) == 2
for recipe in res.data:
assert recipe['id'] in [r1.id, r2.id]
def test_list_filter_text(self, api_client):
r1 = RecipeFactory(name='first', extra_filter_expression='1 + 1 == 2')
r2 = RecipeFactory(name='second', extra_filter_expression='one + one == two')
res = api_client.get('/api/v2/recipe/?text=first')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r1.id
res = api_client.get('/api/v2/recipe/?text=one')
assert res.status_code == 200
assert len(res.data) == 1
assert res.data[0]['id'] == r2.id
res = api_client.get('/api/v2/recipe/?text=t')
assert res.status_code == 200
assert len(res.data) == 2
for recipe in res.data:
assert recipe['id'] in [r1.id, r2.id]
def test_update_recipe_action(self, api_client):
r = RecipeFactory()
a = ActionFactory(name='test')
res = api_client.patch(f'/api/v2/recipe/{r.pk}/', {'action_id': a.id})
assert res.status_code == 200
r.refresh_from_db()
assert r.action == a
def test_update_recipe_locale(self, api_client):
l1 = LocaleFactory(code='fr-FR')
l2 = LocaleFactory(code='en-US')
r = RecipeFactory(locales=[l1])
res = api_client.patch(f'/api/v2/recipe/{r.pk}/', {'locales': ['en-US']})
assert res.status_code == 200
r.refresh_from_db()
assert list(r.locales.all()) == [l2]
def test_update_recipe_country(self, api_client):
c1 = CountryFactory(code='US')
c2 = CountryFactory(code='CA')
r = RecipeFactory(countries=[c1])
res = api_client.patch(f'/api/v2/recipe/{r.pk}/', {'countries': ['CA']})
assert res.status_code == 200
r.refresh_from_db()
assert list(r.countries.all()) == [c2]
def test_update_recipe_channel(self, api_client):
c1 = ChannelFactory(slug='release')
c2 = ChannelFactory(slug='beta')
r = RecipeFactory(channels=[c1])
res = api_client.patch(f'/api/v2/recipe/{r.pk}/', {'channels': ['beta']})
assert res.status_code == 200
r.refresh_from_db()
assert list(r.channels.all()) == [c2]
@pytest.mark.django_db
class TestRecipeRevisionAPI(object):
def test_it_works(self, api_client):
res = api_client.get('/api/v2/recipe_revision/')
assert res.status_code == 200
assert res.data == []
def test_it_serves_revisions(self, api_client):
recipe = RecipeFactory()
res = api_client.get('/api/v2/recipe_revision/%s/' % recipe.latest_revision.id)
assert res.status_code == 200
assert res.data['id'] == recipe.latest_revision.id
def test_request_approval(self, api_client):
recipe = RecipeFactory()
res = api_client.post(
'/api/v2/recipe_revision/{}/request_approval/'.format(recipe.latest_revision.id))
assert res.status_code == 201
assert res.data['id'] == recipe.latest_revision.approval_request.id
def test_cannot_open_second_approval_request(self, api_client):
recipe = RecipeFactory()
ApprovalRequestFactory(revision=recipe.latest_revision)
res = api_client.post(
'/api/v2/recipe_revision/{}/request_approval/'.format(recipe.latest_revision.id))
assert res.status_code == 400
@pytest.mark.django_db
class TestApprovalRequestAPI(object):
def test_it_works(self, api_client):
res = api_client.get('/api/v2/approval_request/')
assert res.status_code == 200
assert res.data == []
def test_approve(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(a.id),
{'comment': 'r+'})
assert res.status_code == 200
r.refresh_from_db()
assert r.is_approved
assert r.approved_revision.approval_request.comment == 'r+'
def test_approve_no_comment(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(a.id))
assert res.status_code == 400
assert res.data['comment'] == 'This field is required.'
def test_approve_not_actionable(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
a.approve(UserFactory(), 'r+')
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(a.id),
{'comment': 'r+'})
assert res.status_code == 400
assert res.data['error'] == 'This approval request has already been approved or rejected.'
def test_reject(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
res = api_client.post('/api/v2/approval_request/{}/reject/'.format(a.id),
{'comment': 'r-'})
assert res.status_code == 200
r.latest_revision.approval_request.refresh_from_db()
assert r.latest_revision.approval_status == r.latest_revision.REJECTED
assert r.latest_revision.approval_request.comment == 'r-'
def test_reject_no_comment(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
res = api_client.post('/api/v2/approval_request/{}/reject/'.format(a.id))
assert res.status_code == 400
assert res.data['comment'] == 'This field is required.'
def test_reject_not_actionable(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
a.approve(UserFactory(), 'r+')
res = api_client.post('/api/v2/approval_request/{}/reject/'.format(a.id),
{'comment': '-r'})
assert res.status_code == 400
assert res.data['error'] == 'This approval request has already been approved or rejected.'
def test_close(self, api_client):
r = RecipeFactory()
a = ApprovalRequestFactory(revision=r.latest_revision)
res = api_client.post('/api/v2/approval_request/{}/close/'.format(a.id))
assert res.status_code == 204
with pytest.raises(ApprovalRequest.DoesNotExist):
ApprovalRequest.objects.get(pk=a.pk)
@pytest.mark.django_db
class TestApprovalFlow(object):
def verify_signatures(self, api_client, expected_count=None):
res = api_client.get('/api/v2/recipe/signed/')
assert res.status_code == 200
signed_data = res.json()
if expected_count is not None:
assert len(signed_data) == expected_count
for recipe_and_signature in signed_data:
recipe = recipe_and_signature['recipe']
expected_signature = recipe_and_signature['signature']['signature']
data = canonical_json_dumps(recipe).encode()
actual_signature = fake_sign([data])[0]['signature']
assert actual_signature == expected_signature
def test_full_approval_flow(self, api_client):
action = ActionFactory()
user1 = UserFactory(is_superuser=True)
user2 = UserFactory(is_superuser=True)
api_client.force_authenticate(user1)
# Create a recipe
res = api_client.post('/api/v2/recipe/', {
'action_id': action.id,
'arguments': {},
'name': 'test recipe',
'extra_filter_expression': 'counter == 0',
'enabled': 'false',
})
assert res.status_code == 201
recipe_data_0 = res.json()
# Request approval for it
res = api_client.post('/api/v2/recipe_revision/{}/request_approval/'
.format(recipe_data_0['latest_revision']['id']))
approval_data = res.json()
assert res.status_code == 201
# The requester isn't allowed to approve a recipe
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(approval_data['id']),
{'comment': 'r+'})
assert res.status_code == 403 # Forbidden
# Approve the recipe
api_client.force_authenticate(user2)
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(approval_data['id']),
{'comment': 'r+'})
assert res.status_code == 200
# It is now visible in the API
res = api_client.get('/api/v2/recipe/{}/'.format(recipe_data_0['id']))
assert res.status_code == 200
recipe_data_1 = res.json()
self.verify_signatures(api_client, expected_count=1)
# Make another change
api_client.force_authenticate(user1)
res = api_client.patch('/api/v2/recipe/{}/'.format(recipe_data_1['id']), {
'extra_filter_expression': 'counter == 1',
})
assert res.status_code == 200
# The change should not be visible yet, since it isn't approved
res = api_client.get('/api/v2/recipe/{}/'.format(recipe_data_1['id']))
assert res.status_code == 200
recipe_data_2 = res.json()
assert recipe_data_2['extra_filter_expression'] == 'counter == 0'
self.verify_signatures(api_client, expected_count=1)
# Request approval for the change
res = api_client.post('/api/v2/recipe_revision/{}/request_approval/'
.format(recipe_data_2['latest_revision']['id']))
approval_data = res.json()
recipe_data_2['approval_request'] = approval_data
assert res.status_code == 201
# The change should not be visible yet, since it isn't approved
res = api_client.get('/api/v2/recipe/{}/'.format(recipe_data_1['id']))
assert res.status_code == 200
assert res.json() == recipe_data_2
self.verify_signatures(api_client, expected_count=1)
# Reject the change
api_client.force_authenticate(user2)
res = api_client.post('/api/v2/approval_request/{}/reject/'.format(approval_data['id']),
{'comment': 'r-'})
approval_data = res.json()
recipe_data_2['approval_request'] = approval_data
assert res.status_code == 200
# The change should not be visible yet, since it isn't approved
res = api_client.get('/api/v2/recipe/{}/'.format(recipe_data_1['id']))
assert res.status_code == 200
assert res.json() == recipe_data_2
self.verify_signatures(api_client, expected_count=1)
# Make a third version of the recipe
api_client.force_authenticate(user1)
res = api_client.patch('/api/v2/recipe/{}/'.format(recipe_data_1['id']), {
'extra_filter_expression': 'counter == 2',
})
recipe_data_3 = res.json()
assert res.status_code == 200
# Request approval
res = api_client.post('/api/v2/recipe_revision/{}/request_approval/'
.format(recipe_data_3['latest_revision']['id']))
approval_data = res.json()
assert res.status_code == 201
# Approve the change
api_client.force_authenticate(user2)
res = api_client.post('/api/v2/approval_request/{}/approve/'.format(approval_data['id']),
{'comment': 'r+'})
assert res.status_code == 200
# The change should be visible now, since it is approved
res = api_client.get('/api/v2/recipe/{}/'.format(recipe_data_1['id']))
assert res.status_code == 200
recipe_data_4 = res.json()
assert recipe_data_4['extra_filter_expression'] == 'counter == 2'
self.verify_signatures(api_client, expected_count=1)
def test_cancel_approval(self, api_client, mocked_autograph):
action = ActionFactory()
user1 = UserFactory(is_superuser=True)
user2 = UserFactory(is_superuser=True)
api_client.force_authenticate(user1)
# Create a recipe
res = api_client.post('/api/v2/recipe/', {
'action_id': action.id,
'arguments': {},
'name': 'test recipe',
'extra_filter_expression': 'counter == 0',
'enabled': 'false',
})
assert res.status_code == 201
recipe_id = res.json()['id']
revision_id = res.json()['latest_revision']['id']
# Request approval
res = api_client.post(f'/api/v2/recipe_revision/{revision_id}/request_approval/')
assert res.status_code == 201
approval_request_id = res.json()['id']
# Approve the recipe
api_client.force_authenticate(user2)
res = api_client.post(
f'/api/v2/approval_request/{approval_request_id}/approve/',
{'comment': 'r+'}
)
assert res.status_code == 200
# Make another change
api_client.force_authenticate(user1)
res = api_client.patch(
f'/api/v2/recipe/{recipe_id}/',
{'extra_filter_expression': 'counter == 1'}
)
assert res.status_code == 200
revision_id = res.json()['latest_revision']['id']
# Request approval for the second change
res = api_client.post(f'/api/v2/recipe_revision/{revision_id}/request_approval/')
approval_request_id = res.json()['id']
assert res.status_code == 201
# Cancel the approval request
res = api_client.post(f'/api/v2/approval_request/{approval_request_id}/close/')
assert res.status_code == 204
# The API should still have correct signatures
self.verify_signatures(api_client, expected_count=1)

Просмотреть файл

@ -1,34 +1,40 @@
from django.conf.urls import url, include
from normandy.base.api.routers import MixedViewRouter
from normandy.recipes.api.v1.views import (
ActionImplementationView,
ActionViewSet,
ApprovalRequestViewSet,
ClassifyClient,
Filters,
RecipeViewSet,
RecipeRevisionViewSet,
)
from normandy.recipes.api.v1 import views as api_v1_views
from normandy.recipes.api.v2 import views as api_v2_views
# API Router
router = MixedViewRouter()
router.register('action', ActionViewSet)
router.register('recipe', RecipeViewSet)
router.register('recipe_revision', RecipeRevisionViewSet)
router.register(r'approval_request', ApprovalRequestViewSet)
v1_router = MixedViewRouter()
v1_router.register('action', api_v1_views.ActionViewSet)
v1_router.register('recipe', api_v1_views.RecipeViewSet)
v1_router.register('recipe_revision', api_v1_views.RecipeRevisionViewSet)
v1_router.register(r'approval_request', api_v1_views.ApprovalRequestViewSet)
router.register_view('classify_client', ClassifyClient, name='classify-client', allow_cdn=False)
router.register_view('filters', Filters, name='filters')
v1_router.register_view('classify_client', api_v1_views.ClassifyClient, name='classify-client',
allow_cdn=False)
v1_router.register_view('filters', api_v1_views.Filters, name='filters')
v2_router = MixedViewRouter()
v2_router.register('action', api_v2_views.ActionViewSet)
v2_router.register('recipe', api_v2_views.RecipeViewSet)
v2_router.register('recipe_revision', api_v2_views.RecipeRevisionViewSet)
v2_router.register(r'approval_request', api_v2_views.ApprovalRequestViewSet)
app_name = 'recipes'
urlpatterns = [
url(r'^api/v1/', include(router.urls)),
url(r'^api/v2/', include(v2_router.urls)),
url(
r'^api/v2/action/(?P<id>[_\-\w]+)/implementation/(?P<impl_hash>[0-9a-f]{40})/$',
api_v2_views.ActionImplementationView.as_view(),
name='action-implementation'
),
url(r'^api/v1/', include(v1_router.urls)),
url(
r'^api/v1/action/(?P<name>[_\-\w]+)/implementation/(?P<impl_hash>[0-9a-f]{40})/$',
ActionImplementationView.as_view(),
api_v1_views.ActionImplementationView.as_view(),
name='action-implementation'
),
]