"""Test module for Deezer flow utilities."""
from datetime import datetime
import os
from unittest import mock, TestCase
from unittest.mock import call, patch
import pytest
from feed_ingestion.flows.deezer_marketshare import config
import feed_ingestion.util.deezer_zephir_utils as zephir
_date = '2016-05-23'
class TestDeezerZephirUtils(TestCase):
"""Test class for deezer_zephir_utils."""
@pytest.fixture(autouse=True)
def set_files_dir(self):
"""Set files directory data."""
self.file_name = 'TheOrchard_201605_7221.zip'
self.local_file_path = f'/tests/util/{self.file_name}'
self.local_dir = '/tests/util'
self.source = 'source.csv'
@pytest.fixture(autouse=True)
def set_zephir_settings(self):
"""Set Zephir settings."""
self.zephir_settings = {
'username': 'test_deezer_user',
'password': 'TestPassword123',
'host': 'https://zephir.deezer.com',
'path': '/external/download?file=invoices/',
'search': '/external?path=invoices',
}
@pytest.fixture(autouse=True)
def set_search_invoice_response(self):
"""Set search invoice page response."""
self.search_invoice_response = \
'\n\n\n\n| \n' \
'TheOrchard_201605_adjustment_7221.zip\n' \
' | \n\n\n\n | \n' \
'TheOrchard_201703_6641.zip\n' \
' | \n\n\n\n | \n' \
'TheOrchard_201511_7081.zip\n' \
' | \n\n\n\n | \n' \
'TheOrchard_201605_4301.zip\n' \
' | \n\n\n\n | \n' \
'TheOrchard_201605_7221.zip\n' \
' | \n'
@patch('os.walk')
@patch('gzip.open')
@patch('shutil.copyfileobj')
@patch('builtins.open', read_data='data')
@patch('feed_ingestion.util.deezer_zephir_utils.ZipFile')
def test_repack_source_file_one_file(
self, mock_zipfile, mock_open, mock_copyfileobj,
mock_gzip_open, mock_walk):
"""Test repack_source_file for one file (str)."""
mock_walk.return_value = [
('/tests', ('util',), ('file_1.test', 'file_2.txt')),
('/tests/util', (), (self.source,)),
]
files = zephir.repack_source_file(
self.local_file_path, self.local_dir, self.source, True)
self.assertEqual(files, [{
'gzip_file': f'{self.local_dir}/{self.source}.gz',
'gzip_file_name': f'{self.source}.gz',
'source_file_name': self.source
}])
self.assertEqual(f'{self.local_dir}/{self.source}',
'/tests/util/source.csv')
@patch('os.walk')
@patch('gzip.open')
@patch('shutil.copyfileobj')
@patch('builtins.open', read_data='data')
@patch('feed_ingestion.util.deezer_zephir_utils.ZipFile')
def test_repack_source_file_multiple_files(
self, mock_zipfile, mock_open, mock_copyfileobj,
mock_gzip_open, mock_walk):
"""Test repack_source_file for source files dict."""
mock_walk.return_value = [
('/tests', ('util',), ('file_1.test', 'file_2.txt')),
('/tests/util', (), (self.source,)),
]
_files = ['file_2.txt', self.source]
files = zephir.repack_source_file(
self.local_file_path, self.local_dir, {'files': _files})
self.assertEqual(files, [{
'gzip_file':
f'{self.local_dir if source is self.source else "/tests"}'
f'/{source}.gz',
'gzip_file_name': f'{source}.gz',
'source_file_name': source} for source in _files])
self.assertEqual(f'{self.local_dir}/{self.source}',
'/tests/util/source.csv')
@patch('requests.session')
def test_log_in_to_zephir(self, mock_session):
"""Test Zephir log in."""
session = zephir.log_in_to_zephir(self.zephir_settings)
self.assertIn(call.get(
f"{self.zephir_settings.get('host')}/accounts/login/"),
session.method_calls)
self.assertIn(call.post(
f"{self.zephir_settings.get('host')}/accounts/login/",
data={
'username': self.zephir_settings.get('username'),
'password': self.zephir_settings.get('password'),
'csrfmiddlewaretoken': mock.ANY,
'next': '/',
}), session.method_calls)
self.assertIn(call.post().raise_for_status(), session.mock_calls)
@patch('requests.session')
def test_search_for_file_on_zephir(self, mock_session):
"""Test Zephir search page."""
mock_session.get().text = self.search_invoice_response
date_obj = datetime.strptime(_date, '%Y-%m-%d')
pattern = config.zip_pattern.format(date=date_obj)
files = zephir.search_for_file_on_zephir(
mock_session, self.zephir_settings, pattern)
self.assertIn(call.get(f"{self.zephir_settings.get('host')}"
f"{self.zephir_settings.get('search')}"),
mock_session.method_calls)
self.assertEqual(files, {'TheOrchard_201605_4301.zip',
'TheOrchard_201605_7221.zip',
'TheOrchard_201605_adjustment_7221.zip'})
@patch('os.path.getsize')
@patch('builtins.open')
@patch('requests.session')
@patch('feed_ingestion.util.deezer_zephir_utils.is_zipfile')
def test_download_file_from_zephir(
self, mock_is_zipfile, mock_session, mock_open, mock_getsize):
"""Test Zephir file successful downloading."""
mock_session.get().content = 'test_content'
mock_getsize.return_value = 10
local_path = zephir.download_file_from_zephir(
mock_session, self.zephir_settings, self.file_name, self.local_dir)
self.assertIn(call().__enter__().write(mock_session.get().content),
mock_open.mock_calls)
self.assertIn(call(f'{self.local_dir}/{self.file_name}', 'wb'),
mock_open.mock_calls)
self.assertIn(call.get().raise_for_status(), mock_session.mock_calls)
self.assertEqual(mock_getsize.call_count, 1)
self.assertEqual(mock_is_zipfile.call_count, 1)
self.assertEqual(local_path, f'{self.local_dir}/{self.file_name}')
@patch('os.path.getsize')
@patch('io.open')
@patch('requests.session')
def test_download_invalid_file_from_zephir(
self, mock_session, mock_io_open, mock_getsize):
"""Test writing invalid file (pattern, i.e. not .zip)."""
mock_session.get().content = b'test_content'
mock_io_open.tell.return_value = 0
mock_getsize.return_value = 10
non_zip_file_name = 'file.py'
exception = 'Zephir service have not returned a .zip file'
with self.assertRaises(FileNotFoundError) as context:
zephir.download_file_from_zephir(
mock_session, self.zephir_settings, non_zip_file_name, './')
self.assertIn(call.get(
f"{self.zephir_settings.get('host')}"
f"{self.zephir_settings.get('path')}{non_zip_file_name}"),
mock_session.method_calls)
self.assertIn(call.get().raise_for_status(), mock_session.mock_calls)
self.assertEqual(context.exception.args[0], exception)
os.remove(f'./{non_zip_file_name}')
@patch('feed_ingestion.util.deezer_zephir_utils.log_in_to_zephir')
@patch('feed_ingestion.util.deezer_zephir_utils.download_file_from_zephir')
def test_download_daily_report_from_zephir(
self, mock_download, mock_login):
"""Test downloading daily report from Zephir."""
zephir.download_daily_report_from_zephir(
self.zephir_settings, self.file_name, './')
self.assertEqual(mock_login.call_count, 1)
self.assertEqual(mock_download.call_count, 1)
self.assertIn(call(self.zephir_settings),
mock_login.mock_calls)
self.assertIn(call(
mock_login(), self.zephir_settings, self.file_name, './'),
mock_download.mock_calls)
@patch('requests.session')
@patch('feed_ingestion.util.deezer_zephir_utils.log_in_to_zephir')
@patch('feed_ingestion.util.deezer_zephir_utils.download_file_from_zephir')
def test_download_invoice_from_zephir(
self, mock_download, mock_login, mock_session):
"""Test downloading invoice from Zephir."""
# mock_search.return_value = self.search_invoice_response
mock_login().get().text = self.search_invoice_response # *
mock_download.return_value = 'file.csv.zip'
date_obj = datetime.strptime(_date, '%Y-%m-%d')
pattern = config.zip_pattern.format(date=date_obj)
files = zephir.download_invoice_from_zephir(
self.zephir_settings, pattern, './')
self.assertEqual(mock_login.call_count, 2)
# one call is for the *-marked line, the 2d is actual
self.assertIn(call(self.zephir_settings),
mock_login.mock_calls)
self.assertIn(call().get(f"{self.zephir_settings.get('host')}"
f"{self.zephir_settings.get('search')}"),
mock_login.mock_calls)
self.assertEqual(mock_download.call_count, 3)
# for all the matching three files
self.assertIn(call(
mock_login(), self.zephir_settings, self.file_name, './'),
mock_download.mock_calls) # matches three times
self.assertEqual(files, ['file.csv.zip'] * 3)