|
|
@@ -0,0 +1,121 @@
|
|
|
+"""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()
|
gc-server3