vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

refactor(archivedlogs): move archivelog handler to endpoints

Browse source
This commit is contained in:
EvB 2017-04-24 15:59:10 -04:00
parent 0d04fd8bd2
commit 5e995fae20
4 changed files with 54 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

29
test/test_endpoints.py
View file

@ -4,7 +4,10 @@ import json as py_json
import time
import unittest
import base64
import zlib
from mock import patch
from io import BytesIO
from urllib import urlencode
from urlparse import urlparse, urlunparse, parse_qs
from datetime import datetime, timedelta
@ -27,6 +30,7 @@ from endpoints.web import web as web_bp
from endpoints.webhooks import webhooks as webhooks_bp
from initdb import setup_database_for_testing, finished_database_for_testing
from test.helpers import assert_action_logged
from util.registry.gzipinputstream import WINDOW_BUFFER_SIZE
try:
app.register_blueprint(web_bp, url_prefix='')
@ -133,6 +137,31 @@ class EndpointTestCase(unittest.TestCase):
headers={"Content-Type": "application/json"})
self.assertEquals(rv.status_code, 200)
class BuildLogsTestCase(EndpointTestCase):
build_uuid = 'deadpork-dead-pork-dead-porkdeadpork'
def test_logarchive_invalid_build_uuid(self):
self.login('public', 'password')
self.getResponse('web.logarchive', file_id='bad_build_uuid', expected_code=403)
def test_logarchive_not_logged_in(self):
self.getResponse('web.logarchive', file_id=self.build_uuid, expected_code=401)
def test_logarchive_unauthorized(self):
self.login('reader', 'password')
self.getResponse('web.logarchive', file_id=self.build_uuid, expected_code=403)
def test_logarchive_file_not_found(self):
self.login('public', 'password')
self.getResponse('web.logarchive', file_id=self.build_uuid, expected_code=403)
def test_logarchive_successful(self):
self.login('public', 'password')
data = b"my_file_stream"
mock_file = BytesIO(zlib.compressobj(-1, zlib.DEFLATED, WINDOW_BUFFER_SIZE).compress(data))
with patch('endpoints.web.log_archive._storage.stream_read_file', return_value=mock_file):
self.getResponse('web.logarchive', file_id=self.build_uuid, expected_code=200)
class WebhookEndpointTestCase(EndpointTestCase):
def test_invalid_build_trigger_webhook(self):