Reisacher-Forderung/tests/test_uploads.py

import io
import json
import multiprocessing
from pathlib import Path
import re
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from unittest.mock import patch

from app.config import settings
from app.file_storage import BERLIN, UploadError, get_file, get_files, save_upload
from tests import test_forderung_history as history_fixture

KEY = {'Client_DB': '1', 'Beleg_Nr': 'R1'}

def worker(root, value):
    settings.LOCAL_STORAGE = root
    save_upload(KEY, 'same.txt', io.BytesIO(value), 100)

class StorageTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        config = patch.object(settings, 'LOCAL_STORAGE', self.temp.name)
        config.start(); self.addCleanup(config.stop)

    def save(self, name='a.txt', data=b'new', limit=100):
        return save_upload(KEY, name, io.BytesIO(data), limit)

    def test_unicode_empty_and_exact_limit(self):
        for name, data in [('Rechnung ä + #%20.pdf', b'12345'), ('empty.bin', b'')]:
            relative = self.save(name, data, 5)
            self.assertEqual(get_file(relative).read_bytes(), data)
        self.assertEqual(len(get_files(KEY)), 2)

    def test_invalid_names_and_paths(self):
        for name in ['', '.', '..', '../escape', 'a/b', 'a\\b', 'CON.txt', 'LPT¹', 'a:', 'x?', 'x ', 'x.', 'a\x00b', 'a'*256]:
            with self.subTest(name=name), self.assertRaises(UploadError):
                self.save(name)
        self.assertFalse(list(self.root.iterdir()))
        for name in ['../outside', '/1_R1/a', '1_R1/../a', '.upload-staging/a']:
            with self.assertRaises(FileNotFoundError): get_file(name)

    def test_size_checked_before_archiving(self):
        self.save(data=b'old')
        with self.assertRaises(UploadError): self.save(data=b'123456', limit=5)
        self.assertEqual([f.name for f in get_files(KEY)], ['a.txt'])
        self.assertEqual(get_file('1_R1/a.txt').read_bytes(), b'old')
        self.assertEqual(list((self.root/'.upload-staging').iterdir()), [])

    def test_both_layouts_archive_and_timestamp_counter(self):
        self.save(data=b'first')
        legacy = self.root/'1'/'R1'; legacy.mkdir(parents=True)
        (legacy/'a.txt').write_bytes(b'legacy')
        with patch('app.file_storage.datetime') as clock:
            clock.now.return_value = datetime(2026, 1, 2, 3, 4, 5, 123456, tzinfo=BERLIN)
            self.save(data=b'second'); self.save(data=b'third')
            clock.now.assert_called_with(BERLIN)
        files = get_files(KEY)
        self.assertEqual(len(files), 4)
        self.assertTrue((self.root/'1_R1/a_20260102_030405_123456_1.txt').exists())
        self.assertTrue((legacy/'a_20260102_030405_123456.txt').exists())
        self.assertEqual(get_file('1_R1/a.txt').read_bytes(), b'third')
        self.assertEqual([f.modified_at.timestamp() for f in files], sorted([f.modified_at.timestamp() for f in files], reverse=True))

    def test_failed_final_replace_restores_both_files(self):
        self.save(data=b'newfolder')
        legacy = self.root/'1/R1'; legacy.mkdir(parents=True); (legacy/'a.txt').write_bytes(b'legacy')
        with patch('app.file_storage.os.replace', side_effect=PermissionError('locked')):
            with self.assertRaises(PermissionError): self.save()
        self.assertEqual(get_file('1_R1/a.txt').read_bytes(), b'newfolder')
        self.assertEqual(get_file('1/R1/a.txt').read_bytes(), b'legacy')
        self.assertEqual(len(get_files(KEY)), 2)

    def test_parallel_threads_preserve_all_versions(self):
        with ThreadPoolExecutor(max_workers=6) as pool:
            list(pool.map(lambda n: self.save(data=str(n).encode()), range(10)))
        self.assertEqual({get_file(f.relative_path).read_bytes() for f in get_files(KEY)}, {str(n).encode() for n in range(10)})

    def test_parallel_processes_preserve_all_versions(self):
        processes = [multiprocessing.get_context('spawn').Process(target=worker, args=(self.temp.name, str(n).encode())) for n in range(4)]
        for process in processes: process.start()
        for process in processes:
            process.join(20)
            if process.is_alive(): process.terminate(); self.fail('Upload process timed out')
            self.assertEqual(process.exitcode, 0)
        self.assertEqual(len(get_files(KEY)), 4)

class UploadRouteTests(unittest.TestCase):
    def setUp(self):
        self.fixture = history_fixture.HistoryTests()
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        self.client, self.db = self.fixture.client, self.fixture.db
        self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup)
        config = patch.multiple(settings, LOCAL_STORAGE=self.temp.name, UPLOAD_MAX_FILE_SIZE_MB=1)
        config.start(); self.addCleanup(config.stop)

    def post(self, files=(), comment='', headers=None, key='1_R1', ids=None):
        return self.client.post('/app/forderungen/details/'+key,
            data={'source':'comments','Kommentar':comment,'attachment_ids': json.dumps(ids if ids is not None else [str(n) for n in range(len(files))]),'comment_scope':'debtor'},
            files=[('attachments', (name, body, 'application/octet-stream')) for name, body in files],
            headers={'X-CSRF-Token':'csrf'} if headers is None else headers)

    def result(self, response):
        self.assertEqual(response.status_code, 200, response.text)
        return json.loads(re.search(r'data-gc-send-result>(.*?)</script>', response.text, re.S)[1])

    def test_file_only_no_comment_and_first_list(self):
        self.db.connection.execute('UPDATE dbo.Forderungen_Kommentar SET Aktuell_Fibu = 1, Aktuell_Abteilung = 1')
        self.db.connection.commit()
        before = self.db.connection.execute('SELECT * FROM dbo.Forderungen_Kommentar').fetchall()
        response = self.post([('ä # + %20.txt', b'content')])
        result = self.result(response)
        self.assertFalse(result['comment_saved']); self.assertTrue(result['files'][0]['success'])
        self.assertEqual(self.db.commits, 0)
        self.assertEqual(self.db.connection.execute('SELECT * FROM dbo.Forderungen_Kommentar').fetchall(), before)
        self.assertIn('id="case-files"', response.text)
        self.assertIn('hx-swap-oob="outerHTML"', response.text)
        name = get_files(KEY)[0].relative_path
        download = self.client.get('/files/', params={'name': name})
        self.assertEqual(download.content, b'content'); self.assertEqual(download.status_code, 200)
        self.assertIn('filename*=', download.headers['content-disposition'])

    def test_combined_partial_and_comment_saved_once(self):
        result = self.result(self.post([('a.txt', b'ok'), ('CON.txt', b'bad')], '123,45 <Text>'))
        self.assertEqual([r['success'] for r in result['files']], [True, False])
        self.assertTrue(result['comment_saved'])
        row = self.db.connection.execute('SELECT Kommentar FROM dbo.Forderungen_Kommentar ORDER BY rowid DESC LIMIT 1').fetchone()
        self.assertEqual(row[0], '123,45 <Text>')
        retry = self.result(self.post([('CON.txt', b'bad')]))
        self.assertFalse(retry['comment_saved']); self.assertEqual(self.db.commits, 1)
        self.assertEqual(len(get_files(KEY)), 1)

    def test_same_filenames_are_matched_by_individual_id(self):
        original = save_upload
        calls = 0
        def save(key, name, stream, limit):
            nonlocal calls
            calls += 1
            if calls == 1: raise UploadError('Please retry this file')
            return original(key, name, stream, limit)
        with patch('app.comment_submission.save_upload', side_effect=save):
            result = self.result(self.post([('same.txt', b'first'), ('same.txt', b'second')], ids=['first', 'second']))
        self.assertEqual([(f['id'], f['success']) for f in result['files']], [('first', False), ('second', True)])
        self.assertEqual(get_file('1_R1/same.txt').read_bytes(), b'second')

    def test_comment_failure_keeps_successful_file(self):
        execute = self.db.execute
        def failing(query, values=None):
            if 'INSERT INTO' in str(query): raise RuntimeError('database unavailable')
            return execute(query, values)
        with self.assertLogs('app.comment_submission', level='ERROR'), patch.object(self.db, 'execute', side_effect=failing), patch.object(self.db, 'rollback', create=True):
            result = self.result(self.post([('a.txt', b'ok')], 'retain me'))
        self.assertTrue(result['files'][0]['success']); self.assertFalse(result['comment_saved'])
        self.assertTrue(result['messages']); self.assertEqual(len(get_files(KEY)), 1)
        self.assertTrue(self.result(self.post(comment='retain me'))['comment_saved'])

    def test_size_boundary_partial_success_and_empty(self):
        result = self.result(self.post([('exact.bin', b'a'*(1024*1024)), ('large.bin', b'a'*(1024*1024+1)), ('empty', b'')]))
        self.assertEqual([r['success'] for r in result['files']], [True, False, True])
        self.assertEqual(len(get_files(KEY)), 2)
        self.assertFalse(self.result(self.post(comment='  '))['comment_saved'])
        self.assertEqual(self.db.commits, 0)

    def test_refresh_failure_still_reports_persisted_results(self):
        with self.assertLogs('app.comment_submission', level='ERROR'), patch('app.comment_submission.history_context', side_effect=RuntimeError('history unavailable')), patch.object(self.db, 'rollback', create=True):
            response = self.post([('a.txt', b'ok')], 'saved text')
        result = self.result(response)
        self.assertEqual(response.headers['hx-reswap'], 'none')
        self.assertTrue(result['comment_saved']); self.assertTrue(result['files'][0]['success'])
        self.assertIn('case-files', response.text)
        self.assertTrue(result['messages'])

    def test_storage_permission_error_is_individual(self):
        with self.assertLogs('app.comment_submission', level='ERROR'), patch('app.comment_submission.save_upload', side_effect=PermissionError('denied')):
            result = self.result(self.post([('a.txt', b'ok')], 'save comment independently'))
        self.assertFalse(result['files'][0]['success']); self.assertIn('Zugriffsrechte', result['files'][0]['error'])
        self.assertTrue(result['comment_saved']); self.assertEqual(get_files(KEY), [])

    def test_security_manifest_and_legacy_download(self):
        self.assertEqual(self.post([('a.txt', b'x')], headers={}).status_code, 403)
        self.assertEqual(self.post([('a.txt', b'x')], key='1_R2').status_code, 404)
        self.assertEqual(self.post([('a.txt', b'x')], ids=[]).status_code, 400)
        self.assertFalse(list(Path(self.temp.name).iterdir()))
        legacy = Path(self.temp.name)/'1/R1'; legacy.mkdir(parents=True); (legacy/'old.txt').write_bytes(b'old')
        self.assertEqual(self.client.get('/files/', params={'name':'1/R1/old.txt'}).content, b'old')
        self.assertEqual(self.client.get('/files/', params={'name':'1/R2/old.txt'}).status_code, 404)
        self.client.cookies.clear()
        response = self.post([('a.txt', b'x')], headers={'HX-Request':'true','HX-Current-URL':'http://testserver/app/forderungen/details/1_R1'})
        self.assertIn('/login?', response.headers['hx-redirect'])
        self.assertEqual(len(get_files(KEY)), 1)

if __name__ == '__main__': unittest.main()