""" 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'