88 lines
4.2 KiB
Python
88 lines
4.2 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
SPEC = importlib.util.spec_from_file_location('finance_import', ROOT / 'deploy/run-finance-import.py')
|
|
RUNNER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(RUNNER)
|
|
|
|
|
|
class FinanceImportTest(unittest.TestCase):
|
|
def run_import(self, arguments, callback=None, memory=8*1024*1024):
|
|
with mock.patch.object(RUNNER, 'available_memory_kib', return_value=memory), \
|
|
mock.patch.object(RUNNER.sys, 'argv', ['run-finance-import.py', *arguments]), \
|
|
mock.patch.object(RUNNER.subprocess, 'check_output', return_value='caiwu_akaunting:123456789abc\n'), \
|
|
mock.patch.object(RUNNER.subprocess, 'run') as run:
|
|
run.return_value = subprocess.CompletedProcess([], 0)
|
|
if callback:
|
|
run.side_effect = callback
|
|
result = RUNNER.main()
|
|
return result, run.call_args_list
|
|
|
|
def test_input_is_read_only_original_is_preserved_and_arguments_are_not_shell_code(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
original = pathlib.Path(folder) / 'file with spaces.xlsx'
|
|
original.write_bytes(b'original workbook')
|
|
staged = []
|
|
|
|
def execute(command, **kwargs):
|
|
if command[1] == 'rm':
|
|
return subprocess.CompletedProcess(command, 0)
|
|
self.assertIn('--no-deps', command)
|
|
self.assertIn('memory_limit=512M', command)
|
|
self.assertEqual('1g', kwargs['env']['FINANCE_BATCH_MEMORY_LIMIT'])
|
|
self.assertEqual('0', kwargs['env']['RUN_MIGRATIONS'])
|
|
self.assertIn('--source-name=$(touch forbidden)', command)
|
|
mount = command[command.index('--volume') + 1]
|
|
self.assertTrue(mount.endswith(':/run/caiwu-import:ro'))
|
|
target = pathlib.Path(mount.rsplit(':', 2)[0]) / '0' / original.name
|
|
self.assertEqual(original.read_bytes(), target.read_bytes())
|
|
self.assertEqual(0, target.stat().st_mode & 0o222)
|
|
self.assertEqual('/run/caiwu-import/0/' + original.name, command[-2])
|
|
staged.append(target)
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
result, calls = self.run_import(['business:import-alibaba-account-balances', str(original), '--source-name=$(touch forbidden)'], execute)
|
|
self.assertEqual(0, result)
|
|
self.assertEqual(b'original workbook', original.read_bytes())
|
|
self.assertFalse(staged[0].exists())
|
|
self.assertEqual('rm', calls[-1].args[0][1])
|
|
|
|
def test_named_file_option_is_mapped_without_changing_its_name(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / 'named.xlsx'
|
|
source.write_bytes(b'workbook')
|
|
result, calls = self.run_import(['jst-order-labels:import', '--file=' + str(source)])
|
|
self.assertEqual(0, result)
|
|
self.assertEqual('--file=/run/caiwu-import/0/' + source.name, calls[0].args[0][-1])
|
|
|
|
def test_low_memory_does_not_start_a_container(self):
|
|
result, calls = self.run_import(['order-data:import-after-sales-cost', '/input.xlsx'], memory=1024*1024)
|
|
self.assertEqual(75, result)
|
|
self.assertEqual([], calls)
|
|
|
|
def test_unknown_import_is_rejected(self):
|
|
result, calls = self.run_import(['unknown:import', '/input.xlsx'])
|
|
self.assertEqual(64, result)
|
|
self.assertEqual([], calls)
|
|
|
|
def test_timeout_removes_the_actual_job_container(self):
|
|
def execute(command, **kwargs):
|
|
if command[1] == 'compose':
|
|
raise subprocess.TimeoutExpired(command, kwargs['timeout'])
|
|
return subprocess.CompletedProcess(command, 0)
|
|
result, calls = self.run_import(['order-data:import-after-sales-cost', '/input.xlsx'], execute)
|
|
self.assertEqual(124, result)
|
|
started = calls[0].args[0]
|
|
self.assertEqual(['docker', 'rm', '--force', started[started.index('--name') + 1]], calls[-1].args[0])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|