test_archiver.py 文件源码

def test_smrt_archiver_lasttime_clear():
    tmpfile = tempfile.mktemp()
    archiver = Archiver(dbfile=tmpfile)
    rule = 'test/smrt/rules/archiver.yml'
    feed = 'lasttime'

    with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
        assert type(s) is Smrt

        for r, f in s.load_feeds(rule, feed=feed):
            x = list(s.process(r, f))
            assert len(x) > 0

            f = {i.indicator: i.__dict__() for i in x}
            assert f['216.243.31.2']['lasttime'] == '2016-03-23T20:22:27.000000Z'

    archiver.clear_memcache()
    with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
        assert type(s) is Smrt

        for r, f in s.load_feeds(rule, feed=feed):
            x = list(s.process(r, f))
            assert len(x) == 0