package routewatch import ( "encoding/json" "fmt" "maps" "slices" "sync" "testing" "time" "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/ristypes" ) const ( testASNA = 64500 testASNB = 64501 testASNC = 64502 testASND = 64503 // testMaxTrackedPaths is a limit of the path map that a test fills // quickly, in place of maxTrackedPaths. testMaxTrackedPaths = 10 ) // recordingStore wraps mockStore to count every RecordPeering call, so a // test can tell how many times a peering was written across separate runs. type recordingStore struct { *mockStore mu sync.Mutex calls int } func (r *recordingStore) RecordPeering( asA, asB int, seen database.SeenTimes, ) error { r.mu.Lock() r.calls++ r.mu.Unlock() return r.mockStore.RecordPeering(asA, asB, seen) } func (r *recordingStore) callCount() int { r.mu.Lock() defer r.mu.Unlock() return r.calls } // newTestHandler builds a PeeringHandler without starting the periodic // processing goroutine, so tests drive processing explicitly. func newTestHandler(db database.Store) *PeeringHandler { return &PeeringHandler{ db: db, logger: logger.New(), asPaths: make(map[string]database.SeenTimes), maxPaths: maxTrackedPaths, } } // snapshot returns how many paths the handler tracks and how many it dropped. func snapshot(h *PeeringHandler) (int, int) { h.mu.Lock() defer h.mu.Unlock() return len(h.asPaths), h.droppedPaths } func pathKey(t *testing.T, asns ...int) string { t.Helper() b, err := json.Marshal(asns) if err != nil { t.Fatalf("failed to marshal path: %v", err) } return string(b) } func handle(h *PeeringHandler, ts time.Time, asns ...int) { h.HandleMessage(&ristypes.RISMessage{ Path: ristypes.ASPath{Sequence: asns}, ParsedTimestamp: ts, }) } // decodePath decodes an AS path from the JSON the feed writes for it, in which // an AS set is an array inside the path: decodePath(t, 1, 2, []int{3, 4}) // decodes [1, 2, [3, 4]]. func decodePath(t *testing.T, elements ...any) ristypes.ASPath { t.Helper() data, err := json.Marshal(elements) if err != nil { t.Fatalf("failed to marshal path: %v", err) } var path ristypes.ASPath err = json.Unmarshal(data, &path) if err != nil { t.Fatalf("failed to decode path %s: %v", data, err) } return path } // TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings handles AS paths that // end in an AS set. The members of a set are not neighbours of each other or of // the AS before the set, so a set of several members is in no peering; a set of // one member counts as that AS. func TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings(t *testing.T) { t.Parallel() tests := []struct { name string path []any // The peerings recorded, keyed as the mock store keys them. wantPeerings []string }{ { name: "a set of several members", path: []any{testASNA, testASNB, []int{testASNC, testASND}}, wantPeerings: []string{fmt.Sprintf("%d_%d", testASNA, testASNB)}, }, { name: "a set of one member", path: []any{testASNA, testASNB, []int{testASNC}}, wantPeerings: []string{ fmt.Sprintf("%d_%d", testASNA, testASNB), fmt.Sprintf("%d_%d", testASNB, testASNC), }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() store := newMockStore() h := newTestHandler(store) h.HandleMessage(&ristypes.RISMessage{ Path: decodePath(t, tt.path...), ParsedTimestamp: time.Now().UTC(), }) h.ProcessPeeringsNow() got := slices.Sorted(maps.Keys(store.Peerings)) if !slices.Equal(got, tt.wantPeerings) { t.Errorf("peerings recorded = %v, want %v", got, tt.wantPeerings) } }) } } // TestPeeringHandlerRecordsNothingFromAPathWithASZero handles AS paths that // hold AS 0. RFC 7607 makes such a path malformed, so no peering is recorded // from it, not even one between two other ASes of the path. func TestPeeringHandlerRecordsNothingFromAPathWithASZero(t *testing.T) { t.Parallel() tests := []struct { name string path []any }{ {name: "AS 0 between two ASes", path: []any{testASNA, 0, testASNB}}, { name: "AS 0 after two ASes", path: []any{testASNA, testASNB, 0, testASNC}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() store := newMockStore() h := newTestHandler(store) h.HandleMessage(&ristypes.RISMessage{ Path: decodePath(t, tt.path...), ParsedTimestamp: time.Now().UTC(), }) h.ProcessPeeringsNow() if len(store.Peerings) > 0 { t.Errorf("peerings recorded = %v, want none", slices.Sorted(maps.Keys(store.Peerings))) } }) } } // TestPeeringHandlerProcessesEachRunAndEmpties verifies that a processing run // empties the path map (the swap) and that a path seen again after a run is // recorded in the next run too. func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) { t.Parallel() store := &recordingStore{mockStore: newMockStore()} h := newTestHandler(store) now := time.Now().UTC() // Run 1: one path, one peering recorded, map emptied afterwards. handle(h, now, testASNA, testASNB) h.ProcessPeeringsNow() if tracked, _ := snapshot(h); tracked != 0 { t.Fatalf("map not empty after first run: %d paths remain", tracked) } if got := store.callCount(); got != 1 { t.Fatalf("want 1 RecordPeering call after first run, got %d", got) } // Run 2: the same path again is recorded again (RecordPeering upserts). handle(h, now.Add(time.Second), testASNA, testASNB) h.ProcessPeeringsNow() if tracked, _ := snapshot(h); tracked != 0 { t.Fatalf("map not empty after second run: %d paths remain", tracked) } if got := store.callCount(); got != 2 { t.Fatalf("want 2 RecordPeering calls after second run, got %d", got) } } // TestPeeringHandlerStopWritesTheRemainingPaths hands a running handler one // path and stops it. The peering must be in the database when Stop returns. func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) { t.Parallel() store := &recordingStore{mockStore: newMockStore()} h := NewPeeringHandler(store, logger.New()) handle(h, time.Now().UTC(), testASNA, testASNB) h.Stop(t.Context()) if got := store.callCount(); got != 1 { t.Errorf("want 1 RecordPeering call when Stop returns, got %d", got) } } // TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths // and counts them, while a path already tracked is refreshed rather than // dropped. The handler's limit is small, and its timer does not fire. func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { t.Parallel() store := &recordingStore{mockStore: newMockStore()} h := newPeeringHandler(store, logger.New(), longerThanAnyTest, testMaxTrackedPaths) defer stopInTime(t, h) now := time.Now().UTC() // Fill the map to exactly its limit, including one real path key so the // "already tracked" branch can be exercised. The other paths are of one // AS, so they carry no peering. existing := pathKey(t, testASNA, testASNB) h.mu.Lock() h.asPaths[existing] = database.SeenTimes{First: now, Last: now} for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} } h.mu.Unlock() // A new path is dropped and counted because the map is full. handle(h, now.Add(time.Second), testASNA, testASNC) tracked, dropped := snapshot(h) if tracked != testMaxTrackedPaths { t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths, tracked) } if dropped != 1 { t.Fatalf("want dropped count 1, got %d", dropped) } // A path already tracked is refreshed, not dropped. refreshed := now.Add(2 * time.Second) handle(h, refreshed, testASNA, testASNB) tracked, dropped = snapshot(h) if tracked != testMaxTrackedPaths { t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths, tracked) } if dropped != 1 { t.Fatalf("want dropped count still 1 after refresh, got %d", dropped) } h.mu.Lock() gotSeen := h.asPaths[existing] h.mu.Unlock() if !gotSeen.Last.Equal(refreshed) { t.Fatalf("existing path last seen not refreshed: want %v, got %v", refreshed, gotSeen.Last) } } // TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler // drop one path, and runs the processing. The count of dropped paths is the // run's own, so it is 0 again after the run. The handler's limit is small, and // its timer does not fire. func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) { t.Parallel() h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest, testMaxTrackedPaths) defer stopInTime(t, h) now := time.Now().UTC() // Paths of one AS carry no peering, so the run records nothing. h.mu.Lock() for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} } h.mu.Unlock() handle(h, now, testASNA, testASNB) if _, dropped := snapshot(h); dropped != 1 { t.Fatalf("want dropped count 1 with the map full, got %d", dropped) } h.ProcessPeeringsNow() if _, dropped := snapshot(h); dropped != 0 { t.Errorf("want dropped count 0 after a run, got %d", dropped) } } // TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering handles two // updates whose path carries a peering the database does not hold yet, the // second 5 seconds after the first, in one run into a real database. The // peering is stored first seen at the time of the first update and last seen // at the time of the second. func TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering( t *testing.T, ) { t.Parallel() db := newTestDatabase(t) h := newTestHandler(db) first := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix) second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) h.HandleMessage(first) h.HandleMessage(second) h.ProcessPeeringsNow() peers, err := db.GetASPeers(testASNA) if err != nil { t.Fatalf("GetASPeers(%d): %v", testASNA, err) } if len(peers) != 1 || peers[0].ASN != testASNB { t.Fatalf("peers of AS%d = %+v, want AS%d alone", testASNA, peers, testASNB) } if !peers[0].FirstSeen.Equal(first.ParsedTimestamp) { t.Errorf("first_seen of peering AS%d-AS%d = %s, want %s", testASNA, testASNB, peers[0].FirstSeen, first.ParsedTimestamp) } if !peers[0].LastSeen.Equal(second.ParsedTimestamp) { t.Errorf("last_seen of peering AS%d-AS%d = %s, want %s", testASNA, testASNB, peers[0].LastSeen, second.ParsedTimestamp) } }