# Regression test for the split-dict iteration NULL-value crash.
# In the fix diffs it is added to Lib/test/test_free_threading/test_dict.py
# (class TestDict). It is reproduced here standalone so it can be run with
# any free-threaded interpreter:  python.exe test_split_dict_iteration.py
# The standalone copy refuses to run with the GIL enabled (exit status 4),
# because it would then pass without testing anything.
import sys
import unittest
from functools import partial

from test.support import threading_helper

if getattr(sys, "_is_gil_enabled", lambda: True)():
    sys.stderr.write("SKIP: needs a free-threaded build with the GIL disabled "
                     "(sys._is_gil_enabled() is not False). Nothing tested.\n")
    sys.exit(4)


@threading_helper.requires_working_threading()
class TestDict(unittest.TestCase):
    def test_racing_split_dict_iteration_and_delete(self):
        # Lock-free iteration of a split (instance) dict must not crash when
        # a value slot is cleared by a concurrent delete or clear().
        # Int values already expose the NULL dereference. The values are
        # lists (mortal objects) to also cover the refcount and lifetime of
        # non-immortal values.
        class C:
            pass

        names = [f"a{i}" for i in range(8)]
        obj = C()
        for i, name in enumerate(names):
            setattr(obj, name, [i])
        d = obj.__dict__

        def delattr_writer():
            for _ in range(500):
                for i, name in enumerate(names):
                    delattr(obj, name)
                    setattr(obj, name, [i])

        def delitem_writer():
            for _ in range(500):
                for i, name in enumerate(names):
                    try:
                        del d[name]
                    except KeyError:
                        pass
                    d[name] = [i]

        def clear_writer():
            for _ in range(500):
                d.clear()
                for i, name in enumerate(names):
                    setattr(obj, name, [i])

        def reader(view):
            for _ in range(2000):
                try:
                    for _ in view():
                        pass
                except RuntimeError:
                    pass

        for writer in (delattr_writer, delitem_writer, clear_writer):
            for view in (d.values, d.items, d.keys):
                with self.subTest(writer=writer.__name__, view=view.__name__):
                    threading_helper.run_concurrently(
                        [writer, partial(reader, view), partial(reader, view)])
                    self.assertEqual(d, {name: [i] for i, name in enumerate(names)})


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