本文整理汇总了Python中testhelper.assertEventEquals函数的典型用法代码示例。如果您正苦于以下问题：Python assertEventEquals函数的具体用法？Python assertEventEquals怎么用？Python assertEventEquals使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了assertEventEquals函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_bad
def test_bad(self):
self.create({"consume": False, "on_error": "tag"}, [Event(data=x) for x in self.bad_packets])
events = self.wait(events=len(self.bad_packets))
bad_events = [Event(data=x, message="invalid syslog", tags=["error"]) for x in self.bad_packets]
for ev in bad_events:
assertEventEquals(self, ev, events.pop(0))
开发者ID:nakatadim，项目名称:logcabin，代码行数:7，代码来源:test_filters.py
示例2: test_good
def test_good(self):
self.create({},
[Event(data=x) for x in self.good_packets])
events = self.wait(events=len(self.good_events))
for ev in self.good_events:
assertEventEquals(self, ev, events.pop(0))
开发者ID:centralman，项目名称:logcabin，代码行数:7，代码来源:test_filters.py
示例3: test_simple
def test_simple(self):
function = mock.Mock()
ev = Event(data="abc123")
self.create({"function": function}, [ev])
q = self.wait()
assertEventEquals(self, Event(data="abc123"), q[0])
function.assert_called_with(ev)
开发者ID:nakatadim，项目名称:logcabin，代码行数:8，代码来源:test_filters.py
示例4: test_bad
def test_bad(self):
self.create({'consume': False, 'on_error': 'tag'},
[Event(data=x) for x in self.bad_packets])
events = self.wait(events=len(self.bad_packets))
bad_events = [Event(data=x, message='invalid syslog', tags=['error']) for x in self.bad_packets]
for ev in bad_events:
assertEventEquals(self, ev, events.pop(0))
开发者ID:centralman，项目名称:logcabin，代码行数:8，代码来源:test_filters.py
示例5: test_event
def test_event(self):
conf = {'port': random.randint(1024, 65535)}
self.create(conf)
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.sendto('abc', ('', conf['port']))
# yield for processing to happen
q = self.waitForQueue()
assertEventEquals(self, Event(data='abc'), q[0])
开发者ID:centralman，项目名称:logcabin，代码行数:10，代码来源:test_inputs.py
示例6: test_allow_hosts
def test_allow_hosts(self):
conf = {'port': random.randint(1024, 65535),
'allow_hosts': ['127.0.0.1']}
self.create(conf)
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.sendto('abc', ('', conf['port']))
q = self.waitForQueue()
assertEventEquals(self, Event(data='abc'), q[0])
开发者ID:srault95，项目名称:logcabin，代码行数:10，代码来源:test_inputs.py
示例7: test_multiple
def test_multiple(self):
with TempDirectory():
conf = {'path': 'test*.log'}
self.create(conf)
with file('test1.log', 'w') as fin:
print >> fin, 'abc'
with file('test2.log', 'w') as fin:
print >> fin, 'abc'
q = self.waitForQueue(events=2)
assertEventEquals(self, Event(data='abc'), q[0])
assertEventEquals(self, Event(data='abc'), q[1])
开发者ID:centralman，项目名称:logcabin，代码行数:13，代码来源:test_inputs.py
示例8: test_truncated
def test_truncated(self):
with TempDirectory():
conf = {'path': 'test*.log'}
self.create(conf)
with file('test1.log', 'w') as fin:
print >> fin, 'abc'
print >> fin, 'def'
gevent.sleep(0.01)
with file('test1.log', 'w') as fin:
print >> fin, 'ghi'
q = self.waitForQueue(events=3)
assertEventEquals(self, Event(data='abc'), q[0])
开发者ID:centralman，项目名称:logcabin，代码行数:14，代码来源:test_inputs.py
示例9: test_success
def test_success(self):
with TempDirectory():
conf = {'path': 'test*.log'}
self.create(conf)
# create log after a short time
gevent.sleep(0.01)
with file('test1.log', 'w') as fin:
print >> fin, 'abc'
gevent.sleep(0.01)
print >> fin, 'def'
q = self.waitForQueue(events=2)
assertEventEquals(self, Event(data='abc'), q[0])
assertEventEquals(self, Event(data='def'), q[1])
开发者ID:centralman，项目名称:logcabin，代码行数:15，代码来源:test_inputs.py
示例10: test_max_size
def test_max_size(self):
with TempDirectory():
self.create({'filename': 'output.log',
'max_size': 16})
map(self.input.put, self.events)
self.waitForEmpty()
self.assertFileContents(self.events[0].to_json()+'\n', 'output.log.1')
self.assertFileContents(self.events[1].to_json()+'\n', 'output.log')
# assert the 'fileroll' event is generated
self.assert_(self.output.qsize())
events = [self.output.get() for i in xrange(self.output.qsize())]
assertEventEquals(self, Event(tags=['fileroll'], filename='output.log.1', last=self.events[0], trigger=self.events[1]), events[1])
开发者ID:artirix，项目名称:logcabin，代码行数:15，代码来源:test_outputs.py
示例11: test_resume
def test_resume(self):
with TempDirectory():
with file('test1.log', 'w') as fin:
print >> fin, 'abc'
conf = {'path': 'test*.log'}
self.create(conf)
q = self.waitForQueue(events=1)
assertEventEquals(self, Event(data='abc'), q[0])
with file('test1.log', 'a') as fin:
print >> fin, 'def'
self.create(conf)
q = self.waitForQueue(events=1)
assertEventEquals(self, Event(data='def'), q[0])
开发者ID:centralman，项目名称:logcabin，代码行数:16，代码来源:test_inputs.py
示例12: test_rolling
def test_rolling(self):
with TempDirectory():
with file('test1.log', 'w') as fin:
print >> fin, 'abc'
conf = {'path': 'test*.log'}
self.create(conf)
gevent.sleep(0.01)
os.rename('test1.log', 'test1.log.1')
with file('test1.log', 'w') as fin:
print >> fin, 'def'
q = self.waitForQueue(events=2)
assertEventEquals(self, Event(data='abc'), q[0])
assertEventEquals(self, Event(data='def'), q[1])
开发者ID:centralman，项目名称:logcabin，代码行数:16，代码来源:test_inputs.py
示例13: test_nested
def test_nested(self):
self.create({"period": 0.1, "metrics": {"rails.{controller}.{action}.{0}": "timings.*"}}, self.events)
# 8 events expected - the above 6, and then 2 stat events
q = self.wait(events=8)
q = [i for i in q if i.stats]
q.sort(key=lambda k: k.metric)
expected = Event(
metric="rails.home.index.timings.view",
stats={
"count": 3,
"max": 2.3,
"mean": 1.5,
"median": 1.2,
"min": 1.0,
"rate": between(1, 100),
"stddev": about(2.34, 2),
"upper95": 2.19,
"upper99": 2.278,
},
tags=["stat"],
)
assertEventEquals(self, expected, q[0])
expected = Event(
metric="rails.home.login.timings.view",
stats={
"count": 1,
"max": 2.0,
"mean": 2.0,
"median": 2.0,
"min": 2.0,
"rate": between(1, 100),
"stddev": 0.0,
"upper95": 2.0,
"upper99": 2.0,
},
tags=["stat"],
)
assertEventEquals(self, expected, q[1])
# wait for further two 'zero' stats
q = self.wait(events=2)
开发者ID:nakatadim，项目名称:logcabin，代码行数:45，代码来源:test_filters.py
示例14: test_nested
def test_nested(self):
self.create({'period': 0.1, 'metrics': {'rails.{controller}.{action}.{0}': 'timings.*'}},
self.events)
# 8 events expected - the above 6, and then 2 stat events
q = self.wait(events=8)
q = [i for i in q if i.stats]
q.sort(key=lambda k: k.metric)
expected = Event(metric='rails.home.index.timings.view',
stats={
'count': 3,
'max': 2.3,
'mean': 1.5,
'median': 1.2,
'min': 1.0,
'rate': between(1, 100),
'stddev': about(2.34, 2),
'upper95': 2.19,
'upper99': 2.278},
tags=['stat'],
)
assertEventEquals(self, expected, q[0])
expected = Event(metric='rails.home.login.timings.view',
stats={
'count': 1,
'max': 2.0,
'mean': 2.0,
'median': 2.0,
'min': 2.0,
'rate': between(1, 100),
'stddev': 0.0,
'upper95': 2.0,
'upper99': 2.0},
tags=['stat'],
)
assertEventEquals(self, expected, q[1])
# wait for further two 'zero' stats
q = self.wait(events=2)
开发者ID:centralman，项目名称:logcabin，代码行数:42，代码来源:test_filters.py
示例15: test_timestamped
def test_timestamped(self):
events = [
Event(message='1', timestamp=datetime(2013, 1, 1, 23, 59, 58)),
Event(message='2', timestamp=datetime(2013, 1, 1, 23, 59, 59)),
Event(message='3', timestamp=datetime(2013, 1, 2, 0, 0, 0)),
Event(message='4', timestamp=datetime(2013, 1, 2, 0, 0, 1))]
with TempDirectory():
self.create({'filename': 'output-{timestamp:%Y%m%d}.log',
'compress': 'gz'})
map(self.input.put, events)
self.waitForEmpty()
self.assertFileContents(events[0].to_json()+'\n'+events[1].to_json()+'\n', 'output-20130101.log.1.gz')
self.assertFileContents(events[2].to_json()+'\n'+events[3].to_json()+'\n', 'output-20130102.log')
# assert the 'fileroll' event is generated
self.assertEquals(len(events)+1, self.output.qsize())
outputs = [self.output.get() for i in xrange(self.output.qsize())]
assertEventEquals(self, Event(tags=['fileroll'], filename='output-20130101.log.1.gz', last=events[1], trigger=events[2]), outputs[2])
开发者ID:artirix，项目名称:logcabin，代码行数:21，代码来源:test_outputs.py
示例16: test_wildcard
def test_wildcard(self):
self.create({"period": 0.1, "metrics": {"rails.{controller}.{action}.{0}": "*"}}, self.events)
# 8 events expected - the above 6, and then 4 stat events
q = self.wait(events=10)
q = [i for i in q if i.stats]
q.sort(key=lambda k: k.metric)
expected = Event(
metric="rails.home.index.bytes",
stats={
"count": 3,
"rate": between(1, 100),
"max": 30159,
"min": 6926,
"median": 18150,
"mean": 18411,
"stddev": about(30789),
"upper95": 28958.1,
"upper99": 29918.82,
},
tags=["stat"],
)
assertEventEquals(self, expected, q[0])
expected = Event(
metric="rails.home.index.duration",
stats={
"count": 3,
"rate": between(1, 100),
"max": 4.0,
"min": 3.0,
"median": 3.5,
"mean": 3.5,
"stddev": 5.0,
"upper95": 3.95,
"upper99": 3.99,
},
tags=["stat"],
)
assertEventEquals(self, expected, q[1])
expected = Event(
metric="rails.home.login.duration",
stats={
"count": 1,
"rate": between(1, 100),
"max": 2.4,
"min": 2.4,
"median": 2.4,
"mean": 2.4,
"stddev": 0.0,
"upper95": 2.4,
"upper99": 2.4,
},
tags=["stat"],
)
assertEventEquals(self, expected, q[3])
开发者ID:nakatadim，项目名称:logcabin，代码行数:59，代码来源:test_filters.py
示例17: test_wildcard
def test_wildcard(self):
self.create({'period': 0.1, 'metrics': {'rails.{controller}.{action}.{0}': '*'}},
self.events)
# 8 events expected - the above 6, and then 4 stat events
q = self.wait(events=10)
q = [i for i in q if i.stats]
q.sort(key=lambda k: k.metric)
expected = Event(metric='rails.home.index.bytes',
stats={
'count': 3,
'rate': between(1, 100),
'max': 30159,
'min': 6926,
'median': 18150,
'mean': 18411,
'stddev': about(30789),
'upper95': 28958.1,
'upper99': 29918.82,
},
tags=['stat'],
)
assertEventEquals(self, expected, q[0])
expected = Event(metric='rails.home.index.duration',
stats={
'count': 3,
'rate': between(1, 100),
'max': 4.0,
'min': 3.0,
'median': 3.5,
'mean': 3.5,
'stddev': 5.0,
'upper95': 3.95,
'upper99': 3.99,
},
tags=['stat'],
)
assertEventEquals(self, expected, q[1])
expected = Event(metric='rails.home.login.duration',
stats={
'count': 1,
'rate': between(1, 100),
'max': 2.4,
'min': 2.4,
'median': 2.4,
'mean': 2.4,
'stddev': 0.0,
'upper95': 2.4,
'upper99': 2.4,
},
tags=['stat'],
)
assertEventEquals(self, expected, q[3])
开发者ID:centralman，项目名称:logcabin，代码行数:57，代码来源:test_filters.py
示例18: test_true
def test_true(self):
self.create({},
[Event(a=1)])
q = self.wait()
assertEventEquals(self, Event(a=1, b=1), q[0])
开发者ID:artirix，项目名称:logcabin，代码行数:5，代码来源:test_flow.py
示例19: test_missing
def test_missing(self):
self.create({},
[Event()])
q = self.wait()
assertEventEquals(self, Event(b=False), q[0])
开发者ID:artirix，项目名称:logcabin，代码行数:5，代码来源:test_flow.py
示例20: test_default
def test_default(self):
self.create({},
[Event(a=3)])
q = self.wait()
assertEventEquals(self, Event(a=3, b=False), q[0])
开发者ID:artirix，项目名称:logcabin，代码行数:5，代码来源:test_flow.py
注：本文中的testhelper.assertEventEquals函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论