I would like to write a unit test which ensures that a race condition can't happen if Thread A wants to write a row out and Thread B wants to delete it. (I already know that at the database level, the race condition will be resolved, but the way my application is written currently, if such a race condition happens, it will get settled by one of the threads dying and needing to get restarted... it's something the application can recover from, but it's a situation that is good to avoid.)
Inconveniently, Thread A polls something that doesn't change once it reaches a final state, and then when it sees that the final state has been reached, Thread A writes stuff out in the database to signal to Thread B to do new stuff. Thread B meanwhile looks at the row, does things, then deletes it when it's done. Since Thread A enables Thread B to do work on something which doesn't change in between Thread A being done with it and Thread B starting to deal with it, the only way I can have this race condition which I'm testing for is actually just writing the same data out again. That is, I can't just make an assertion that the data in the table is the same as it was before and say that's good enough - what I want to do is to assert that no updates on a given row of the table were committed at all by Thread A. (If I were to add a column like "last written at this time" that would be silly because that would be adding a column just to make my unit tests more convenient to write, not for the benefit of the application, so that's also not an option.)
I am using SQLAlchemy + postgres and my test is in a pytest suite. What's a nice way to write code which will fail the test if the code the test is exercising writes out a row to the database?
One thing I am thinking of is adding a postgres-level constraint just for the duration of the test but I think this feels really messy to me, and if I could figure out a way to do it by inspecting the SQLAlchemy Engine that was in use by Thread A, or something like that, that would be way better. But it is not obvious to me how to tell this from the SQLAlchemy docs - maybe it is obvious to someone else? :)
I don't have a code example at this time because I don't know where to get started.