dak/tests/test_graph_new.py

# Tests for dak/graph_new.py
# Copyright 2020 Federico Ceratto <federico@debian.org>
# License: GPL-2+

from datetime import datetime, timedelta

from dak.graph_new import gen_stats, process_events


def test_gen_stats():
    now = datetime(2012, 8, 13, 0, 17, 49)
    q = {
        "python-pyknon_1.0-1_amd64": datetime(2012, 8, 12, 13, 46),
        "seascope_0.6.1.hgr112-1_amd64": datetime(2012, 8, 12, 13, 46, 6),
        "simgear_2.6.0-2_i386": datetime(2012, 8, 12, 13, 46, 8),
        "simgear_2.6.0-3_i386": datetime(2012, 8, 12, 13, 46, 10),
        "storymaps_1.0-1_i386": datetime(2012, 8, 12, 13, 46, 13),
        "telepathy-salut_0.8.0-3_amd64": datetime(2012, 8, 12, 13, 46, 17),
        "webauth_4.3.0-1_i386": datetime(2012, 8, 12, 13, 46, 20),
        "webauth_4.3.1-1_i386": datetime(2012, 8, 12, 13, 46, 24),
        "wifite_2.0r85-1_i386": datetime(2012, 8, 12, 13, 46, 27),
        "xwiimote_0.3+20120630-1_amd64": datetime(2012, 8, 12, 13, 46, 28),
        "yii_1.1.11-1_amd64": datetime(2012, 8, 12, 13, 46, 30),
        "nut_2.6.5-1_amd64": datetime(2012, 8, 13, 0, 17, 49),
    }
    stats = gen_stats(now, q)
    assert stats == (
        timedelta(seconds=37909),
        timedelta(seconds=37909),
        timedelta(seconds=37889),
    )


def test_processing():
    queue = {}
    init_time = last_update = None
    events = []
    rrdtool_updates, begin_time = process_events(events, last_update, init_time, queue)
    assert rrdtool_updates == []
    assert queue == {}
    events = [
        ("a", "Policy Queue ACCEPT", datetime(2020, 11, 19, 0, 0, 1)),
        ("x", "ACCEPT-TO-NEW", datetime(2020, 11, 19, 0, 0, 2)),
        ("y", "ACCEPT-TO-NEW", datetime(2020, 11, 19, 0, 0, 3)),
        ("a", "ACCEPT-TO-NEW", datetime(2020, 11, 22, 0, 0, 1)),
        ("b", "ACCEPT-TO-NEW", datetime(2020, 11, 22, 0, 0, 2)),
        ("a", "REJECT", datetime(2020, 11, 23, 0, 0, 1)),
    ]
    # for pname, event, event_time in events:
    rrdtool_updates, begin_time = process_events(events, last_update, init_time, queue)
    assert queue == {
        "b": datetime(2020, 11, 22, 0, 0, 2),
        "x": datetime(2020, 11, 19, 0, 0, 2),
        "y": datetime(2020, 11, 19, 0, 0, 3),
    }
    assert rrdtool_updates == ["1606003201:2:2:2", "1606089601:3:3:3"]