Reisacher-Forderung/tests/test_comments.py

"""Exercise both production comment entry points and transactional flag updates."""

import json
import re
import unittest
from unittest.mock import patch

from tests import test_wiedervorlage as fixture


class CommentFlagsTests(unittest.TestCase):
    def setUp(self):
        self.fixture = fixture.WiedervorlageTests()
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        self.db = self.fixture.db
        self.db.connection.execute(
            'UPDATE dbo.Forderungen_Kommentar SET Aktuell_Fibu = 1, Aktuell_Abteilung = 1'
        )
        self.db.connection.execute("""
            INSERT INTO dbo.Forderungen_Kommentar
                (Client_DB, Beleg_Nr, Timestamp, Benutzer_ID, Kommentar, Aktuell_Fibu, Aktuell_Abteilung)
            VALUES ('1', 'R1', '2026-01-03T12:00:00', 'other', 'Second old comment', 1, 1)
        """)
        self.db.connection.commit()

    def snapshot(self):
        return self.db.connection.execute(
            'SELECT * FROM dbo.Forderungen_Kommentar ORDER BY rowid'
        ).fetchall()

    def post(self, channel, day=6):
        if channel == 'wv':
            response = self.fixture.post(f'2026-11-{day:02}', begruendung='Neue Zusage')
            return response.status_code == 200
        response = self.fixture.client.post(
            '/app/forderungen/details/1_R1',
            data={'source': 'comments', '_csrf': 'csrf', 'Kommentar': 'Chat comment',
                  'Client_DB': '2', 'Beleg_Nr': 'R2', 'Benutzer_ID': 'intruder', 'Rolle_ID': 'FIBU'},
        )
        self.assertEqual(response.status_code, 200)
        result = json.loads(re.search(r'data-gc-send-result>(.*?)</script>', response.text, re.S)[1])
        return result['comment_saved']

    def assert_flags(self, before, fibu):
        rows = self.snapshot()
        self.assertEqual(len(rows), len(before) + 1)
        for previous, current in zip(before, rows):
            expected = list(previous)
            if previous[:2] == ('1', 'R1'):
                expected[5 if fibu else 6] = 0
            self.assertEqual(current, tuple(expected))
        self.assertEqual(rows[-1][:2], ('1', 'R1'))
        self.assertEqual(rows[-1][3], 'tester')
        self.assertEqual(rows[-1][5:], (int(fibu), int(not fibu)))

    def test_both_channels_reset_only_author_group_and_invoice(self):
        # Alternate roles and entry points on the same invoice, including repeated writes.
        for day, (channel, fibu) in enumerate(
            [('chat', False), ('wv', False), ('chat', True), ('wv', True), ('chat', True), ('wv', False)], 6
        ):
            with self.subTest(channel=channel, fibu=fibu):
                self.db.connection.execute("DELETE FROM dbo.Benutzer_Rolle WHERE Rolle_ID = 'FIBU'")
                if fibu:
                    # Existing MA role remains: any FIBU assignment must qualify.
                    self.db.connection.execute(
                        "INSERT INTO dbo.Benutzer_Rolle VALUES ('tester', 'Tester', 'Buchhaltung', 'FIBU')"
                    )
                self.db.connection.commit()
                before = self.snapshot()
                self.assertTrue(self.post(channel, day))
                self.assert_flags(before, fibu)

    def test_insert_update_role_lookup_and_commit_failures_roll_back_everything(self):
        for channel in ['chat', 'wv']:
            for failure in ['INSERT INTO dbo.Forderungen_Kommentar', 'UPDATE dbo.Forderungen_Kommentar',
                            'FROM dbo.Benutzer_Rolle', 'commit']:
                with self.subTest(channel=channel, failure=failure):
                    before = self.snapshot()
                    execute = self.db.execute

                    def failing(query, values=None):
                        if failure in str(query):
                            raise RuntimeError('database unavailable')
                        return execute(query, values)

                    target = 'app.wiedervorlage' if channel == 'wv' else 'app.comment_submission'
                    if failure == 'commit':
                        mocked = patch.object(self.db, 'commit', side_effect=RuntimeError('commit failed'))
                    else:
                        mocked = patch.object(self.db, 'execute', side_effect=failing)
                    with mocked, self.assertLogs(target, level='ERROR'):
                        self.assertFalse(self.post(channel))
                    self.assertEqual(self.snapshot(), before)
                    self.assertIsNone(self.fixture.row())

    def test_lock_is_acquired_before_any_write_in_both_channels(self):
        for channel in ['chat', 'wv']:
            queries = []
            execute = self.db.execute

            def recording(query, values=None):
                queries.append(str(query))
                return execute(query, values)

            with patch.object(self.db, 'execute', side_effect=recording):
                self.assertTrue(self.post(channel))
            lock = next(i for i, sql in enumerate(queries)
                        if 'Forderungen_Kopf WITH (UPDLOCK, HOLDLOCK)' in sql)
            writes = [i for i, sql in enumerate(queries) if re.match(r'\s*(INSERT|UPDATE)\b', sql)]
            self.assertTrue(writes)
            self.assertLess(lock, min(writes))

    def test_unchanged_date_does_not_reset_flags(self):
        before = self.snapshot()
        self.assertEqual(self.fixture.post('2026-10-15', begruendung='Text').status_code, 200)
        self.assertEqual(self.snapshot(), before)


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