def _test_run_expirer_ttl_enabled(self, ttl_name, data_name):
content = ("[database]\n"
"%s=1\n"
"connection=log://localhost\n" % ttl_name)
if six.PY3:
content = content.encode('utf-8')
self.tempfile = fileutils.write_to_tempfile(content=content,
prefix='panko',
suffix='.conf')
subp = subprocess.Popen(['panko-expirer',
'-d',
"--config-file=%s" % self.tempfile],
stdout=subprocess.PIPE)
out, __ = subp.communicate()
self.assertEqual(0, subp.poll())
msg = "Dropping %s data with TTL 1" % data_name
if six.PY3:
msg = msg.encode('utf-8')
self.assertIn(msg, out)
评论列表
文章目录