Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2ee58665ff |
No files matched your search
@@ -424,8 +424,9 @@ The lists:
|
||||
leased by spammers or cyber criminals, according to Spamhaus. From
|
||||
[Spamhaus DROP](https://www.spamhaus.org/blocklists/do-not-route-or-peer/),
|
||||
downloaded once a day.
|
||||
- `tor_exits`: Tor exit nodes, where traffic from the Tor network leaves. From
|
||||
the [Tor Project](https://check.torproject.org/torbulkexitlist), downloaded
|
||||
- `tor_exits`: the IPv4 and IPv6 addresses of Tor exit nodes, where traffic from
|
||||
the Tor network leaves. From the Tor Project's
|
||||
[Onionoo](https://metrics.torproject.org/onionoo.html) service, downloaded
|
||||
every hour.
|
||||
|
||||
No list is downloaded more often than its publisher updates it or allows. Each
|
||||
|
||||
@@ -53,9 +53,9 @@ problem the run finds.
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-11-10: the tests no longer fail when a busy host stalls for several
|
||||
seconds. Most such waits are now 15 seconds. The test of the stream's idle
|
||||
timeout runs on a fake clock (closes #240)
|
||||
- 2026-11-10: `tor_exits` comes from Onionoo, so it holds IPv6 exits and
|
||||
`is_tor` is true for them (closes
|
||||
https://git.eeqj.de/sneak/routewatch/issues/234).
|
||||
- 2026-11-10: each result of the IP lookup has a `flags` object with `is_tor`
|
||||
and `is_bogon`. The public IP lists are named `ip_lists` in the answer,
|
||||
`internal/iplist` in the code and `iplists` in the state directory. A list
|
||||
|
||||
@@ -12,7 +12,7 @@ import (
|
||||
|
||||
// parseTimeLimit is how long a test waits for parseCSV to return before it
|
||||
// fails.
|
||||
const parseTimeLimit = 15 * time.Second
|
||||
const parseTimeLimit = 2 * time.Second
|
||||
|
||||
// TestParseCSVReadsTodaysLayout parses an as.csv with the country code column
|
||||
// that the source added after the description. It expects each AS with its
|
||||
|
||||
@@ -72,34 +72,17 @@ const (
|
||||
// lockTestPrefix is the prefix the batch writes in the checkpoint tests
|
||||
// store.
|
||||
lockTestPrefix = "192.0.2.0/24"
|
||||
// longBusyTimeoutMs is the busy timeout setLongBusyTimeout gives the
|
||||
// writing connection. A checkpoint that waits for an open read takes this
|
||||
// long.
|
||||
longBusyTimeoutMs = 30000
|
||||
// readerCheckpointLimit is how long a checkpoint, and a batch write
|
||||
// waiting for it, may take while a reader is open: well under
|
||||
// longBusyTimeoutMs, with room for a stall on a busy host.
|
||||
readerCheckpointLimit = 15 * time.Second
|
||||
// waiting for it, may take while a reader is open: well under the busy
|
||||
// timeout, which a checkpoint that waits for the reader takes.
|
||||
readerCheckpointLimit = time.Second
|
||||
// truncateCheckpointLimit is how long a checkpoint of a WAL file over
|
||||
// sqliteWALSizeLimitBytes may take to give up while a reader is open: its
|
||||
// wait for the reader plus readerCheckpointLimit.
|
||||
// wait for the reader plus a second, well under the busy timeout.
|
||||
truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond +
|
||||
readerCheckpointLimit
|
||||
time.Second
|
||||
)
|
||||
|
||||
// setLongBusyTimeout gives the writing connection of db longBusyTimeoutMs, so
|
||||
// that a checkpoint that waits for an open read takes far longer than a stall
|
||||
// on a busy host.
|
||||
func setLongBusyTimeout(ctx context.Context, t *testing.T, db *Database) {
|
||||
t.Helper()
|
||||
|
||||
_, err := db.writer.ExecContext(ctx,
|
||||
fmt.Sprintf("PRAGMA busy_timeout=%d", longBusyTimeoutMs))
|
||||
if err != nil {
|
||||
t.Fatalf("failed to set the busy timeout: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
||||
// checks each one carries the per-connection settings from the DSN, plus the
|
||||
// process-wide heap limits, and sees the new file with auto_vacuum
|
||||
@@ -596,8 +579,6 @@ func TestCheckpointDoesNotWaitForReader(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ts := time.Now().UTC()
|
||||
|
||||
setLongBusyTimeout(ctx, t, db)
|
||||
|
||||
err = db.UpdatePrefixesBatch(map[string]time.Time{"198.51.100.0/24": ts})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to store the prefix the reader reads: %v", err)
|
||||
@@ -709,7 +690,6 @@ func TestCheckpointEmptiesWALOnceNoReadIsOpen(t *testing.T) {
|
||||
}
|
||||
|
||||
fillWALPastLimit(ctx, t, db)
|
||||
setLongBusyTimeout(ctx, t, db)
|
||||
|
||||
started := time.Now()
|
||||
|
||||
|
||||
@@ -29,6 +29,11 @@ const (
|
||||
// start with the network's first and last address. Lines that start
|
||||
// with # are comments.
|
||||
FormatDShield
|
||||
|
||||
// FormatOnionooJSON is one Onionoo details document. Its relays have
|
||||
// addresses in two fields: exit_addresses, which has IPv4 only, and
|
||||
// or_addresses, which has IPv4 and IPv6, each with a port.
|
||||
FormatOnionooJSON
|
||||
)
|
||||
|
||||
// List is one public IP list.
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -28,6 +29,20 @@ const (
|
||||
newListBody = "# A test list\n" + newListedAddr + "\n"
|
||||
)
|
||||
|
||||
// A small Onionoo details document: a relay with an exit address and IPv4 and
|
||||
// IPv6 relay addresses, one without exit addresses, and one with a list of
|
||||
// ports.
|
||||
const (
|
||||
onionooExitAddr = "192.0.2.10"
|
||||
onionooIPv6ExitAddr = "2001:db8::1"
|
||||
onionooBody = `{"version":"8.0","relays":[
|
||||
{"exit_addresses":["` + onionooExitAddr + `"],
|
||||
"or_addresses":["192.0.2.11:443","[` + onionooIPv6ExitAddr + `]:9001"]},
|
||||
{"or_addresses":["192.0.2.20:443"]},
|
||||
{"or_addresses":["192.0.2.30:80,443"]}
|
||||
],"bridges":[]}`
|
||||
)
|
||||
|
||||
// testETag is the ETag a publisher sends with a version of the test list.
|
||||
const testETag = `"v1"`
|
||||
|
||||
@@ -181,6 +196,16 @@ func TestLookupFindsListedAddresses(t *testing.T) {
|
||||
listed: []string{"192.0.2.0", "192.0.2.77"},
|
||||
unlisted: []string{"192.0.3.1", "2001:db8::1"},
|
||||
},
|
||||
{
|
||||
name: "Onionoo relays",
|
||||
format: FormatOnionooJSON,
|
||||
body: onionooBody,
|
||||
listed: []string{
|
||||
onionooExitAddr, "192.0.2.11", "192.0.2.20", "192.0.2.30",
|
||||
onionooIPv6ExitAddr,
|
||||
},
|
||||
unlisted: []string{"192.0.2.12", "2001:db8::2"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
@@ -195,26 +220,34 @@ func TestLookupFindsListedAddresses(t *testing.T) {
|
||||
t.Fatalf("download: %v", err)
|
||||
}
|
||||
|
||||
for _, addr := range tt.listed {
|
||||
listings := lookup(t, b, addr)
|
||||
if len(listings) != 1 || listings[0].Name != testListName ||
|
||||
listings[0].Description != testListDescription ||
|
||||
listings[0].Updated.IsZero() {
|
||||
t.Errorf("%s: listings %+v, want one for %s", addr,
|
||||
listings, testListName)
|
||||
}
|
||||
}
|
||||
|
||||
for _, addr := range tt.unlisted {
|
||||
listings := lookup(t, b, addr)
|
||||
if listings == nil || len(listings) != 0 {
|
||||
t.Errorf("%s: listings %#v, want an empty list", addr, listings)
|
||||
}
|
||||
}
|
||||
checkLookups(t, b, tt.listed, tt.unlisted)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// checkLookups checks that each of listed has one listing that names the test
|
||||
// list, and each of unlisted has an empty list of listings.
|
||||
func checkLookups(t *testing.T, b *IPLists, listed, unlisted []string) {
|
||||
t.Helper()
|
||||
|
||||
for _, addr := range listed {
|
||||
listings := lookup(t, b, addr)
|
||||
if len(listings) != 1 || listings[0].Name != testListName ||
|
||||
listings[0].Description != testListDescription ||
|
||||
listings[0].Updated.IsZero() {
|
||||
t.Errorf("%s: listings %+v, want one for %s", addr,
|
||||
listings, testListName)
|
||||
}
|
||||
}
|
||||
|
||||
for _, addr := range unlisted {
|
||||
listings := lookup(t, b, addr)
|
||||
if listings == nil || len(listings) != 0 {
|
||||
t.Errorf("%s: listings %#v, want an empty list", addr, listings)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestFailedDownloadKeepsThePreviousCopy downloads a good version of a list,
|
||||
// then one that fails. The good copy stays in memory and on disk, and no
|
||||
// part of the failed download is left beside it.
|
||||
@@ -263,6 +296,58 @@ func TestFailedDownloadKeepsThePreviousCopy(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileThatIsNotAnOnionooDocumentKeepsThePreviousCopy downloads a good
|
||||
// Onionoo document, then files that are not one. Each fails the parse, and
|
||||
// the good copy stays.
|
||||
func TestFileThatIsNotAnOnionooDocumentKeepsThePreviousCopy(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
body string
|
||||
}{
|
||||
{"not JSON", "<!DOCTYPE html>\n<html></html>\n"},
|
||||
{"other JSON", `{"status":"ok","data":{"relays_count":1}}`},
|
||||
{"JSON array", `[{"or_addresses":["192.0.2.1:443"]}]`},
|
||||
{"no relays", `{"relays":[],"bridges":[]}`},
|
||||
{"relays without addresses", `{"relays":[{"nickname":"example"}]}`},
|
||||
{"relay address without a port", `{"relays":[{"or_addresses":["192.0.2.1"]}]}`},
|
||||
{"exit address that is not one", `{"relays":[{"exit_addresses":["example"]}]}`},
|
||||
{"text after the document", `{"relays":[{"or_addresses":["192.0.2.1:1"]}]} x`},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
p, url := newPublisher(t, onionooBody)
|
||||
b := newTestIPLists(dir, testList(url, FormatOnionooJSON))
|
||||
|
||||
err := b.download(t.Context(), b.lists[0])
|
||||
if err != nil {
|
||||
t.Fatalf("first download: %v", err)
|
||||
}
|
||||
|
||||
p.answer(http.StatusOK, tt.body)
|
||||
|
||||
err = b.download(t.Context(), b.lists[0])
|
||||
if err == nil {
|
||||
t.Fatal("second download returned no error")
|
||||
}
|
||||
|
||||
if len(lookup(t, b, onionooExitAddr)) != 1 {
|
||||
t.Errorf("%s is no longer listed", onionooExitAddr)
|
||||
}
|
||||
|
||||
onDisk, err := fs.ReadFile(os.DirFS(dir), testListName)
|
||||
if err != nil || string(onDisk) != onionooBody {
|
||||
t.Errorf("copy on disk %q (%v), want the good version", onDisk, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// checkGoodCopyOnDisk checks that dir holds the good version of the test list
|
||||
// and no part of a failed download.
|
||||
func checkGoodCopyOnDisk(t *testing.T, dir string) {
|
||||
@@ -380,13 +465,9 @@ func TestRestartLoadsCopiesFromDisk(t *testing.T) {
|
||||
second.Start(t.Context())
|
||||
second.Stop(t.Context())
|
||||
|
||||
// The download takes its time after writing the file, and the restart
|
||||
// reads the file's time. A busy host can put seconds between the two.
|
||||
const maxUpdatedDifference = 15 * time.Second
|
||||
|
||||
listings := lookup(t, second, listedAddr)
|
||||
if len(listings) != 1 ||
|
||||
listings[0].Updated.Sub(downloaded).Abs() > maxUpdatedDifference {
|
||||
listings[0].Updated.Sub(downloaded).Abs() > time.Second {
|
||||
t.Errorf("after the restart, listings %+v, want one updated at %s",
|
||||
listings, downloaded)
|
||||
}
|
||||
@@ -559,6 +640,33 @@ func TestStopCancelsADownloadInProgress(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestTorExitsListHoldsIPv6Exits downloads an Onionoo document as the public
|
||||
// Tor exit list, which is what an IPv6 address's is_tor answer comes from.
|
||||
func TestTorExitsListHoldsIPv6Exits(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
index := slices.IndexFunc(Lists(), func(list List) bool {
|
||||
return list.Name == TorExitsList
|
||||
})
|
||||
if index < 0 {
|
||||
t.Fatalf("no list named %s", TorExitsList)
|
||||
}
|
||||
|
||||
list := Lists()[index]
|
||||
_, list.URL = newPublisher(t, onionooBody)
|
||||
b := newTestIPLists(t.TempDir(), list)
|
||||
|
||||
err := b.download(t.Context(), list)
|
||||
if err != nil {
|
||||
t.Fatalf("download: %v", err)
|
||||
}
|
||||
|
||||
listings := lookup(t, b, onionooIPv6ExitAddr)
|
||||
if len(listings) != 1 || listings[0].Name != TorExitsList {
|
||||
t.Errorf("listings %+v, want one for %s", listings, TorExitsList)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPublicListsAreComplete checks that every public list has what a
|
||||
// download and a lookup answer need, and a unique name.
|
||||
func TestPublicListsAreComplete(t *testing.T) {
|
||||
|
||||
@@ -28,8 +28,9 @@ const (
|
||||
// https://www.spamhaus.org/blocklists/do-not-route-or-peer/
|
||||
spamhausDROPInterval = 24 * time.Hour
|
||||
|
||||
// Tor publishes a new exit list every hour and sets no download limit:
|
||||
// https://collector.torproject.org/recent/exit-lists/
|
||||
// Onionoo's relay data follows Tor's hourly network consensus, and
|
||||
// Onionoo sets no download limit:
|
||||
// https://metrics.torproject.org/onionoo.html
|
||||
torExitsInterval = time.Hour
|
||||
)
|
||||
|
||||
@@ -88,9 +89,10 @@ func Lists() []List {
|
||||
{
|
||||
Name: TorExitsList,
|
||||
Description: "A Tor exit node, where traffic from the Tor network leaves",
|
||||
URL: "https://check.torproject.org/torbulkexitlist",
|
||||
Interval: torExitsInterval,
|
||||
Format: FormatLines,
|
||||
URL: "https://onionoo.torproject.org/details?" +
|
||||
"flag=Exit&running=true&fields=exit_addresses,or_addresses",
|
||||
Interval: torExitsInterval,
|
||||
Format: FormatOnionooJSON,
|
||||
},
|
||||
}
|
||||
}
|
||||
+99
-16
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/netip"
|
||||
"strings"
|
||||
|
||||
@@ -32,12 +33,32 @@ var (
|
||||
errUnknownFormat = errors.New("unknown list format")
|
||||
errNoNetwork = errors.New("no network on the line")
|
||||
errBadRange = errors.New("first and last address make no range")
|
||||
errNoAddresses = errors.New("no relay addresses in the document")
|
||||
)
|
||||
|
||||
// parse reads a list file written in format and returns the addresses it
|
||||
// holds. A line it cannot read fails the whole file, so a file that is not
|
||||
// holds. Anything it cannot read fails the whole file, so a file that is not
|
||||
// the list, such as an error page, never replaces a good copy.
|
||||
func parse(r io.Reader, format Format) (*netipx.IPSet, error) {
|
||||
switch format {
|
||||
case FormatLines:
|
||||
return parseLines(r, addAddressOrNetwork)
|
||||
case FormatSpamhausJSON:
|
||||
return parseLines(r, addSpamhausNetwork)
|
||||
case FormatDShield:
|
||||
return parseLines(r, addDShieldNetwork)
|
||||
case FormatOnionooJSON:
|
||||
return parseOnionoo(r)
|
||||
default:
|
||||
return nil, errUnknownFormat
|
||||
}
|
||||
}
|
||||
|
||||
// parseLines reads a list file line by line, skipping blank lines and
|
||||
// comments, and calls addLine with each other line.
|
||||
func parseLines(
|
||||
r io.Reader, addLine func(*netipx.IPSetBuilder, string) error,
|
||||
) (*netipx.IPSet, error) {
|
||||
var builder netipx.IPSetBuilder
|
||||
|
||||
scanner := bufio.NewScanner(r)
|
||||
@@ -48,7 +69,7 @@ func parse(r io.Reader, format Format) (*netipx.IPSet, error) {
|
||||
continue
|
||||
}
|
||||
|
||||
err := addLine(&builder, line, format)
|
||||
err := addLine(&builder, line)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("line %d: %w", lineNumber, err)
|
||||
}
|
||||
@@ -62,20 +83,6 @@ func parse(r io.Reader, format Format) (*netipx.IPSet, error) {
|
||||
return builder.IPSet()
|
||||
}
|
||||
|
||||
// addLine adds the address or network on one line of a list file to builder.
|
||||
func addLine(builder *netipx.IPSetBuilder, line string, format Format) error {
|
||||
switch format {
|
||||
case FormatLines:
|
||||
return addAddressOrNetwork(builder, line)
|
||||
case FormatSpamhausJSON:
|
||||
return addSpamhausNetwork(builder, line)
|
||||
case FormatDShield:
|
||||
return addDShieldNetwork(builder, line)
|
||||
default:
|
||||
return errUnknownFormat
|
||||
}
|
||||
}
|
||||
|
||||
// addAddressOrNetwork adds an address, such as 192.0.2.1, or a network, such
|
||||
// as 192.0.2.0/24, to builder.
|
||||
func addAddressOrNetwork(builder *netipx.IPSetBuilder, text string) error {
|
||||
@@ -156,3 +163,79 @@ func addDShieldNetwork(builder *netipx.IPSetBuilder, line string) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// parseOnionoo returns every address in the exit_addresses and or_addresses
|
||||
// fields of the relays in an Onionoo details document. A document with no
|
||||
// addresses fails, because the Tor network always has relays.
|
||||
func parseOnionoo(r io.Reader) (*netipx.IPSet, error) {
|
||||
data, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
//nolint:tagliatelle // Onionoo's own field names
|
||||
var details struct {
|
||||
Relays []struct {
|
||||
ExitAddresses []string `json:"exit_addresses"`
|
||||
ORAddresses []string `json:"or_addresses"`
|
||||
} `json:"relays"`
|
||||
}
|
||||
|
||||
err = json.Unmarshal(data, &details)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var builder netipx.IPSetBuilder
|
||||
|
||||
for _, relay := range details.Relays {
|
||||
for _, text := range relay.ExitAddresses {
|
||||
err = addAddress(&builder, text)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
for _, text := range relay.ORAddresses {
|
||||
err = addAddressWithPort(&builder, text)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
addresses, err := builder.IPSet()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(addresses.Ranges()) == 0 {
|
||||
return nil, errNoAddresses
|
||||
}
|
||||
|
||||
return addresses, nil
|
||||
}
|
||||
|
||||
// addAddress adds an address, such as 192.0.2.1, to builder.
|
||||
func addAddress(builder *netipx.IPSetBuilder, text string) error {
|
||||
addr, err := netip.ParseAddr(text)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
builder.Add(addr)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// addAddressWithPort adds the address of text, such as 192.0.2.1:443 or
|
||||
// [2001:db8::1]:443, to builder. Onionoo allows a list of ports after the
|
||||
// address, so the part after the address is not read.
|
||||
func addAddressWithPort(builder *netipx.IPSetBuilder, text string) error {
|
||||
host, _, err := net.SplitHostPort(text)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return addAddress(builder, host)
|
||||
}
|
||||
@@ -17,7 +17,6 @@ import (
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
@@ -174,9 +173,6 @@ type tableStats struct {
|
||||
// carried 203.0.113.0/24, the two with kept routes withdrew it, so its origin
|
||||
// keeps no route. The oldest and newest routes are those of the other two
|
||||
// prefixes.
|
||||
//
|
||||
// The requests run in a synctest bubble, whose clock is fake, so a stall on
|
||||
// the host does not use up the handlers' 4-second limit.
|
||||
func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -222,11 +218,7 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
||||
|
||||
for name, handler := range handlers {
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
handler(rec,
|
||||
httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
|
||||
})
|
||||
handler(rec, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("%s: status %d, want %d; body %s",
|
||||
@@ -261,20 +253,15 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
||||
// TestStatsAnswerCarriesChurnPerMinute checks that the stream object of the
|
||||
// /api/v1/stats answer holds route_churn_per_min as a whole number, and no
|
||||
// longer route_churn_per_sec, the moving average it replaces
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/160). The request runs in a
|
||||
// synctest bubble, whose clock is fake, so a stall on the host does not use up
|
||||
// the handler's 4-second limit.
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/160).
|
||||
func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
s.handleStats()(rec,
|
||||
httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil))
|
||||
})
|
||||
s.handleStats()(rec,
|
||||
httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
||||
@@ -322,26 +309,18 @@ func (heldBackHandler) HeldBackCount() int { return heldBackMess
|
||||
|
||||
// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
|
||||
// that holds back messages. Its figures in /api/v1/stats, which the status
|
||||
// page shows, must give how many. The request runs in a synctest bubble, whose
|
||||
// clock is fake, so a stall on the host does not use up the handler's 4-second
|
||||
// limit.
|
||||
// page shows, must give how many.
|
||||
func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
s.streamer.RegisterHandler(heldBackHandler{})
|
||||
|
||||
var answer []byte
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
answer = get(t, s, statsPath).Body.Bytes()
|
||||
})
|
||||
|
||||
var body struct {
|
||||
Data StatsResponse `json:"data"`
|
||||
}
|
||||
|
||||
err := json.Unmarshal(answer, &body)
|
||||
err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
|
||||
if err != nil {
|
||||
t.Fatalf("decoding the answer: %v", err)
|
||||
}
|
||||
@@ -385,9 +364,7 @@ type unitStats struct {
|
||||
// names. They used to be strings with the unit in them, such as "1h18m34s" or
|
||||
// "1.1 GB" (https://git.eeqj.de/sneak/routewatch/issues/199). The stream has
|
||||
// never connected, so the uptime is 0. The memory use in /api/v1/stats is the
|
||||
// same number of bytes as its gc.heap_alloc_bytes. The requests run in a
|
||||
// synctest bubble, whose clock is fake, so a stall on the host does not use up
|
||||
// the handlers' 4-second limit.
|
||||
// same number of bytes as its gc.heap_alloc_bytes.
|
||||
func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -397,11 +374,7 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
|
||||
s.SetASNFetcher(fixedIntervalASNFetcher{interval: fetchInterval})
|
||||
|
||||
for _, path := range []string{statusJSONPath, statsPath} {
|
||||
var answer []byte
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
answer = get(t, s, path).Body.Bytes()
|
||||
})
|
||||
answer := get(t, s, path).Body.Bytes()
|
||||
|
||||
var body struct {
|
||||
Data unitStats `json:"data"`
|
||||
@@ -442,89 +415,82 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
|
||||
// connection_duration_seconds of /api/v1/stats is null while the stream is not
|
||||
// connected, and a whole number of seconds once it is. It used to be the
|
||||
// string "disconnected", or a Go duration such as "1h18m34s"
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/199). The test runs in a
|
||||
// synctest bubble, whose clock is fake, so a stall on the host does not use up
|
||||
// the handler's 4-second limit.
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/199).
|
||||
func TestStatsAnswerGivesConnectionDurationInSeconds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
connectionDuration := func() json.RawMessage {
|
||||
answer := get(t, s, statsPath).Body.Bytes()
|
||||
connectionDuration := func() json.RawMessage {
|
||||
answer := get(t, s, statsPath).Body.Bytes()
|
||||
|
||||
var body struct {
|
||||
Data map[string]json.RawMessage `json:"data"`
|
||||
}
|
||||
|
||||
err := json.Unmarshal(answer, &body)
|
||||
if err != nil {
|
||||
t.Fatalf("decoding the answer: %v; answer %s", err, answer)
|
||||
}
|
||||
|
||||
duration, ok := body.Data["connection_duration_seconds"]
|
||||
if !ok {
|
||||
t.Fatalf("no connection_duration_seconds: %s", answer)
|
||||
}
|
||||
|
||||
return duration
|
||||
var body struct {
|
||||
Data map[string]json.RawMessage `json:"data"`
|
||||
}
|
||||
|
||||
disconnected := connectionDuration()
|
||||
if string(disconnected) != "null" {
|
||||
t.Errorf("disconnected: connection_duration_seconds = %s, want null",
|
||||
disconnected)
|
||||
err := json.Unmarshal(answer, &body)
|
||||
if err != nil {
|
||||
t.Fatalf("decoding the answer: %v; answer %s", err, answer)
|
||||
}
|
||||
|
||||
s.streamer.GetMetricsTracker().SetConnected(true)
|
||||
|
||||
var seconds int64
|
||||
|
||||
connected := connectionDuration()
|
||||
|
||||
err := json.Unmarshal(connected, &seconds)
|
||||
if err != nil || seconds != 0 {
|
||||
t.Errorf("just connected: connection_duration_seconds = %s, want 0",
|
||||
connected)
|
||||
duration, ok := body.Data["connection_duration_seconds"]
|
||||
if !ok {
|
||||
t.Fatalf("no connection_duration_seconds: %s", answer)
|
||||
}
|
||||
})
|
||||
|
||||
return duration
|
||||
}
|
||||
|
||||
disconnected := connectionDuration()
|
||||
if string(disconnected) != "null" {
|
||||
t.Errorf("disconnected: connection_duration_seconds = %s, want null",
|
||||
disconnected)
|
||||
}
|
||||
|
||||
s.streamer.GetMetricsTracker().SetConnected(true)
|
||||
|
||||
var seconds int64
|
||||
|
||||
connected := connectionDuration()
|
||||
|
||||
err := json.Unmarshal(connected, &seconds)
|
||||
if err != nil || seconds != 0 {
|
||||
t.Errorf("just connected: connection_duration_seconds = %s, want 0",
|
||||
connected)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamDurationsCountSeconds checks that the uptime and the connection
|
||||
// duration of a stream that connected 90 seconds ago are both 90. A stream
|
||||
// that has never connected has an uptime of 0, and one that is not connected
|
||||
// has no connection duration. The test runs in a synctest bubble, whose clock
|
||||
// is fake, so a stall on the host does not add to the 90 seconds.
|
||||
// has no connection duration.
|
||||
func TestStreamDurationsCountSeconds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
const connectedFor = 90 * time.Second
|
||||
const connectedFor = 90 * time.Second
|
||||
|
||||
connectedSince := time.Now().Add(-connectedFor)
|
||||
want := int64(connectedFor / time.Second)
|
||||
connectedSince := time.Now().Add(-connectedFor)
|
||||
want := int64(connectedFor / time.Second)
|
||||
|
||||
if got := uptimeSeconds(connectedSince); got != want {
|
||||
t.Errorf("uptimeSeconds = %d, want %d", got, want)
|
||||
}
|
||||
if got := uptimeSeconds(connectedSince); got != want {
|
||||
t.Errorf("uptimeSeconds = %d, want %d", got, want)
|
||||
}
|
||||
|
||||
if got := uptimeSeconds(time.Time{}); got != 0 {
|
||||
t.Errorf("uptimeSeconds before any connection = %d, want 0", got)
|
||||
}
|
||||
if got := uptimeSeconds(time.Time{}); got != 0 {
|
||||
t.Errorf("uptimeSeconds before any connection = %d, want 0", got)
|
||||
}
|
||||
|
||||
switch got := connectionDurationSeconds(true, connectedSince); {
|
||||
case got == nil:
|
||||
t.Errorf("connectionDurationSeconds = nil, want %d", want)
|
||||
case *got != want:
|
||||
t.Errorf("connectionDurationSeconds = %d, want %d", *got, want)
|
||||
}
|
||||
switch got := connectionDurationSeconds(true, connectedSince); {
|
||||
case got == nil:
|
||||
t.Errorf("connectionDurationSeconds = nil, want %d", want)
|
||||
case *got != want:
|
||||
t.Errorf("connectionDurationSeconds = %d, want %d", *got, want)
|
||||
}
|
||||
|
||||
if got := connectionDurationSeconds(false, connectedSince); got != nil {
|
||||
t.Errorf("connectionDurationSeconds while not connected = %d, want nil",
|
||||
*got)
|
||||
}
|
||||
})
|
||||
if got := connectionDurationSeconds(false, connectedSince); got != nil {
|
||||
t.Errorf("connectionDurationSeconds while not connected = %d, want nil",
|
||||
*got)
|
||||
}
|
||||
}
|
||||
|
||||
// newTestTable returns a routing table that holds a route for testPrefixV4,
|
||||
@@ -1168,9 +1134,7 @@ func (failingDB) GetPrefixFirstSeen(
|
||||
// with the driver's error text. It also sends a prefix that does not parse.
|
||||
// No answer may carry the driver's or the prefix parser's text. Before the fix
|
||||
// the answers sent it to the client
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/108). The requests run in a
|
||||
// synctest bubble, whose clock is fake, so a stall on the host does not use up
|
||||
// the stats handlers' 4-second limit.
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/108).
|
||||
func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -1191,11 +1155,8 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, tt.path, nil))
|
||||
})
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, tt.path, nil))
|
||||
|
||||
if rec.Code != tt.wantCode {
|
||||
t.Errorf("%s: status %d, want %d; body %s",
|
||||
@@ -1436,11 +1397,8 @@ func TestIPLookupKeepsRouteWhenPTRLookupDoesNotAnswer(t *testing.T) {
|
||||
s := newServerWithTestRoutes(t)
|
||||
s.resolver = &fakeResolver{ptrNeverAnswers: true}
|
||||
|
||||
// The request's time limit passes well after the PTR lookup's, so that a
|
||||
// stall on a busy host does not use up the difference.
|
||||
const requestTimeLimit = ptrLookupTimeout + 15*time.Second
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), requestTimeLimit)
|
||||
// The request's time limit passes after the PTR lookup's.
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 2*ptrLookupTimeout)
|
||||
defer cancel()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
@@ -2329,13 +2287,9 @@ func settledGoroutineCount() int {
|
||||
return prev
|
||||
}
|
||||
|
||||
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
|
||||
// target or below.
|
||||
// waitForGoroutines waits until the goroutine count drops to target or below.
|
||||
func waitForGoroutines(target int) bool {
|
||||
const maxWait = 15 * time.Second
|
||||
|
||||
deadline := time.Now().Add(maxWait)
|
||||
for time.Now().Before(deadline) {
|
||||
for range 100 {
|
||||
if runtime.NumGoroutine() <= target {
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -76,9 +76,7 @@ func asLink(asn int) string {
|
||||
}
|
||||
|
||||
// TestHomePageCardsShowTheTableFigures fills a routing table and the asns
|
||||
// table, works out the cards, and checks that each card shows its figure. The
|
||||
// cards are worked out in a synctest bubble, whose clock is fake, so a stall on
|
||||
// the host does not use up the 4-second limit on their database read.
|
||||
// table, works out the cards, and checks that each card shows its figure.
|
||||
func TestHomePageCardsShowTheTableFigures(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -97,9 +95,7 @@ func TestHomePageCardsShowTheTableFigures(t *testing.T) {
|
||||
t.Fatalf("GetOrCreateASNBatch: %v", err)
|
||||
}
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
s.refreshHomeCards(t.Context())
|
||||
})
|
||||
s.refreshHomeCards(t.Context())
|
||||
|
||||
page := get(t, s, "/").Body.String()
|
||||
|
||||
@@ -283,7 +279,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
|
||||
|
||||
const (
|
||||
interval = 10 * time.Millisecond
|
||||
maxWait = 15 * time.Second
|
||||
maxWait = 5 * time.Second
|
||||
)
|
||||
|
||||
table := emptyTable()
|
||||
@@ -325,7 +321,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
|
||||
func TestStartedServerWorksOutTheCards(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxWait = 15 * time.Second
|
||||
const maxWait = 5 * time.Second
|
||||
|
||||
s := newServerWithTable(t, newCardsTable())
|
||||
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
@@ -164,10 +163,6 @@ func TestJSONAnswersPutNoUnitInAString(t *testing.T) {
|
||||
// router and the middleware write. Before the fix a panic in a handler, and a
|
||||
// method /status.json or the health check does not take, answered with an
|
||||
// empty body (https://git.eeqj.de/sneak/routewatch/issues/201).
|
||||
//
|
||||
// The requests run in a synctest bubble, whose clock is fake, so a stall on
|
||||
// the host does not use up the 2-second limit of the health check or the
|
||||
// 4-second limit of the stats handlers.
|
||||
func TestJSONAnswersCarryMeta(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -186,10 +181,6 @@ func TestJSONAnswersCarryMeta(t *testing.T) {
|
||||
s.router.Get(panicsPath, JSONValidationMiddleware(http.HandlerFunc(
|
||||
func(http.ResponseWriter, *http.Request) { panic(errDiskIO) })).ServeHTTP)
|
||||
|
||||
failing := New(failingDB{}, newTestTable(),
|
||||
streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||
failing.resolver = &fakeResolver{}
|
||||
|
||||
tests := []struct {
|
||||
method string
|
||||
path string
|
||||
@@ -225,40 +216,42 @@ func TestJSONAnswersCarryMeta(t *testing.T) {
|
||||
{http.MethodGet, panicsPath, http.StatusInternalServerError},
|
||||
}
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
for _, tt := range tests {
|
||||
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
|
||||
t.Context(), tt.method, tt.path, nil), tt.wantCode)
|
||||
}
|
||||
for _, tt := range tests {
|
||||
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
|
||||
t.Context(), tt.method, tt.path, nil), tt.wantCode)
|
||||
}
|
||||
|
||||
// A request whose time limit has already passed answers 504. The
|
||||
// handlers write the answers of the first five routes, and the JSON
|
||||
// check on the route writes the last.
|
||||
pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
// A request whose time limit has already passed answers 504. The handlers
|
||||
// write the answers of the first five routes, and the JSON check on the
|
||||
// route writes the last.
|
||||
pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
for _, path := range []string{
|
||||
statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4,
|
||||
"/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath,
|
||||
} {
|
||||
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
|
||||
pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout)
|
||||
}
|
||||
for _, path := range []string{
|
||||
statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4,
|
||||
"/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath,
|
||||
} {
|
||||
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
|
||||
pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout)
|
||||
}
|
||||
|
||||
// With a store whose reads fail, the health check answers 503, and
|
||||
// the handlers of the other routes 500.
|
||||
// With a store whose reads fail, the health check answers 503, and the
|
||||
// handlers of the other routes 500.
|
||||
failing := New(failingDB{}, newTestTable(),
|
||||
streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||
failing.resolver = &fakeResolver{}
|
||||
|
||||
checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, healthCheckPath, nil),
|
||||
http.StatusServiceUnavailable)
|
||||
|
||||
for _, path := range []string{
|
||||
statusJSONPath, statsPath, asJSONPath,
|
||||
"/api/v1/ip/" + addrInTestPrefixV4, "/ip/" + addrInTestPrefixV4,
|
||||
} {
|
||||
checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, healthCheckPath, nil),
|
||||
http.StatusServiceUnavailable)
|
||||
|
||||
for _, path := range []string{
|
||||
statusJSONPath, statsPath, asJSONPath,
|
||||
"/api/v1/ip/" + addrInTestPrefixV4, "/ip/" + addrInTestPrefixV4,
|
||||
} {
|
||||
checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError)
|
||||
}
|
||||
})
|
||||
t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError)
|
||||
}
|
||||
}
|
||||
|
||||
// checkAnswerCarriesMeta serves req through the server's router, and checks
|
||||
|
||||
@@ -5,7 +5,6 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
)
|
||||
|
||||
// TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the
|
||||
@@ -13,23 +12,15 @@ import (
|
||||
// lookup of the caller's own address, are in the list because chi's GetHead
|
||||
// does not reach them. Before the fix every route answered HEAD with 405
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/111).
|
||||
//
|
||||
// The requests run in a synctest bubble, whose clock is fake, so a stall on
|
||||
// the host does not use up the stats handlers' 4-second limit. The resolver
|
||||
// is fake because no time limit would cut off a real DNS lookup in a bubble.
|
||||
func TestRoutesAnswerHEAD(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newServerWithTestRoutes(t)
|
||||
s.resolver = &fakeResolver{}
|
||||
|
||||
for _, path := range []string{"/", statusJSONPath, statsPath, "/ip", "/ip/"} {
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodHead, path, nil))
|
||||
})
|
||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodHead, path, nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("HEAD %s: status %d, want %d; body %s",
|
||||
|
||||
@@ -22,10 +22,10 @@ func TestStopClosesTheConnectionOfARequestStillRunning(t *testing.T) {
|
||||
stopDeadline = 100 * time.Millisecond
|
||||
// waitTimeout is how long the test waits for the request to reach
|
||||
// the handler, and then for the client to see its connection closed.
|
||||
waitTimeout = 15 * time.Second
|
||||
waitTimeout = 2 * time.Second
|
||||
// readHeaderTimeout bounds the test server's read of the request
|
||||
// header, as every http.Server must.
|
||||
readHeaderTimeout = waitTimeout
|
||||
readHeaderTimeout = time.Second
|
||||
)
|
||||
|
||||
release := make(chan struct{})
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
@@ -399,7 +398,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
|
||||
const (
|
||||
held = 2
|
||||
lines = held + 1
|
||||
maxWait = 15 * time.Second
|
||||
maxWait = 5 * time.Second
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
@@ -479,15 +478,15 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
|
||||
// TestStopReturnsByItsDeadline stops a streamer whose handler has a full queue
|
||||
// and a STATE message held for it, and a worker that stays busy until the test
|
||||
// ends, as one does whose handler waits for a database write. Stop must return
|
||||
// within its deadline plus a margin instead of waiting for room on the queue
|
||||
// and for the worker, and log as not handed over the held message and every
|
||||
// message on the queue.
|
||||
// within its deadline plus a small margin instead of waiting for room on the
|
||||
// queue and for the worker, and log as not handed over the held message and
|
||||
// every message on the queue.
|
||||
func TestStopReturnsByItsDeadline(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
deadline = 100 * time.Millisecond
|
||||
margin = 15 * time.Second
|
||||
margin = 2 * time.Second
|
||||
)
|
||||
|
||||
var logs bytes.Buffer
|
||||
@@ -611,34 +610,10 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
|
||||
checkReconnect(t, neverAnswer, idleTimeout)
|
||||
}
|
||||
|
||||
// pipeTransport answers every request with a response whose body write fills.
|
||||
// The body fails when the request's context ends, as a connection's does.
|
||||
// write gets that context, so that it can end when the request does.
|
||||
type pipeTransport struct {
|
||||
write func(context.Context, io.Writer)
|
||||
}
|
||||
|
||||
func (p pipeTransport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
body, w := io.Pipe()
|
||||
|
||||
context.AfterFunc(req.Context(), func() {
|
||||
_ = w.CloseWithError(req.Context().Err())
|
||||
})
|
||||
|
||||
go func(ctx context.Context) {
|
||||
p.write(ctx, w)
|
||||
|
||||
_ = w.Close()
|
||||
}(req.Context())
|
||||
|
||||
return &http.Response{StatusCode: http.StatusOK, Body: body}, nil
|
||||
}
|
||||
|
||||
// TestStreamNotDroppedWhileLinesArrive has a response send lines well within
|
||||
// the idle timeout for several times the timeout, then end. Each line must
|
||||
// start the idle timeout over, so the stream lasts until the response ends
|
||||
// instead of being dropped as idle. It runs in a synctest bubble, whose clock
|
||||
// is fake, so a stall on the host does not use up the timeout.
|
||||
// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
|
||||
// the idle timeout for several times the timeout, then end the response. Each
|
||||
// line must start the idle timeout over, so the stream lasts until the server
|
||||
// ends it instead of being dropped as idle.
|
||||
func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -648,35 +623,26 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
|
||||
sendFor = 5 * idleTimeout
|
||||
)
|
||||
|
||||
// The streamer is made outside the bubble. Its metrics start a goroutine
|
||||
// that never returns, and a bubble waits for each of its goroutines to
|
||||
// return.
|
||||
s := New(logger.New(), metrics.New())
|
||||
s.idleTimeout = idleTimeout
|
||||
s.client = &http.Client{Transport: pipeTransport{
|
||||
write: func(ctx context.Context, w io.Writer) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
deadline := time.Now().Add(sendFor)
|
||||
for time.Now().Before(deadline) {
|
||||
_, _ = io.WriteString(w, updateLine)
|
||||
|
||||
// The wait ends with the stream. A bubble's clock stops when
|
||||
// the test function returns, and a goroutine then still
|
||||
// asleep panics the test binary.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(lineInterval):
|
||||
}
|
||||
flush(w)
|
||||
time.Sleep(lineInterval)
|
||||
}
|
||||
},
|
||||
}}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
err := s.stream(context.Background())
|
||||
if !errors.Is(err, errStreamEnded) {
|
||||
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
|
||||
}
|
||||
})
|
||||
s := New(logger.New(), metrics.New())
|
||||
s.url = srv.URL
|
||||
s.idleTimeout = idleTimeout
|
||||
|
||||
err := s.stream(context.Background())
|
||||
if !errors.Is(err, errStreamEnded) {
|
||||
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamReadsLongLine has the server send one UPDATE line of about
|
||||
@@ -740,7 +706,7 @@ func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
|
||||
const (
|
||||
lines = 5
|
||||
wantDropped = lines - 2
|
||||
maxWait = 15 * time.Second
|
||||
maxWait = 5 * time.Second
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
@@ -859,7 +825,7 @@ func checkReconnect(
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
const maxWait = 15 * time.Second
|
||||
const maxWait = 5 * time.Second
|
||||
|
||||
tracker := metrics.New()
|
||||
connectedAtReconnect := make(chan bool, 1)
|
||||
@@ -930,13 +896,9 @@ func settledGoroutineCount() int {
|
||||
return prev
|
||||
}
|
||||
|
||||
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
|
||||
// target or below.
|
||||
// waitForGoroutines waits until the goroutine count drops to target or below.
|
||||
func waitForGoroutines(target int) bool {
|
||||
const maxWait = 15 * time.Second
|
||||
|
||||
deadline := time.Now().Add(maxWait)
|
||||
for time.Now().Before(deadline) {
|
||||
for range 100 {
|
||||
if runtime.NumGoroutine() <= target {
|
||||
return true
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user