| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768 |
- Subject: Add test
- Origin: upstream, commit 0.4.27-52-g892543d <https://github.com/ahupp/python-magic/commit/0.4.27-52-g892543d>
- Author: ddelange <14880945+ddelange@users.noreply.github.com>
- Date: Tue Oct 14 14:39:45 2025 +0300
- Forwarded: not-needed
- --- a/magic/__init__.py
- +++ b/magic/__init__.py
- @@ -416,7 +416,8 @@
- raise NotImplementedError("magic_getparam not implemented")
- val = c_size_t()
- with LOCK:
- - return _magic_getparam(cookie, param, byref(val)).value
- + _magic_getparam(cookie, param, byref(val))
- + return val.value
-
-
- _has_version = False
- @@ -427,7 +428,7 @@
- magic_version.argtypes = []
-
-
- -def version(lock=None):
- +def version():
- if not _has_version:
- raise NotImplementedError("magic_version not implemented")
- with LOCK:
- --- a/test/python_magic_test.py
- +++ b/test/python_magic_test.py
- @@ -10,6 +10,12 @@
-
- import pytest
-
- +try:
- + from concurrent.futures import ThreadPoolExecutor
- + HAS_CONCURRENT_FUTURES = True
- +except ImportError: # python 2.7
- + HAS_CONCURRENT_FUTURES = False
- +
- # for output which reports a local time
- os.environ["TZ"] = "GMT"
-
- @@ -321,6 +327,25 @@
-
- self.assertRaises(IOError, m_follow.from_file, tmp_broken)
-
- + @unittest.skipIf(not HAS_CONCURRENT_FUTURES, "concurrent.futures not available in Python 2.7")
- + def test_thread_safety(self):
- + """Test that concurrent from_file calls don't crash (would SEGV without global lock)"""
- + filename = os.path.join(self.TESTDATA_DIR, "test.pdf")
- +
- + m = magic.Magic(mime=True)
- +
- + def check_file(_):
- + result = m.from_file(filename)
- + self.assertEqual(result, "application/pdf")
- + return result
- +
- + with ThreadPoolExecutor(100) as executor:
- + results = list(executor.map(check_file, range(100)))
- +
- + # All calls should complete successfully
- + self.assertEqual(len(results), 100)
- + self.assertTrue(all(r == "application/pdf" for r in results))
- +
-
- if __name__ == "__main__":
- unittest.main()
|
