1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
"""Tests for the node audit store (legal compliance IP/action logging)."""
import time
import pytest
from meshbay_node.audit import AuditStore
@pytest.fixture
async def audit(tmp_path):
store = AuditStore(db_path=tmp_path / "audit.db")
await store.open()
yield store
await store.close()
@pytest.mark.asyncio
async def test_log_and_retrieve(audit):
await audit.log_event(
user_id="u1", event="handshake", ip="1.2.3.4",
username="alice", group_id="g1", detail="test")
await audit.log_event(
user_id="u2", event="file_download", ip="5.6.7.8",
username="bob", group_id="g1", detail="video.mp4")
entries = await audit.get_entries()
assert len(entries) == 2
assert entries[0].event == "file_download"
assert entries[0].ip == "5.6.7.8"
assert entries[1].event == "handshake"
@pytest.mark.asyncio
async def test_filter_by_event(audit):
await audit.log_event(user_id="u1", event="handshake", ip="1.1.1.1")
await audit.log_event(user_id="u1", event="file_download", ip="1.1.1.1")
await audit.log_event(user_id="u1", event="handshake", ip="1.1.1.1")
entries = await audit.get_entries(event="handshake")
assert len(entries) == 2
assert all(e.event == "handshake" for e in entries)
@pytest.mark.asyncio
async def test_filter_by_user(audit):
await audit.log_event(user_id="u1", event="handshake")
await audit.log_event(user_id="u2", event="handshake")
entries = await audit.get_entries(user_id="u1")
assert len(entries) == 1
assert entries[0].user_id == "u1"
@pytest.mark.asyncio
async def test_pagination_newest_first(audit):
for i in range(5):
await audit.log_event(user_id="u1", event="connect", detail=f"e{i}")
# distinct timestamps so ORDER BY is deterministic
await audit._db.execute(
"UPDATE audit_log SET timestamp = ? WHERE detail = ?",
(1000 + i, f"e{i}"))
await audit._db.commit()
page1 = await audit.get_entries(limit=2, offset=0)
page2 = await audit.get_entries(limit=2, offset=2)
page3 = await audit.get_entries(limit=2, offset=4)
assert [e.detail for e in page1] == ["e4", "e3"] # newest first
assert [e.detail for e in page2] == ["e2", "e1"]
assert [e.detail for e in page3] == ["e0"] # last, partial page
# offset past the end is empty, not an error
assert await audit.get_entries(limit=2, offset=99) == []
@pytest.mark.asyncio
async def test_entry_count(audit):
assert await audit.entry_count() == 0
await audit.log_event(user_id="u1", event="connect")
await audit.log_event(user_id="u2", event="connect")
assert await audit.entry_count() == 2
@pytest.mark.asyncio
async def test_cleanup_old_entries(audit):
await audit.log_event(user_id="u1", event="old")
# manually backdate
await audit._db.execute(
"UPDATE audit_log SET timestamp = ? WHERE event = 'old'",
(time.time() - 400 * 86400,))
await audit._db.commit()
await audit.log_event(user_id="u2", event="recent")
deleted = await audit.cleanup(retention_days=365)
assert deleted == 1
assert await audit.entry_count() == 1
|