(self, method)
| 1452 | OID = "test_watch_notify" |
| 1453 | |
| 1454 | def setup_method(self, method): |
| 1455 | self.rados = Rados(conffile='') |
| 1456 | self.rados.connect() |
| 1457 | self.rados.create_pool('test_pool') |
| 1458 | assert self.rados.pool_exists('test_pool') |
| 1459 | self.ioctx = self.rados.open_ioctx('test_pool') |
| 1460 | self.ioctx.write(self.OID, b'test watch notify') |
| 1461 | self.lock = threading.Condition() |
| 1462 | self.notify_cnt = {} |
| 1463 | self.notify_data = {} |
| 1464 | self.notify_error = {} |
| 1465 | # aio related |
| 1466 | self.ack_cnt = {} |
| 1467 | self.ack_data = {} |
| 1468 | self.instance_id = self.rados.get_instance_id() |
| 1469 | |
| 1470 | def teardown_method(self, method): |
| 1471 | self.ioctx.close() |
nothing calls this directly
no test coverage detected