64 lines
2.8 KiB
Python
64 lines
2.8 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from pathlib import Path
|
|
|
|
# A separate temporary service root and disabled worker prevent any GPU/service action.
|
|
_root = tempfile.TemporaryDirectory()
|
|
os.environ['AUDIOBOOK_SERVICE_ROOT'] = _root.name
|
|
os.environ['AUDIOBOOK_DISABLE_WORKER'] = '1'
|
|
import audiobook_service as service
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
class ApiOwnershipTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.health = patch('voicestudio_client.check_available')
|
|
self.health.start()
|
|
self.addCleanup(self.health.stop)
|
|
self.client = TestClient(service.app)
|
|
self.one = self.enroll('first')
|
|
self.two = self.enroll('second')
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def enroll(self, name):
|
|
response = self.client.post('/v1/devices/enroll', json={'code': service.devices.create_code(), 'label': name})
|
|
self.assertEqual(200, response.status_code)
|
|
return {'Authorization': 'Bearer ' + response.json()['token']}
|
|
|
|
def test_owner_is_required_for_status_download_and_delete(self):
|
|
response = self.client.post('/v1/audiobooks', headers=self.one,
|
|
data={'title': 'Fixture', 'author': 'Test'}, files={'book': ('fixture.fb2', b'<FictionBook/>', 'application/xml')})
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
job_id = response.json()['id']
|
|
path = '/v1/audiobooks/' + job_id
|
|
self.assertEqual(401, self.client.get(path).status_code)
|
|
self.assertEqual(200, self.client.get(path, headers=self.one).status_code)
|
|
self.assertEqual(404, self.client.get(path, headers=self.two).status_code)
|
|
self.assertEqual(404, self.client.get(path + '/file', headers=self.two).status_code)
|
|
self.assertEqual(404, self.client.delete(path, headers=self.two).status_code)
|
|
self.assertEqual(200, self.client.delete(path, headers=self.one).status_code)
|
|
self.assertEqual(404, self.client.get(path, headers=self.one).status_code)
|
|
|
|
def test_health_does_not_disclose_job_id_and_legacy_token_is_rejected(self):
|
|
self.assertNotIn('activeJobId', self.client.get('/health').json())
|
|
response = self.client.post('/v1/audiobooks', headers={'Authorization': 'Bearer legacy-shared-key'},
|
|
data={'title': 'Test'}, files={'book': ('a.fb2', b'<FictionBook/>')})
|
|
self.assertEqual(401, response.status_code)
|
|
|
|
def test_enrollment_code_cannot_be_replayed(self):
|
|
code = service.devices.create_code()
|
|
self.assertEqual(200, self.client.post('/v1/devices/enroll', json={'code': code}).status_code)
|
|
self.assertEqual(403, self.client.post('/v1/devices/enroll', json={'code': code}).status_code)
|
|
|
|
|
|
def tearDownModule():
|
|
_root.cleanup()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|