"""
Excel 2007 / JSON microservice Flask Route Handler Tests
==================
This file provides flask tests of the flask route handlers
"""
from io import BytesIO
from os import path
import json
import moto
from app import handlers
handlers.application.testing = True
client = handlers.application.test_client()
def mock_xlsx_file_from_s3_(s3, bucket_name, folder, filename, fp):
"""
Mock the library method to load the fixture file
"""
with open(path.join(
path.dirname(__file__),
'fixtures',
'Perf_Rights_Test.xlsx'), 'rb') as xlsx_fixture_file:
# populate the mock xlsx file with the
# binary data from the fixture file
fp.write(xlsx_fixture_file.read())
return fp
def mock_json_file_to_s3_(s3, bucket_name, folder, filename, fp):
"""
Mock the function to an empty body - no return is needed
"""
return
def get_s3_():
return moto.mock_s3
def test_index():
"""
Tests the index route
"""
response = client.get('/')
assert b'
XLSX/JSON Converter' in response.data
def test_xlsxtojson(monkeypatch):
"""
Tests the xlsxtojson route
"""
# monkeypatch the s3 library functions since moto @mock_s3
# doesn't seem to work here
monkeypatch.setattr('app.logic.converter.get_file_from_s3',
mock_xlsx_file_from_s3_)
monkeypatch.setattr('app.logic.converter.put_file_to_s3',
mock_json_file_to_s3_)
monkeypatch.setattr('app.logic.converter.get_s3', get_s3_)
def open_fp_(temp_dir, file_type, owner_type, owner_id, filename, mode):
"""
Mock to only return BytesIO file-like objects
"""
return BytesIO()
monkeypatch.setattr('app.logic.converter.open_fp', open_fp_)
def makedirs_(xlsx_fs_folder, exist_ok=True):
"""
Mock the function to an empty body - no return is needed
"""
return
monkeypatch.setattr('app.logic.converter.makedirs', makedirs_)
# Mock to only return the original BytesIO instead of a new one
def reopen_fp_(temp_dir,
file_type,
owner_type,
owner_id,
filename,
mode,
original_fp):
return original_fp
monkeypatch.setattr('app.logic.converter.reopen_fp', reopen_fp_)
response = client.post('/xlsxtojson',
data=dict(owner_type='vendor',
owner_id=1236,
filename='Perf_Rights_Test.xlsx')
)
assert response.status_code == 200
assert response.data == \
b'ows-xlsx/json/vendor_1236/Perf_Rights_Test.json'
def test_xlsxtojson_no_filename(monkeypatch):
"""
Tests the xlsxtojson route
"""
# monkeypatch the two s3 library functions since moto @mock_s3
# doesn't seem to work here
monkeypatch.setattr('app.logic.converter.get_file_from_s3',
mock_xlsx_file_from_s3_)
monkeypatch.setattr('app.logic.converter.put_file_to_s3',
mock_json_file_to_s3_)
# Mock open to only return BytesIO file-like objects
def open_fp_(temp_dir, file_type, owner_type, owner_id, filename):
return BytesIO()
monkeypatch.setattr('app.logic.converter.open_fp', open_fp_)
response = client.post('/xlsxtojson',
data=dict(owner_type='vendor',
owner_id=1236)
)
assert response.status_code == 400
def test_jsontoxlsx(monkeypatch):
"""
Tests the jsontoxlsx route
"""
# monkeypatch the s3 library functions since moto @mock_s3
# doesn't seem to work here
monkeypatch.setattr('app.logic.converter.get_file_from_s3',
mock_xlsx_file_from_s3_)
monkeypatch.setattr('app.logic.converter.put_file_to_s3',
mock_json_file_to_s3_)
monkeypatch.setattr('app.logic.converter.get_s3', get_s3_)
def open_fp_(temp_dir, file_type, owner_type, owner_id, filename, mode):
"""
Mock to only return BytesIO file-like objects
"""
return BytesIO()
monkeypatch.setattr('app.logic.converter.open_fp', open_fp_)
def makedirs_(xlsx_fs_folder, exist_ok=True):
"""
Mock the function to an empty body - no return is needed
"""
return
monkeypatch.setattr('app.logic.converter.makedirs', makedirs_)
# Mock to only return the original BytesIO instead of a new one
def reopen_fp_(temp_dir,
file_type,
owner_type,
owner_id,
filename,
mode,
original_fp):
return original_fp
monkeypatch.setattr('app.logic.converter.reopen_fp', reopen_fp_)
monkeypatch.setattr('app.logic.async.celery.conf.BROKER_URL',
'sqla+sqlite://')
monkeypatch.setattr('app.logic.async.celery.conf.BROKER_TRANSPORT_OPTIONS',
'')
monkeypatch.setattr('app.logic.async.celery.backend.dburi', 'sqlite://')
response = client.post('/jsontoxlsx',
data=dict(owner_type='vendor',
owner_id=1236,
filename='Perf_Rights_Test.json',
template='Perf_Rights_Template.xlsx')
)
assert response.status_code == 200
json_response = json.loads(response.get_data(True))
assert json_response.get('job_status') == 'PENDING'