▽

Source file src/github.com/cybertec-postgresql/pgwatch/v6/internal/sources/conn_test.go

Documentation: github.com/cybertec-postgresql/pgwatch/v6/internal/sources

     1  package sources_test
     2  
     3  import (
     4  	"context"
     5  	"fmt"
     6  	"net/http"
     7  	"net/http/httptest"
     8  	"net/url"
     9  	"strings"
    10  	"sync"
    11  	"testing"
    12  	"testing/synctest"
    13  	"time"
    14  
    15  	"github.com/jackc/pgx/v5"
    16  	"github.com/jackc/pgx/v5/pgconn"
    17  	"github.com/jackc/pgx/v5/pgxpool"
    18  	"github.com/pashagolub/pgxmock/v5"
    19  	"github.com/sirupsen/logrus"
    20  	"github.com/stretchr/testify/assert"
    21  	"github.com/stretchr/testify/require"
    22  
    23  	"github.com/cybertec-postgresql/pgwatch/v6/internal/db"
    24  	"github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
    25  	"github.com/cybertec-postgresql/pgwatch/v6/internal/sources"
    26  	"github.com/cybertec-postgresql/pgwatch/v6/internal/testutil"
    27  )
    28  
    29  func TestSourceConn_Connect(t *testing.T) {
    30  
    31  	t.Run("failed config parsing", func(t *testing.T) {
    32  		md := &sources.DbConn{}
    33  		md.ConnStr = "invalid connection string"
    34  		err := md.Connect(ctx, sources.CmdOpts{})
    35  		assert.Error(t, err)
    36  	})
    37  
    38  	t.Run("failed connection", func(t *testing.T) {
    39  		md := &sources.DbConn{}
    40  		sources.NewConnWithConfig = func(_ context.Context, _ *pgxpool.Config, _ ...db.ConnConfigCallback) (db.PgxPoolIface, error) {
    41  			return nil, assert.AnError
    42  		}
    43  		err := md.Connect(ctx, sources.CmdOpts{})
    44  		assert.ErrorIs(t, err, assert.AnError)
    45  	})
    46  
    47  	t.Run("successful connection to pgbouncer", func(t *testing.T) {
    48  		mock, err := pgxmock.NewPool()
    49  		require.NoError(t, err)
    50  		sources.NewConnWithConfig = func(_ context.Context, _ *pgxpool.Config, _ ...db.ConnConfigCallback) (db.PgxPoolIface, error) {
    51  			return mock, nil
    52  		}
    53  
    54  		md := &sources.DbConn{}
    55  		md.Kind = sources.SourcePgBouncer
    56  
    57  		opts := sources.CmdOpts{}
    58  		opts.MaxParallelConnectionsPerDb = 3
    59  
    60  		mock.ExpectExec("SHOW VERSION").WillReturnResult(pgconn.NewCommandTag("SELECT 1"))
    61  
    62  		err = md.Connect(ctx, opts)
    63  		assert.NoError(t, err)
    64  
    65  		assert.NoError(t, mock.ExpectationsWereMet())
    66  	})
    67  }
    68  
    69  func TestSourceConn_ParseConfig(t *testing.T) {
    70  	md := &sources.DbConn{}
    71  	assert.NoError(t, md.ParseConfig())
    72  	//cached config
    73  	assert.NoError(t, md.ParseConfig())
    74  }
    75  
    76  func TestSourceConn_GetDatabaseName(t *testing.T) {
    77  	md := &sources.DbConn{}
    78  	md.ConnStr = "postgres://user:password@localhost:5432/mydatabase"
    79  	expected := "mydatabase"
    80  	// check pgx.ConnConfig related code
    81  	got := md.GetDatabaseName()
    82  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
    83  	// check ConnStr parsing
    84  	got = md.Source.GetDatabaseName()
    85  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
    86  
    87  	md = &sources.DbConn{}
    88  	md.ConnStr = "foo boo"
    89  	expected = ""
    90  	got = md.GetDatabaseName()
    91  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
    92  }
    93  
    94  func TestSourceConn_SetDatabaseName(t *testing.T) {
    95  	md := &sources.DbConn{}
    96  	md.ConnStr = "postgres://user:password@localhost:5432/mydatabase"
    97  	expected := "mydatabase"
    98  	// check ConnStr parsing
    99  	md.SetDatabaseName(expected)
   100  	got := md.GetDatabaseName()
   101  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
   102  	// check pgx.ConnConfig related code
   103  	expected = "newdatabase"
   104  	md.SetDatabaseName(expected)
   105  	got = md.GetDatabaseName()
   106  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
   107  
   108  	md = &sources.DbConn{}
   109  	md.ConnStr = "foo boo"
   110  	expected = ""
   111  	md.SetDatabaseName("ingored due to invalid ConnStr")
   112  	got = md.GetDatabaseName()
   113  	assert.Equal(t, expected, got, "GetDatabaseName() = %v, want %v", got, expected)
   114  }
   115  
   116  func TestSourceConn_DiscoverPlatform(t *testing.T) {
   117  	ctx := context.Background()
   118  	mock, err := pgxmock.NewPool()
   119  	require.NoError(t, err)
   120  	md := &sources.DbConn{Conn: mock}
   121  
   122  	mock.ExpectQuery("select").WillReturnRows(pgxmock.NewRows([]string{"exec_env"}).AddRow("AZURE_SINGLE"))
   123  	assert.NoError(t, md.DiscoverPlatform(ctx))
   124  	assert.Equal(t, "AZURE_SINGLE", md.ExecEnv)
   125  	assert.NoError(t, md.DiscoverPlatform(ctx)) // cached, no query expected
   126  	assert.NoError(t, mock.ExpectationsWereMet())
   127  }
   128  
   129  func TestSourceConn_GetApproxSize(t *testing.T) {
   130  	mock, err := pgxmock.NewPool()
   131  	require.NoError(t, err)
   132  	md := &sources.DbConn{Conn: mock}
   133  
   134  	mock.ExpectQuery("select").WillReturnRows(pgxmock.NewRows([]string{"size"}).AddRow(42))
   135  
   136  	assert.NoError(t, md.FetchApproxSize(ctx))
   137  	assert.EqualValues(t, 42, md.ApproxDbSize)
   138  	assert.NoError(t, mock.ExpectationsWereMet())
   139  }
   140  
   141  func TestSourceConn_FetchControlInfo(t *testing.T) {
   142  	ctx := context.Background()
   143  
   144  	t.Run("success", func(t *testing.T) {
   145  		mock, err := pgxmock.NewPool()
   146  		require.NoError(t, err)
   147  		md := &sources.DbConn{Conn: mock}
   148  
   149  		mock.ExpectQuery("select").WillReturnRows(pgxmock.NewRows(
   150  			[]string{"ver", "version", "pg_is_in_recovery", "current_database", "system_identifier", "is_superuser"},
   151  		).AddRow(16, "PostgreSQL 16.0", false, "testdb", "12345", true))
   152  
   153  		assert.NoError(t, md.FetchControlInfo(ctx))
   154  		assert.Equal(t, 16, md.Version)
   155  		assert.Equal(t, "PostgreSQL 16.0", md.VersionStr)
   156  		assert.False(t, md.IsInRecovery)
   157  		assert.Equal(t, "testdb", md.RealDbname)
   158  		assert.Equal(t, "12345", md.SystemIdentifier)
   159  		assert.True(t, md.IsSuperuser)
   160  		assert.NoError(t, mock.ExpectationsWereMet())
   161  	})
   162  
   163  	t.Run("query error", func(t *testing.T) {
   164  		mock, err := pgxmock.NewPool()
   165  		require.NoError(t, err)
   166  		md := &sources.DbConn{Conn: mock}
   167  
   168  		mock.ExpectQuery("select").WillReturnError(assert.AnError)
   169  		assert.Error(t, md.FetchControlInfo(ctx))
   170  		assert.NoError(t, mock.ExpectationsWereMet())
   171  	})
   172  }
   173  
   174  func TestSourceConn_FetchExtensions(t *testing.T) {
   175  	ctx := context.Background()
   176  
   177  	t.Run("success", func(t *testing.T) {
   178  		mock, err := pgxmock.NewPool()
   179  		require.NoError(t, err)
   180  		md := &sources.DbConn{Conn: mock, RuntimeInfo: sources.RuntimeInfo{Extensions: make(map[string]int)}}
   181  
   182  		mock.ExpectQuery("select").WillReturnRows(pgxmock.NewRows([]string{"extname", "extversion"}).
   183  			AddRow("pg_stat_statements", "1.10").
   184  			AddRow("plpgsql", "1.0"))
   185  
   186  		assert.NoError(t, md.FetchExtensions(ctx))
   187  		assert.Equal(t, 1_10_00, md.Extensions["pg_stat_statements"])
   188  		assert.Equal(t, 1_00_00, md.Extensions["plpgsql"])
   189  		assert.NoError(t, mock.ExpectationsWereMet())
   190  	})
   191  
   192  	t.Run("invalid extension version", func(t *testing.T) {
   193  		mock, err := pgxmock.NewPool()
   194  		require.NoError(t, err)
   195  		md := &sources.DbConn{Conn: mock, RuntimeInfo: sources.RuntimeInfo{Extensions: make(map[string]int)}}
   196  
   197  		mock.ExpectQuery("select").WillReturnRows(pgxmock.NewRows([]string{"extname", "extversion"}).
   198  			AddRow("badext", "notaversion"))
   199  
   200  		assert.Error(t, md.FetchExtensions(ctx))
   201  		assert.NoError(t, mock.ExpectationsWereMet())
   202  	})
   203  
   204  	t.Run("query error", func(t *testing.T) {
   205  		mock, err := pgxmock.NewPool()
   206  		require.NoError(t, err)
   207  		md := &sources.DbConn{Conn: mock, RuntimeInfo: sources.RuntimeInfo{Extensions: make(map[string]int)}}
   208  
   209  		mock.ExpectQuery("select").WillReturnError(assert.AnError)
   210  		assert.Error(t, md.FetchExtensions(ctx))
   211  		assert.NoError(t, mock.ExpectationsWereMet())
   212  	})
   213  }
   214  
   215  func TestSourceConn_FunctionExists(t *testing.T) {
   216  	mock, err := pgxmock.NewPool()
   217  	require.NoError(t, err)
   218  	md := &sources.DbConn{Conn: mock}
   219  
   220  	mock.ExpectQuery("select").WithArgs("get_foo").WillReturnRows(pgxmock.NewRows([]string{"exists"}))
   221  
   222  	assert.False(t, md.FunctionExists(ctx, "get_foo"))
   223  	assert.NoError(t, mock.ExpectationsWereMet())
   224  }
   225  
   226  func TestSourceConn_IsPostgresSource(t *testing.T) {
   227  	md := &sources.DbConn{}
   228  	md.Kind = sources.SourcePostgres
   229  	assert.True(t, md.IsPostgresSource(), "IsPostgresSource() = false, want true")
   230  
   231  	md.Kind = sources.SourcePgBouncer
   232  	assert.False(t, md.IsPostgresSource(), "IsPostgresSource() = true, want false")
   233  
   234  	md.Kind = sources.SourcePgPool
   235  	assert.False(t, md.IsPostgresSource(), "IsPostgresSource() = true, want false")
   236  
   237  	md.Kind = sources.SourcePatroniDiscovery
   238  	assert.True(t, md.IsPostgresSource(), "IsPostgresSource() = false, want true")
   239  
   240  	md.Kind = sources.SourcePrometheus
   241  	assert.False(t, md.IsPostgresSource(), "IsPostgresSource() = true, want false")
   242  }
   243  
   244  func TestSourceConn_Ping(t *testing.T) {
   245  	db, err := pgxmock.NewPool()
   246  	require.NoError(t, err)
   247  	md := &sources.DbConn{Conn: db}
   248  
   249  	db.ExpectPing()
   250  	md.Kind = sources.SourcePostgres
   251  	assert.NoError(t, md.Ping(ctx), "Ping() = error, want nil")
   252  
   253  	db.ExpectExec("SHOW VERSION").WillReturnResult(pgconn.NewCommandTag("SELECT 1"))
   254  	md.Conn = db
   255  	md.Kind = sources.SourcePgBouncer
   256  	assert.NoError(t, md.Ping(ctx), "Ping() = error, want nil")
   257  }
   258  
   259  func TestSourceConn_GetMetricInterval(t *testing.T) {
   260  	md := &sources.DbConn{
   261  		Source: sources.Source{
   262  			Metrics:        metrics.MetricIntervals{"foo": 15, "bar": 25},
   263  			MetricsStandby: metrics.MetricIntervals{"foo": 35},
   264  		},
   265  	}
   266  
   267  	t.Run("primary uses Metrics", func(t *testing.T) {
   268  		md.IsInRecovery = false
   269  		assert.Equal(t, 15*time.Second, md.GetMetricInterval("foo"))
   270  		assert.Equal(t, 25*time.Second, md.GetMetricInterval("bar"))
   271  	})
   272  
   273  	t.Run("standby uses MetricsStandby if present", func(t *testing.T) {
   274  		md.IsInRecovery = true
   275  		assert.Equal(t, 35*time.Second, md.GetMetricInterval("foo"))
   276  		assert.Equal(t, time.Duration(0), md.GetMetricInterval("bar"))
   277  	})
   278  
   279  	t.Run("standby with empty MetricsStandby falls back to Metrics", func(t *testing.T) {
   280  		md.IsInRecovery = true
   281  		md.MetricsStandby = metrics.MetricIntervals{}
   282  		assert.Equal(t, 15*time.Second, md.GetMetricInterval("foo"))
   283  	})
   284  }
   285  
   286  func TestVersionToInt(t *testing.T) {
   287  	tests := []struct {
   288  		arg  string
   289  		want int
   290  	}{
   291  		{"", 0},
   292  		{"foo", 0},
   293  		{"13", 13_00_00},
   294  		{"3.0", 3_00_00},
   295  		{"9.6.3", 9_06_03},
   296  		{"v9.6-beta2", 9_06_00},
   297  	}
   298  	for _, tt := range tests {
   299  		if got := sources.VersionToInt(tt.arg); got != tt.want {
   300  			t.Errorf("VersionToInt() = %v, want %v", got, tt.want)
   301  		}
   302  	}
   303  }
   304  
   305  func TestSourceConn_FetchRuntimeInfo(t *testing.T) {
   306  	ctx := context.Background()
   307  
   308  	t.Run("cancelled context", func(t *testing.T) {
   309  		ctxNew, cancel := context.WithCancel(ctx)
   310  		cancel()
   311  		err := (&sources.DbConn{}).FetchRuntimeInfo(ctxNew, true)
   312  		assert.Error(t, err)
   313  	})
   314  
   315  	t.Run("cached version", func(t *testing.T) {
   316  		md := sources.NewDbConn(sources.Source{})
   317  		md.SetLastCheckedForTesting(time.Now().Add(-time.Minute)) // within 5-minute TTL
   318  		md.Version = 42
   319  		err := md.FetchRuntimeInfo(ctx, false)
   320  		assert.NoError(t, err)
   321  		assert.Equal(t, 42, md.Version)
   322  	})
   323  
   324  	t.Run("pgbouncer version fetch", func(t *testing.T) {
   325  		mock, err := pgxmock.NewPool()
   326  		require.NoError(t, err)
   327  		md := sources.NewDbConn(sources.Source{Kind: sources.SourcePgBouncer})
   328  		md.Conn = mock
   329  		mock.ExpectQuery("SHOW VERSION").
   330  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   331  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("PgBouncer 1.12.0"))
   332  		err = md.FetchRuntimeInfo(ctx, true)
   333  		assert.NoError(t, err)
   334  		assert.Contains(t, md.VersionStr, "PgBouncer")
   335  		assert.True(t, md.Version > 0)
   336  		assert.NoError(t, mock.ExpectationsWereMet())
   337  	})
   338  
   339  	t.Run("pgpool version fetch", func(t *testing.T) {
   340  		mock, err := pgxmock.NewPool()
   341  		require.NoError(t, err)
   342  		md := sources.NewDbConn(sources.Source{Kind: sources.SourcePgPool})
   343  		md.Conn = mock
   344  		mock.ExpectQuery("SHOW POOL_VERSION").
   345  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   346  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("4.1.2"))
   347  		err = md.FetchRuntimeInfo(ctx, true)
   348  		assert.NoError(t, err)
   349  		assert.Contains(t, md.VersionStr, "4.1.2")
   350  		assert.True(t, md.Version > 0)
   351  		assert.NoError(t, mock.ExpectationsWereMet())
   352  	})
   353  
   354  	t.Run("postgres version and extensions", func(t *testing.T) {
   355  		mock, err := pgxmock.NewPool()
   356  		require.NoError(t, err)
   357  		md := sources.NewDbConn(sources.Source{Kind: sources.SourcePostgres})
   358  		md.Conn = mock
   359  		mock.ExpectQuery("select").WillReturnRows(
   360  			pgxmock.NewRows([]string{"ver", "version", "pg_is_in_recovery", "current_database", "system_identifier", "is_superuser"}).
   361  				AddRow(13, "PostgreSQL 13.3", false, "testdb", "42424242", true),
   362  		)
   363  		mock.ExpectQuery("select").WillReturnRows(
   364  			pgxmock.NewRows([]string{"exec_env"}).AddRow("UNKNOWN"),
   365  		)
   366  		mock.ExpectQuery("select").WillReturnRows(
   367  			pgxmock.NewRows([]string{"approx_size"}).AddRow(42),
   368  		)
   369  
   370  		mock.ExpectQuery("select").WillReturnRows(
   371  			pgxmock.NewRows([]string{"extname", "extversion"}).AddRow("pg_stat_statements", "1.8"),
   372  		)
   373  		err = md.FetchRuntimeInfo(ctx, true)
   374  		assert.NoError(t, err)
   375  		assert.Equal(t, 13, md.Version)
   376  		assert.Equal(t, "testdb", md.RealDbname)
   377  		assert.Contains(t, md.Extensions, "pg_stat_statements")
   378  		assert.NoError(t, mock.ExpectationsWereMet())
   379  	})
   380  
   381  	t.Run("query error", func(t *testing.T) {
   382  		mock, err := pgxmock.NewPool()
   383  		require.NoError(t, err)
   384  		md := sources.NewDbConn(sources.Source{Kind: sources.SourcePgBouncer})
   385  		md.Conn = mock
   386  		mock.ExpectQuery("SHOW VERSION").
   387  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   388  			WillReturnError(fmt.Errorf("db error"))
   389  		err = md.FetchRuntimeInfo(ctx, true)
   390  		assert.Error(t, err)
   391  		assert.NoError(t, mock.ExpectationsWereMet())
   392  	})
   393  }
   394  
   395  func TestSourceConn_FetchVersion(t *testing.T) {
   396  	ctx := context.Background()
   397  
   398  	t.Run("valid version string", func(t *testing.T) {
   399  		mock, err := pgxmock.NewPool()
   400  		require.NoError(t, err)
   401  		md := &sources.DbConn{Conn: mock, Source: sources.Source{Kind: sources.SourcePgBouncer}}
   402  		mock.ExpectQuery("SHOW VERSION").
   403  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   404  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("FooBar 1.12.0"))
   405  		assert.NoError(t, md.FetchVersion(ctx, sources.SourcePgBouncer))
   406  		assert.Equal(t, "FooBar 1.12.0", md.VersionStr)
   407  		assert.Equal(t, 1_12_00, md.Version)
   408  		assert.NoError(t, mock.ExpectationsWereMet())
   409  	})
   410  
   411  	t.Run("invalid version string", func(t *testing.T) {
   412  		mock, err := pgxmock.NewPool()
   413  		require.NoError(t, err)
   414  		md := &sources.DbConn{Conn: mock, Source: sources.Source{Kind: sources.SourcePgBouncer}}
   415  		mock.ExpectQuery("SHOW VERSION").
   416  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   417  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("invalid version"))
   418  		assert.NoError(t, md.FetchVersion(ctx, sources.SourcePgBouncer))
   419  		assert.Equal(t, 0, md.Version)
   420  		assert.NoError(t, mock.ExpectationsWereMet())
   421  	})
   422  
   423  	t.Run("query error", func(t *testing.T) {
   424  		mock, err := pgxmock.NewPool()
   425  		require.NoError(t, err)
   426  		md := &sources.DbConn{Conn: mock, Source: sources.Source{Kind: sources.SourcePgBouncer}}
   427  		mock.ExpectQuery("SHOW VERSION").
   428  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   429  			WillReturnError(assert.AnError)
   430  		assert.Error(t, md.FetchVersion(ctx, sources.SourcePgBouncer))
   431  		assert.NoError(t, mock.ExpectationsWereMet())
   432  	})
   433  }
   434  
   435  func TestSourceConn_GetClusterIdentifier(t *testing.T) {
   436  	md := &sources.DbConn{
   437  		Source: sources.Source{
   438  			Name:    "test",
   439  			Kind:    sources.SourcePostgres,
   440  			ConnStr: "postgres://user:password@localhost:5432/mydatabase",
   441  		},
   442  		RuntimeInfo: sources.RuntimeInfo{
   443  			SystemIdentifier: "42424242",
   444  		},
   445  	}
   446  	assert.Equal(t, "42424242:localhost:5432", md.GetClusterIdentifier())
   447  
   448  	md = &sources.DbConn{
   449  		Source: sources.Source{
   450  			Name:    "test",
   451  			Kind:    sources.SourcePostgres,
   452  			ConnStr: "foo boo",
   453  		},
   454  	}
   455  	assert.Equal(t, "", md.GetClusterIdentifier())
   456  }
   457  
   458  func TestTryCreateMetricsFetchingHelpers(t *testing.T) {
   459  	ctx := context.Background()
   460  	mock, err := pgxmock.NewPool()
   461  	assert.NoError(t, err)
   462  	defer mock.Close()
   463  
   464  	fn := func(metric string) string {
   465  		if metric == "metric1" {
   466  			return "CREATE FUNCTION metric1"
   467  		}
   468  		return ""
   469  	}
   470  
   471  	md := &sources.DbConn{
   472  		Conn: mock,
   473  		Source: sources.Source{
   474  			Name:           "testdb",
   475  			Metrics:        metrics.MetricIntervals{"metric1": 42, "nonexistent": 0},
   476  			MetricsStandby: metrics.MetricIntervals{"metric1": 42},
   477  		},
   478  	}
   479  
   480  	t.Run("success", func(t *testing.T) {
   481  		mock.ExpectExec("CREATE FUNCTION metric1").WillReturnResult(pgxmock.NewResult("CREATE", 1))
   482  
   483  		err = md.TryCreateMetricsHelpers(ctx, fn)
   484  		assert.NoError(t, err)
   485  		assert.NoError(t, mock.ExpectationsWereMet())
   486  	})
   487  
   488  	t.Run("error on exec", func(t *testing.T) {
   489  		mock.ExpectExec("CREATE FUNCTION metric1").WillReturnError(assert.AnError)
   490  
   491  		err = md.TryCreateMetricsHelpers(ctx, fn)
   492  		assert.Error(t, err)
   493  		assert.NoError(t, mock.ExpectationsWereMet())
   494  	})
   495  
   496  }
   497  
   498  func TestTryCreateMissingExtensions(t *testing.T) {
   499  	ctx := context.Background()
   500  
   501  	availableExtsRows := func() *pgxmock.Rows {
   502  		return pgxmock.NewRows([]string{"name"}).
   503  			AddRow("pg_stat_statements").
   504  			AddRow("pg_trgm")
   505  	}
   506  
   507  	t.Run("extension already loaded", func(t *testing.T) {
   508  		mock, err := pgxmock.NewPool()
   509  		require.NoError(t, err)
   510  		defer mock.Close()
   511  
   512  		md := &sources.DbConn{
   513  			Conn: mock,
   514  			RuntimeInfo: sources.RuntimeInfo{
   515  				Extensions: map[string]int{"pg_stat_statements": 10800},
   516  			},
   517  		}
   518  
   519  		mock.ExpectQuery("select name").WillReturnRows(availableExtsRows())
   520  
   521  		created, err := md.TryCreateMissingExtensions(ctx, []string{"pg_stat_statements"})
   522  		assert.NoError(t, err)
   523  		assert.Empty(t, created)
   524  		assert.NoError(t, mock.ExpectationsWereMet())
   525  	})
   526  
   527  	t.Run("extension not available on instance", func(t *testing.T) {
   528  		mock, err := pgxmock.NewPool()
   529  		require.NoError(t, err)
   530  		defer mock.Close()
   531  
   532  		md := &sources.DbConn{
   533  			Conn:        mock,
   534  			RuntimeInfo: sources.RuntimeInfo{Extensions: map[string]int{}},
   535  		}
   536  
   537  		mock.ExpectQuery("select name").WillReturnRows(availableExtsRows())
   538  
   539  		created, err := md.TryCreateMissingExtensions(ctx, []string{"missing_ext"})
   540  		assert.Error(t, err)
   541  		assert.Empty(t, created)
   542  		assert.NoError(t, mock.ExpectationsWereMet())
   543  	})
   544  
   545  	t.Run("extension created successfully", func(t *testing.T) {
   546  		mock, err := pgxmock.NewPool()
   547  		require.NoError(t, err)
   548  		defer mock.Close()
   549  
   550  		md := &sources.DbConn{
   551  			Conn:        mock,
   552  			RuntimeInfo: sources.RuntimeInfo{Extensions: map[string]int{}},
   553  		}
   554  
   555  		mock.ExpectQuery("select name").WillReturnRows(availableExtsRows())
   556  		mock.ExpectExec(`create extension if not exists`).WillReturnResult(pgxmock.NewResult("CREATE EXTENSION", 1))
   557  
   558  		created, err := md.TryCreateMissingExtensions(ctx, []string{"pg_stat_statements"})
   559  		assert.NoError(t, err)
   560  		assert.Equal(t, "pg_stat_statements", created)
   561  		assert.NoError(t, mock.ExpectationsWereMet())
   562  	})
   563  
   564  	t.Run("extension creation fails", func(t *testing.T) {
   565  		mock, err := pgxmock.NewPool()
   566  		require.NoError(t, err)
   567  		defer mock.Close()
   568  
   569  		md := &sources.DbConn{
   570  			Conn:        mock,
   571  			RuntimeInfo: sources.RuntimeInfo{Extensions: map[string]int{}},
   572  		}
   573  
   574  		mock.ExpectQuery("select name").WillReturnRows(availableExtsRows())
   575  		mock.ExpectExec(`create extension if not exists`).WillReturnError(assert.AnError)
   576  
   577  		created, err := md.TryCreateMissingExtensions(ctx, []string{"pg_stat_statements"})
   578  		assert.Error(t, err)
   579  		assert.Empty(t, created)
   580  		assert.NoError(t, mock.ExpectationsWereMet())
   581  	})
   582  
   583  	t.Run("query available extensions fails", func(t *testing.T) {
   584  		mock, err := pgxmock.NewPool()
   585  		require.NoError(t, err)
   586  		defer mock.Close()
   587  
   588  		md := &sources.DbConn{
   589  			Conn:        mock,
   590  			RuntimeInfo: sources.RuntimeInfo{Extensions: map[string]int{}},
   591  		}
   592  
   593  		mock.ExpectQuery("select name").WillReturnError(assert.AnError)
   594  
   595  		created, err := md.TryCreateMissingExtensions(ctx, []string{"pg_stat_statements"})
   596  		assert.Error(t, err)
   597  		assert.Empty(t, created)
   598  		assert.NoError(t, mock.ExpectationsWereMet())
   599  	})
   600  
   601  	t.Run("mixed: one created, one already loaded, one unavailable", func(t *testing.T) {
   602  		mock, err := pgxmock.NewPool()
   603  		require.NoError(t, err)
   604  		defer mock.Close()
   605  
   606  		md := &sources.DbConn{
   607  			Conn: mock,
   608  			RuntimeInfo: sources.RuntimeInfo{
   609  				Extensions: map[string]int{"pg_trgm": 10000},
   610  			},
   611  		}
   612  
   613  		mock.ExpectQuery("select name").WillReturnRows(availableExtsRows())
   614  		mock.ExpectExec(`create extension if not exists`).WillReturnResult(pgxmock.NewResult("CREATE EXTENSION", 1))
   615  
   616  		created, err := md.TryCreateMissingExtensions(ctx, []string{"pg_stat_statements", "pg_trgm", "missing_ext"})
   617  		assert.Error(t, err) // missing_ext not available
   618  		assert.Equal(t, "pg_stat_statements", created)
   619  		assert.NoError(t, mock.ExpectationsWereMet())
   620  	})
   621  }
   622  
   623  // DbConn.IsPostgresSource returns true for Postgres-family kinds, false for pgbouncer/pgpool.
   624  func TestDbConn_IsPostgresSource(t *testing.T) {
   625  	tests := []struct {
   626  		kind sources.Kind
   627  		want bool
   628  	}{
   629  		{sources.SourcePostgres, true},
   630  		{sources.SourcePatroniDiscovery, true},
   631  		{sources.SourcePostgresDiscovery, true},
   632  		{sources.SourcePgBouncer, false},
   633  		{sources.SourcePgPool, false},
   634  	}
   635  	for _, tt := range tests {
   636  		md := &sources.DbConn{Source: sources.Source{Kind: tt.kind}}
   637  		assert.Equal(t, tt.want, md.IsPostgresSource(), "kind=%v", tt.kind)
   638  	}
   639  }
   640  
   641  // PromConn.IsPostgresSource always returns false.
   642  func TestPromConn_IsPostgresSource(t *testing.T) {
   643  	pc := &sources.PromConn{}
   644  	assert.False(t, pc.IsPostgresSource())
   645  }
   646  
   647  func TestPromConn_Connect_TLSSkipVerify(t *testing.T) {
   648  	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
   649  		assert.Empty(t, r.URL.Query().Get("tlsskipverify"), "tlsskipverify param should be stripped")
   650  		assert.Empty(t, r.URL.Query().Get("tlsrootcert"), "tlsrootcert param should be stripped")
   651  		w.WriteHeader(http.StatusOK)
   652  	}))
   653  	t.Cleanup(srv.Close)
   654  
   655  	ctx := testutil.TestContext
   656  
   657  	connStr := srv.URL + "/metrics?tlsskipverify=true"
   658  	pc := sources.NewPromConn(sources.Source{ConnStr: connStr})
   659  
   660  	err := pc.Connect(ctx, sources.CmdOpts{})
   661  	assert.NoError(t, err, "Connect should succeed with tlsskipverify=true")
   662  	assert.NotNil(t, pc.HTTPClient, "HTTPClient should be set after Connect")
   663  }
   664  
   665  func TestPromConn_Connect_BasicAuth(t *testing.T) {
   666  	var capturedAuth string
   667  	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
   668  		capturedAuth = r.Header.Get("Authorization")
   669  		w.WriteHeader(http.StatusOK)
   670  	}))
   671  	t.Cleanup(srv.Close)
   672  
   673  	u, err := url.Parse(srv.URL)
   674  	require.NoError(t, err)
   675  	u.Path = "/metrics"
   676  	u.User = url.UserPassword("pgwatch", "supersecret")
   677  
   678  	ctx, logOutput := testutil.NewTestLogger(t, logrus.DebugLevel)
   679  
   680  	pc := sources.NewPromConn(sources.Source{ConnStr: u.String()})
   681  	err = pc.Connect(ctx, sources.CmdOpts{})
   682  	assert.NoError(t, err)
   683  	assert.NotEmpty(t, capturedAuth, "Authorization header should be sent")
   684  	assert.True(t, strings.HasPrefix(capturedAuth, "Basic "), "should be Basic auth")
   685  	assert.NotContains(t, logOutput.String(), "supersecret", "password must not be logged (SEC-001)")
   686  }
   687  
   688  func TestPromConn_Connect_Unreachable(t *testing.T) {
   689  	ctx := t.Context()
   690  	pc := sources.NewPromConn(sources.Source{
   691  		ConnStr: "http://127.0.0.1:1/metrics",
   692  	})
   693  	assert.Error(t, pc.Connect(ctx, sources.CmdOpts{}))
   694  }
   695  
   696  func TestPromConn_Ping(t *testing.T) {
   697  	tests := []struct {
   698  		name       string
   699  		statusCode int
   700  		wantErr    bool
   701  	}{
   702  		{"200 OK", http.StatusOK, false},
   703  		{"204 No Content", http.StatusNoContent, false},
   704  		{"400 Bad Request", http.StatusBadRequest, true},
   705  		{"500 Internal Server Error", http.StatusInternalServerError, true},
   706  		{"301 Redirect", http.StatusMovedPermanently, true},
   707  	}
   708  
   709  	for _, tt := range tests {
   710  		t.Run(tt.name, func(t *testing.T) {
   711  			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
   712  				w.WriteHeader(tt.statusCode)
   713  			}))
   714  			t.Cleanup(srv.Close)
   715  
   716  			pc := sources.NewPromConn(sources.Source{ConnStr: srv.URL + "/metrics"})
   717  			pc.HTTPClient = srv.Client()
   718  			require.NoError(t, pc.ParseConfig())
   719  			err := pc.Ping(t.Context())
   720  			if tt.wantErr {
   721  				assert.Error(t, err)
   722  			} else {
   723  				assert.NoError(t, err)
   724  			}
   725  		})
   726  	}
   727  }
   728  
   729  func TestPromConn_FetchRuntimeInfo(t *testing.T) {
   730  	pc := sources.NewPromConn(sources.Source{})
   731  	err := pc.FetchRuntimeInfo(t.Context(), false)
   732  	assert.NoError(t, err)
   733  }
   734  
   735  func TestRedactURL(t *testing.T) {
   736  	tests := []struct {
   737  		name  string
   738  		input string
   739  		want  string
   740  	}{
   741  		{
   742  			name:  "password is redacted",
   743  			input: "http://user:secret@localhost:9187/metrics",
   744  			want:  "http://user:xxxxx@localhost:9187/metrics",
   745  		},
   746  		{
   747  			name:  "no userinfo unchanged",
   748  			input: "http://localhost:9187/metrics",
   749  			want:  "http://localhost:9187/metrics",
   750  		},
   751  		{
   752  			name:  "username only (no password) unchanged",
   753  			input: "http://user@localhost:9187/metrics",
   754  			want:  "http://user@localhost:9187/metrics",
   755  		},
   756  		{
   757  			name:  "query params preserved",
   758  			input: "http://user:pass@localhost:9187/metrics?tlsskipverify=true",
   759  			want:  "http://user:xxxxx@localhost:9187/metrics?tlsskipverify=true",
   760  		},
   761  	}
   762  
   763  	for _, tt := range tests {
   764  		t.Run(tt.name, func(t *testing.T) {
   765  			got := sources.RedactURL(tt.input)
   766  			assert.Equal(t, tt.want, got)
   767  		})
   768  	}
   769  }
   770  
   771  // TestRace_GetClusterIdentifier verifies that concurrent FetchRuntimeInfo writes
   772  // and GetClusterIdentifier reads do not cause a data race.
   773  func TestRace_GetClusterIdentifier(t *testing.T) {
   774  	mock, err := pgxmock.NewPool()
   775  	require.NoError(t, err)
   776  	defer mock.Close()
   777  
   778  	md := sources.NewDbConn(sources.Source{
   779  		Kind:    sources.SourcePgBouncer,
   780  		ConnStr: "postgres://user:pass@localhost:5432/db",
   781  	})
   782  	md.Conn = mock
   783  
   784  	const iterations = 50
   785  	// FetchRuntimeInfo for pgbouncer needs one SHOW VERSION per forceRefetch=true call.
   786  	for range iterations {
   787  		mock.ExpectQuery("SHOW VERSION").
   788  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   789  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("PgBouncer 1.12.0"))
   790  	}
   791  
   792  	var wg sync.WaitGroup
   793  	wg.Add(2)
   794  
   795  	go func() {
   796  		defer wg.Done()
   797  		for range iterations {
   798  			_ = md.FetchRuntimeInfo(context.Background(), true)
   799  		}
   800  	}()
   801  
   802  	go func() {
   803  		defer wg.Done()
   804  		for range iterations {
   805  			_ = md.GetClusterIdentifier()
   806  		}
   807  	}()
   808  
   809  	wg.Wait()
   810  }
   811  
   812  // TestRace_TryCreateMissingExtensions verifies that concurrent FetchRuntimeInfo
   813  // writes to Extensions and TryCreateMissingExtensions reads do not race.
   814  func TestRace_TryCreateMissingExtensions(t *testing.T) {
   815  	mock, err := pgxmock.NewPool()
   816  	require.NoError(t, err)
   817  	defer mock.Close()
   818  
   819  	md := sources.NewDbConn(sources.Source{Kind: sources.SourcePostgres})
   820  	md.Conn = mock
   821  
   822  	const iterations = 50
   823  	// Each TryCreateMissingExtensions call queries available extensions then skips
   824  	// creation because "pg_stat_statements" will be in knownExts after the first write.
   825  	for range iterations {
   826  		mock.ExpectQuery("select name").
   827  			WillReturnRows(pgxmock.NewRows([]string{"name"}).AddRow("pg_stat_statements"))
   828  	}
   829  
   830  	var wg sync.WaitGroup
   831  	wg.Add(2)
   832  
   833  	// Writer: simulate FetchRuntimeInfo updating Extensions under Lock.
   834  	go func() {
   835  		defer wg.Done()
   836  		for range iterations {
   837  			md.Lock()
   838  			md.Extensions = map[string]int{"pg_stat_statements": 10800}
   839  			md.Unlock()
   840  		}
   841  	}()
   842  
   843  	// Reader: TryCreateMissingExtensions snapshots Extensions under RLock then does I/O unlocked.
   844  	go func() {
   845  		defer wg.Done()
   846  		for range iterations {
   847  			_, _ = md.TryCreateMissingExtensions(context.Background(), []string{"pg_stat_statements"})
   848  		}
   849  	}()
   850  
   851  	wg.Wait()
   852  }
   853  
   854  // TestRace_TryCreateMetricsHelpers verifies that concurrent SetMetricIntervals
   855  // writes and TryCreateMetricsHelpers reads do not race.
   856  func TestRace_TryCreateMetricsHelpers(t *testing.T) {
   857  	mock, err := pgxmock.NewPool()
   858  	require.NoError(t, err)
   859  	defer mock.Close()
   860  
   861  	md := sources.NewDbConn(sources.Source{
   862  		Kind:    sources.SourcePostgres,
   863  		Metrics: metrics.MetricIntervals{"metric1": 30},
   864  	})
   865  	md.Conn = mock
   866  
   867  	const iterations = 50
   868  	for range iterations {
   869  		mock.ExpectExec("CREATE FUNCTION metric1").
   870  			WillReturnResult(pgxmock.NewResult("CREATE", 1))
   871  	}
   872  
   873  	getSQLFn := func(metric string) string {
   874  		if metric == "metric1" {
   875  			return "CREATE FUNCTION metric1"
   876  		}
   877  		return ""
   878  	}
   879  
   880  	var wg sync.WaitGroup
   881  	wg.Add(2)
   882  
   883  	// Writer: simulate SetMetricIntervals updating Metrics under Lock.
   884  	go func() {
   885  		defer wg.Done()
   886  		for range iterations {
   887  			md.SetMetricIntervals(metrics.MetricIntervals{"metric1": 30}, nil)
   888  		}
   889  	}()
   890  
   891  	// Reader: TryCreateMetricsHelpers clones Metrics under RLock then does I/O unlocked.
   892  	go func() {
   893  		defer wg.Done()
   894  		for range iterations {
   895  			_ = md.TryCreateMetricsHelpers(context.Background(), getSQLFn)
   896  		}
   897  	}()
   898  
   899  	wg.Wait()
   900  }
   901  
   902  // TestRace_FetchRuntimeInfoAtomicCache verifies that the atomic lastCheckedNs
   903  // fast path and double-checked lock in FetchRuntimeInfo are race-free under
   904  // concurrent calls from multiple goroutines.
   905  func TestRace_FetchRuntimeInfoAtomicCache(t *testing.T) {
   906  	mock, err := pgxmock.NewPool()
   907  	require.NoError(t, err)
   908  	defer mock.Close()
   909  
   910  	md := sources.NewDbConn(sources.Source{Kind: sources.SourcePgBouncer})
   911  	md.Conn = mock
   912  
   913  	const goroutines = 4
   914  	const iterations = 50
   915  
   916  	// Pre-seed enough mock responses for the worst-case where every call bypasses the cache.
   917  	for range goroutines * iterations {
   918  		mock.ExpectQuery("SHOW VERSION").
   919  			WithArgs(pgx.QueryExecModeSimpleProtocol).
   920  			WillReturnRows(pgxmock.NewRows([]string{"version"}).AddRow("PgBouncer 1.12.0"))
   921  	}
   922  
   923  	var wg sync.WaitGroup
   924  	wg.Add(goroutines)
   925  
   926  	for range goroutines {
   927  		go func() {
   928  			defer wg.Done()
   929  			for range iterations {
   930  				// Mix forced and cached calls to exercise both fast and slow paths.
   931  				_ = md.FetchRuntimeInfo(context.Background(), false)
   932  			}
   933  		}()
   934  	}
   935  
   936  	wg.Wait()
   937  }
   938  
   939  // Ping against a half-open TCP peer (BlackholeListener) must return
   940  // within PingTimeoutMargin (ConnectTimeout is unset in the conn string,
   941  // so the wired bound collapses to PingTimeoutMargin alone). Today Ping
   942  // has no client-side bound: pgx's net.Dialer fallback is on the order
   943  // of minutes, so the test fails RED at the safety-net cancel. With the
   944  // wiring added to Ping, Ping derives a "ping"-tagged ctx and the call
   945  // returns at ~PingTimeoutMargin.
   946  func TestSourceConn_Ping_BoundedAgainstHalfOpenTCP(t *testing.T) {
   947  	addr, _ := testutil.BlackholeListener(t)
   948  
   949  	// Shrink the margin so the bound is small. Do NOT set connect_timeout
   950  	// in the conn string — that would let pgx bound the dial itself and
   951  	// mask whether the wiring is actually firing.
   952  	origMargin := db.PingTimeoutMargin
   953  	t.Cleanup(func() { db.PingTimeoutMargin = origMargin })
   954  	db.PingTimeoutMargin = 100 * time.Millisecond
   955  
   956  	connStr := "postgres://x@" + addr + "/postgres"
   957  	cfg, err := pgxpool.ParseConfig(connStr)
   958  	require.NoError(t, err)
   959  	require.Zero(t, cfg.ConnConfig.ConnectTimeout, "precondition: no connect_timeout in conn string")
   960  	// Use a real pgxpool so Ping's Acquire→dial exercises the half-open
   961  	// peer. The pool does not dial until Acquire is called; the dial is
   962  	// what we want to bound.
   963  	pool, err := pgxpool.NewWithConfig(context.Background(), cfg)
   964  	require.NoError(t, err)
   965  	t.Cleanup(pool.Close)
   966  
   967  	md := &sources.DbConn{
   968  		Source: sources.Source{
   969  			Name:    "blackhole_ping_src",
   970  			Kind:    sources.SourcePostgres,
   971  			ConnStr: connStr,
   972  		},
   973  		Conn:       pool,
   974  		ConnConfig: cfg,
   975  	}
   976  
   977  	ctx, cancel := context.WithCancel(ctx)
   978  	defer cancel()
   979  	// Safety net so a regressed implementation does not hang the test
   980  	// indefinitely. With the wiring the ping deadline fires first.
   981  	time.AfterFunc(5*time.Second, cancel)
   982  
   983  	start := time.Now()
   984  	err = md.Ping(ctx)
   985  	elapsed := time.Since(start)
   986  
   987  	require.Error(t, err)
   988  	// Without wiring, pgx waits its OS dialer fallback (~75s on Windows,
   989  	// often tens of seconds elsewhere) so the test only finishes when the
   990  	// safety net fires. With wiring the call returns at ~PingTimeoutMargin.
   991  	if elapsed > 1500*time.Millisecond {
   992  		t.Fatalf("Ping took %v, want ~PingTimeoutMargin (100ms)", elapsed)
   993  	}
   994  }
   995  
   996  // Ping on a DbConn whose pool Acquire() blocks (every conn in the pool
   997  // is wedged) must fail at the PingTimeoutMargin bound rather than
   998  // hanging. Today Ping forwards the caller's ctx directly to pool.Ping,
   999  // which internally acquires a conn — Acquire blocks until ctx is
  1000  // cancelled. With the wiring, the derived ctx fires first.
  1001  //
  1002  // Runs in a synctest bubble: BlockingPool.Acquire blocks durably on
  1003  // ctx.Done(), so the test goroutine drives Ping on a worker and uses
  1004  // time.Sleep to schedule a bubble wake-up past the 100ms bound.
  1005  func TestSourceConn_Ping_BoundedBehindWedgedPool(t *testing.T) {
  1006  	synctest.Test(t, func(t *testing.T) {
  1007  		// Shrink the margin so the bound is small. The BlockingPool
  1008  		// ignores ConnectTimeout, so the effective bound comes
  1009  		// entirely from PingTimeoutMargin.
  1010  		db.PingTimeoutMargin = 100 * time.Millisecond
  1011  		t.Cleanup(func() { db.PingTimeoutMargin = 5 * time.Second })
  1012  
  1013  		md := &sources.DbConn{
  1014  			Source: sources.Source{
  1015  				Name: "wedged_pool_ping_src",
  1016  				Kind: sources.SourcePostgres,
  1017  			},
  1018  			Conn: testutil.BlockingPool{},
  1019  		}
  1020  		// Provide a minimal ConnConfig so Ping's derived ctx can read
  1021  		// ConnectTimeout (zero here, so the wired bound collapses to
  1022  		// just PingTimeoutMargin).
  1023  		md.ConnConfig = &pgxpool.Config{}
  1024  
  1025  		// Bounded callCtx so the bubble has a wake-up event even when
  1026  		// the production call blocks indefinitely (RED). With the
  1027  		// wiring, Ping derives a 100ms ctx internally and returns long
  1028  		// before callCtx fires; without the wiring the call hangs
  1029  		// until callCtx fires at t=2s and the elapsed-time assertion
  1030  		// catches it.
  1031  		callCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
  1032  		defer cancel()
  1033  
  1034  		done := make(chan error, 1)
  1035  		start := time.Now()
  1036  		go func() { done <- md.Ping(callCtx) }()
  1037  
  1038  		// Yield to the bubble, then sleep past the expected bound. The
  1039  		// bubble advances virtual time either when the wired call
  1040  		// returns (GREEN) or when callCtx fires (RED).
  1041  		time.Sleep(500 * time.Millisecond)
  1042  		synctest.Wait()
  1043  
  1044  		err := <-done
  1045  		elapsed := time.Since(start)
  1046  
  1047  		require.Error(t, err)
  1048  		if elapsed > 1500*time.Millisecond {
  1049  			t.Fatalf("Ping took %v, want ~PingTimeoutMargin (100ms)", elapsed)
  1050  		}
  1051  	})
  1052  }
  1053  func TestSourceConn_FetchRuntimeInfo_BoundedByRuntimeInfoTimeout(t *testing.T) {
  1054  	synctest.Test(t, func(t *testing.T) {
  1055  		db.RuntimeInfoTimeout = 100 * time.Millisecond
  1056  		t.Cleanup(func() { db.RuntimeInfoTimeout = 30 * time.Second })
  1057  
  1058  		md := sources.NewDbConn(sources.Source{
  1059  			Name: "bounded_runtime_info_src",
  1060  			Kind: sources.SourcePostgres,
  1061  		})
  1062  		md.Conn = testutil.BlockingPool{}
  1063  
  1064  		// Bounded callCtx so the bubble has a wake-up event even when
  1065  		// the production call blocks indefinitely (RED). With the
  1066  		// wiring, each sub-query derives a 100ms ctx internally and
  1067  		// returns long before callCtx fires; without the wiring the
  1068  		// call hangs until callCtx fires at t=2s and the elapsed-time
  1069  		// assertion catches it.
  1070  		callCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
  1071  		defer cancel()
  1072  
  1073  		done := make(chan error, 1)
  1074  		start := time.Now()
  1075  		go func() { done <- md.FetchRuntimeInfo(callCtx, true) }()
  1076  
  1077  		time.Sleep(500 * time.Millisecond)
  1078  		synctest.Wait()
  1079  
  1080  		err := <-done
  1081  		elapsed := time.Since(start)
  1082  
  1083  		require.Error(t, err)
  1084  		if elapsed > 1500*time.Millisecond {
  1085  			t.Fatalf("FetchRuntimeInfo took %v, want ~RuntimeInfoTimeout (100ms)", elapsed)
  1086  		}
  1087  	})
  1088  }
  1089