(mm_obj)
| 232 | per_thread_loop = 10 |
| 233 | |
| 234 | def find_and_rfind(mm_obj): |
| 235 | pattern = b'Thread-Ident:"%d"' % threading.get_ident() |
| 236 | mm_obj.write(pattern) |
| 237 | for _ in range(per_thread_loop): |
| 238 | found_at = mm_obj.find(pattern, 0) |
| 239 | self.assertNotEqual(found_at, -1) |
| 240 | # Should not find it after the `found_at` |
| 241 | self.assertEqual(mm_obj.find(pattern, found_at + 1), -1) |
| 242 | found_at_rev = mm_obj.rfind(pattern, 0) |
| 243 | self.assertEqual(found_at, found_at_rev) |
| 244 | # Should not find it after the `found_at` |
| 245 | self.assertEqual(mm_obj.rfind(pattern, found_at + 1), -1) |
| 246 | |
| 247 | with mmap.mmap(ANONYMOUS_MEM, 1024) as mm_obj: |
| 248 | run_concurrently( |
nothing calls this directly
no test coverage detected