client_test.go

   1 package torrent
   2
   3 import (
   4         "encoding/binary"
   5         "fmt"
   6         "io"
   7         "io/ioutil"
   8         "os"
   9         "path/filepath"
  10         "reflect"
  11         "sync"
  12         "testing"
  13         "time"
  14
  15         "github.com/bradfitz/iter"
  16         "github.com/stretchr/testify/assert"
  17         "github.com/stretchr/testify/require"
  18         "golang.org/x/time/rate"
  19
  20         "github.com/anacrolix/dht"
  21         _ "github.com/anacrolix/envpprof"
  22         "github.com/anacrolix/missinggo"
  23         "github.com/anacrolix/missinggo/filecache"
  24         "github.com/anacrolix/torrent/bencode"
  25         "github.com/anacrolix/torrent/internal/testutil"
  26         "github.com/anacrolix/torrent/iplist"
  27         "github.com/anacrolix/torrent/metainfo"
  28         "github.com/anacrolix/torrent/storage"
  29 )
  30
  31 func TestingConfig() *ClientConfig {
  32         cfg := NewDefaultClientConfig()
  33         cfg.ListenHost = LoopbackListenHost
  34         cfg.NoDHT = true
  35         cfg.DataDir = tempDir()
  36         cfg.DisableTrackers = true
  37         cfg.NoDefaultPortForwarding = true
  38         cfg.DisableAcceptRateLimiting = true
  39         cfg.ListenPort = 0
  40         return cfg
  41 }
  42
  43 func TestClientDefault(t *testing.T) {
  44         cl, err := NewClient(TestingConfig())
  45         require.NoError(t, err)
  46         cl.Close()
  47 }
  48
  49 func TestClientNilConfig(t *testing.T) {
  50         cl, err := NewClient(nil)
  51         require.NoError(t, err)
  52         cl.Close()
  53 }
  54
  55 func TestBoltPieceCompletionClosedWhenClientClosed(t *testing.T) {
  56         cfg := TestingConfig()
  57         pc, err := storage.NewBoltPieceCompletion(cfg.DataDir)
  58         require.NoError(t, err)
  59         ci := storage.NewFileWithCompletion(cfg.DataDir, pc)
  60         defer ci.Close()
  61         cfg.DefaultStorage = ci
  62         cl, err := NewClient(cfg)
  63         require.NoError(t, err)
  64         cl.Close()
  65         // And again, https://github.com/anacrolix/torrent/issues/158
  66         cl, err = NewClient(cfg)
  67         require.NoError(t, err)
  68         cl.Close()
  69 }
  70
  71 func TestAddDropTorrent(t *testing.T) {
  72         cl, err := NewClient(TestingConfig())
  73         require.NoError(t, err)
  74         defer cl.Close()
  75         dir, mi := testutil.GreetingTestTorrent()
  76         defer os.RemoveAll(dir)
  77         tt, new, err := cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
  78         require.NoError(t, err)
  79         assert.True(t, new)
  80         tt.SetMaxEstablishedConns(0)
  81         tt.SetMaxEstablishedConns(1)
  82         tt.Drop()
  83 }
  84
  85 func TestAddTorrentNoSupportedTrackerSchemes(t *testing.T) {
  86         // TODO?
  87         t.SkipNow()
  88 }
  89
  90 func TestAddTorrentNoUsableURLs(t *testing.T) {
  91         // TODO?
  92         t.SkipNow()
  93 }
  94
  95 func TestAddPeersToUnknownTorrent(t *testing.T) {
  96         // TODO?
  97         t.SkipNow()
  98 }
  99
 100 func TestPieceHashSize(t *testing.T) {
 101         assert.Equal(t, 20, pieceHash.Size())
 102 }
 103
 104 func TestTorrentInitialState(t *testing.T) {
 105         dir, mi := testutil.GreetingTestTorrent()
 106         defer os.RemoveAll(dir)
 107         cl := &Client{
 108                 config: &ClientConfig{},
 109         }
 110         cl.initLogger()
 111         tor := cl.newTorrent(
 112                 mi.HashInfoBytes(),
 113                 storage.NewFileWithCompletion(tempDir(), storage.NewMapPieceCompletion()),
 114         )
 115         tor.setChunkSize(2)
 116         tor.cl.lock()
 117         err := tor.setInfoBytes(mi.InfoBytes)
 118         tor.cl.unlock()
 119         require.NoError(t, err)
 120         require.Len(t, tor.pieces, 3)
 121         tor.pendAllChunkSpecs(0)
 122         tor.cl.lock()
 123         assert.EqualValues(t, 3, tor.pieceNumPendingChunks(0))
 124         tor.cl.unlock()
 125         assert.EqualValues(t, chunkSpec{4, 1}, chunkIndexSpec(2, tor.pieceLength(0), tor.chunkSize))
 126 }
 127
 128 func TestReducedDialTimeout(t *testing.T) {
 129         cfg := NewDefaultClientConfig()
 130         for _, _case := range []struct {
 131                 Max             time.Duration
 132                 HalfOpenLimit   int
 133                 PendingPeers    int
 134                 ExpectedReduced time.Duration
 135         }{
 136                 {cfg.NominalDialTimeout, 40, 0, cfg.NominalDialTimeout},
 137                 {cfg.NominalDialTimeout, 40, 1, cfg.NominalDialTimeout},
 138                 {cfg.NominalDialTimeout, 40, 39, cfg.NominalDialTimeout},
 139                 {cfg.NominalDialTimeout, 40, 40, cfg.NominalDialTimeout / 2},
 140                 {cfg.NominalDialTimeout, 40, 80, cfg.NominalDialTimeout / 3},
 141                 {cfg.NominalDialTimeout, 40, 4000, cfg.NominalDialTimeout / 101},
 142         } {
 143                 reduced := reducedDialTimeout(cfg.MinDialTimeout, _case.Max, _case.HalfOpenLimit, _case.PendingPeers)
 144                 expected := _case.ExpectedReduced
 145                 if expected < cfg.MinDialTimeout {
 146                         expected = cfg.MinDialTimeout
 147                 }
 148                 if reduced != expected {
 149                         t.Fatalf("expected %s, got %s", _case.ExpectedReduced, reduced)
 150                 }
 151         }
 152 }
 153
 154 func TestAddDropManyTorrents(t *testing.T) {
 155         cl, err := NewClient(TestingConfig())
 156         require.NoError(t, err)
 157         defer cl.Close()
 158         for i := range iter.N(1000) {
 159                 var spec TorrentSpec
 160                 binary.PutVarint(spec.InfoHash[:], int64(i))
 161                 tt, new, err := cl.AddTorrentSpec(&spec)
 162                 assert.NoError(t, err)
 163                 assert.True(t, new)
 164                 defer tt.Drop()
 165         }
 166 }
 167
 168 type FileCacheClientStorageFactoryParams struct {
 169         Capacity    int64
 170         SetCapacity bool
 171         Wrapper     func(*filecache.Cache) storage.ClientImpl
 172 }
 173
 174 func NewFileCacheClientStorageFactory(ps FileCacheClientStorageFactoryParams) storageFactory {
 175         return func(dataDir string) storage.ClientImpl {
 176                 fc, err := filecache.NewCache(dataDir)
 177                 if err != nil {
 178                         panic(err)
 179                 }
 180                 if ps.SetCapacity {
 181                         fc.SetCapacity(ps.Capacity)
 182                 }
 183                 return ps.Wrapper(fc)
 184         }
 185 }
 186
 187 type storageFactory func(string) storage.ClientImpl
 188
 189 func TestClientTransferDefault(t *testing.T) {
 190         testClientTransfer(t, testClientTransferParams{
 191                 ExportClientStatus: true,
 192                 LeecherStorage: NewFileCacheClientStorageFactory(FileCacheClientStorageFactoryParams{
 193                         Wrapper: fileCachePieceResourceStorage,
 194                 }),
 195         })
 196 }
 197
 198 func TestClientTransferRateLimitedUpload(t *testing.T) {
 199         started := time.Now()
 200         testClientTransfer(t, testClientTransferParams{
 201                 // We are uploading 13 bytes (the length of the greeting torrent). The
 202                 // chunks are 2 bytes in length. Then the smallest burst we can run
 203                 // with is 2. Time taken is (13-burst)/rate.
 204                 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
 205                 ExportClientStatus:      true,
 206         })
 207         require.True(t, time.Since(started) > time.Second)
 208 }
 209
 210 func TestClientTransferRateLimitedDownload(t *testing.T) {
 211         testClientTransfer(t, testClientTransferParams{
 212                 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
 213         })
 214 }
 215
 216 func fileCachePieceResourceStorage(fc *filecache.Cache) storage.ClientImpl {
 217         return storage.NewResourcePieces(fc.AsResourceProvider())
 218 }
 219
 220 func TestClientTransferSmallCache(t *testing.T) {
 221         testClientTransfer(t, testClientTransferParams{
 222                 LeecherStorage: NewFileCacheClientStorageFactory(FileCacheClientStorageFactoryParams{
 223                         SetCapacity: true,
 224                         // Going below the piece length means it can't complete a piece so
 225                         // that it can be hashed.
 226                         Capacity: 5,
 227                         Wrapper:  fileCachePieceResourceStorage,
 228                 }),
 229                 SetReadahead: true,
 230                 // Can't readahead too far or the cache will thrash and drop data we
 231                 // thought we had.
 232                 Readahead:          0,
 233                 ExportClientStatus: true,
 234         })
 235 }
 236
 237 func TestClientTransferVarious(t *testing.T) {
 238         // Leecher storage
 239         for _, ls := range []storageFactory{
 240                 NewFileCacheClientStorageFactory(FileCacheClientStorageFactoryParams{
 241                         Wrapper: fileCachePieceResourceStorage,
 242                 }),
 243                 storage.NewBoltDB,
 244         } {
 245                 // Seeder storage
 246                 for _, ss := range []func(string) storage.ClientImpl{
 247                         storage.NewFile,
 248                         storage.NewMMap,
 249                 } {
 250                         for _, responsive := range []bool{false, true} {
 251                                 testClientTransfer(t, testClientTransferParams{
 252                                         Responsive:     responsive,
 253                                         SeederStorage:  ss,
 254                                         LeecherStorage: ls,
 255                                 })
 256                                 for _, readahead := range []int64{-1, 0, 1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 13, 14, 15, 20} {
 257                                         testClientTransfer(t, testClientTransferParams{
 258                                                 SeederStorage:  ss,
 259                                                 Responsive:     responsive,
 260                                                 SetReadahead:   true,
 261                                                 Readahead:      readahead,
 262                                                 LeecherStorage: ls,
 263                                         })
 264                                 }
 265                         }
 266                 }
 267         }
 268 }
 269
 270 type testClientTransferParams struct {
 271         Responsive                 bool
 272         Readahead                  int64
 273         SetReadahead               bool
 274         ExportClientStatus         bool
 275         LeecherStorage             func(string) storage.ClientImpl
 276         SeederStorage              func(string) storage.ClientImpl
 277         SeederUploadRateLimiter    *rate.Limiter
 278         LeecherDownloadRateLimiter *rate.Limiter
 279 }
 280
 281 // Creates a seeder and a leecher, and ensures the data transfers when a read
 282 // is attempted on the leecher.
 283 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
 284         greetingTempDir, mi := testutil.GreetingTestTorrent()
 285         defer os.RemoveAll(greetingTempDir)
 286         // Create seeder and a Torrent.
 287         cfg := TestingConfig()
 288         cfg.Seed = true
 289         if ps.SeederUploadRateLimiter != nil {
 290                 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
 291         }
 292         // cfg.ListenAddr = "localhost:4000"
 293         if ps.SeederStorage != nil {
 294                 cfg.DefaultStorage = ps.SeederStorage(greetingTempDir)
 295                 defer cfg.DefaultStorage.Close()
 296         } else {
 297                 cfg.DataDir = greetingTempDir
 298         }
 299         seeder, err := NewClient(cfg)
 300         require.NoError(t, err)
 301         if ps.ExportClientStatus {
 302                 defer testutil.ExportStatusWriter(seeder, "s")()
 303         }
 304         seederTorrent, _, _ := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 305         // Run a Stats right after Closing the Client. This will trigger the Stats
 306         // panic in #214 caused by RemoteAddr on Closed uTP sockets.
 307         defer seederTorrent.Stats()
 308         defer seeder.Close()
 309         seederTorrent.VerifyData()
 310         // Create leecher and a Torrent.
 311         leecherDataDir, err := ioutil.TempDir("", "")
 312         require.NoError(t, err)
 313         defer os.RemoveAll(leecherDataDir)
 314         cfg = TestingConfig()
 315         if ps.LeecherStorage == nil {
 316                 cfg.DataDir = leecherDataDir
 317         } else {
 318                 cfg.DefaultStorage = ps.LeecherStorage(leecherDataDir)
 319         }
 320         if ps.LeecherDownloadRateLimiter != nil {
 321                 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
 322         }
 323         cfg.Seed = false
 324         leecher, err := NewClient(cfg)
 325         require.NoError(t, err)
 326         defer leecher.Close()
 327         if ps.ExportClientStatus {
 328                 defer testutil.ExportStatusWriter(leecher, "l")()
 329         }
 330         leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 331                 ret = TorrentSpecFromMetaInfo(mi)
 332                 ret.ChunkSize = 2
 333                 return
 334         }())
 335         require.NoError(t, err)
 336         assert.True(t, new)
 337         // Now do some things with leecher and seeder.
 338         leecherTorrent.AddClientPeer(seeder)
 339         // The Torrent should not be interested in obtaining peers, so the one we
 340         // just added should be the only one.
 341         assert.False(t, leecherTorrent.Seeding())
 342         assert.EqualValues(t, 1, leecherTorrent.Stats().PendingPeers)
 343         r := leecherTorrent.NewReader()
 344         defer r.Close()
 345         if ps.Responsive {
 346                 r.SetResponsive()
 347         }
 348         if ps.SetReadahead {
 349                 r.SetReadahead(ps.Readahead)
 350         }
 351         assertReadAllGreeting(t, r)
 352
 353         seederStats := seederTorrent.Stats()
 354         assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
 355         assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
 356
 357         leecherStats := leecherTorrent.Stats()
 358         assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
 359         assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
 360
 361         // Try reading through again for the cases where the torrent data size
 362         // exceeds the size of the cache.
 363         assertReadAllGreeting(t, r)
 364 }
 365
 366 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
 367         pos, err := r.Seek(0, io.SeekStart)
 368         assert.NoError(t, err)
 369         assert.EqualValues(t, 0, pos)
 370         _greeting, err := ioutil.ReadAll(r)
 371         assert.NoError(t, err)
 372         assert.EqualValues(t, testutil.GreetingFileContents, _greeting)
 373 }
 374
 375 // Check that after completing leeching, a leecher transitions to a seeding
 376 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
 377 func TestSeedAfterDownloading(t *testing.T) {
 378         greetingTempDir, mi := testutil.GreetingTestTorrent()
 379         defer os.RemoveAll(greetingTempDir)
 380
 381         cfg := TestingConfig()
 382         cfg.Seed = true
 383         cfg.DataDir = greetingTempDir
 384         seeder, err := NewClient(cfg)
 385         require.NoError(t, err)
 386         defer seeder.Close()
 387         defer testutil.ExportStatusWriter(seeder, "s")()
 388         seederTorrent, ok, err := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 389         require.NoError(t, err)
 390         assert.True(t, ok)
 391         seederTorrent.VerifyData()
 392
 393         cfg = TestingConfig()
 394         cfg.Seed = true
 395         cfg.DataDir, err = ioutil.TempDir("", "")
 396         require.NoError(t, err)
 397         defer os.RemoveAll(cfg.DataDir)
 398         leecher, err := NewClient(cfg)
 399         require.NoError(t, err)
 400         defer leecher.Close()
 401         defer testutil.ExportStatusWriter(leecher, "l")()
 402
 403         cfg = TestingConfig()
 404         cfg.Seed = false
 405         cfg.DataDir, err = ioutil.TempDir("", "")
 406         require.NoError(t, err)
 407         defer os.RemoveAll(cfg.DataDir)
 408         leecherLeecher, _ := NewClient(cfg)
 409         require.NoError(t, err)
 410         defer leecherLeecher.Close()
 411         defer testutil.ExportStatusWriter(leecherLeecher, "ll")()
 412         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 413                 ret = TorrentSpecFromMetaInfo(mi)
 414                 ret.ChunkSize = 2
 415                 return
 416         }())
 417         require.NoError(t, err)
 418         assert.True(t, ok)
 419         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 420                 ret = TorrentSpecFromMetaInfo(mi)
 421                 ret.ChunkSize = 3
 422                 return
 423         }())
 424         require.NoError(t, err)
 425         assert.True(t, ok)
 426         // Simultaneously DownloadAll in Leecher, and read the contents
 427         // consecutively in LeecherLeecher. This non-deterministically triggered a
 428         // case where the leecher wouldn't unchoke the LeecherLeecher.
 429         var wg sync.WaitGroup
 430         wg.Add(1)
 431         go func() {
 432                 defer wg.Done()
 433                 r := llg.NewReader()
 434                 defer r.Close()
 435                 b, err := ioutil.ReadAll(r)
 436                 require.NoError(t, err)
 437                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 438         }()
 439         done := make(chan struct{})
 440         defer close(done)
 441         go leecherGreeting.AddClientPeer(seeder)
 442         go leecherGreeting.AddClientPeer(leecherLeecher)
 443         wg.Add(1)
 444         go func() {
 445                 defer wg.Done()
 446                 leecherGreeting.DownloadAll()
 447                 leecher.WaitAll()
 448         }()
 449         wg.Wait()
 450 }
 451
 452 func TestMergingTrackersByAddingSpecs(t *testing.T) {
 453         cl, err := NewClient(TestingConfig())
 454         require.NoError(t, err)
 455         defer cl.Close()
 456         spec := TorrentSpec{}
 457         T, new, _ := cl.AddTorrentSpec(&spec)
 458         if !new {
 459                 t.FailNow()
 460         }
 461         spec.Trackers = [][]string{{"http://a"}, {"udp://b"}}
 462         _, new, _ = cl.AddTorrentSpec(&spec)
 463         assert.False(t, new)
 464         assert.EqualValues(t, [][]string{{"http://a"}, {"udp://b"}}, T.metainfo.AnnounceList)
 465         // Because trackers are disabled in TestingConfig.
 466         assert.EqualValues(t, 0, len(T.trackerAnnouncers))
 467 }
 468
 469 // We read from a piece which is marked completed, but is missing data.
 470 func TestCompletedPieceWrongSize(t *testing.T) {
 471         cfg := TestingConfig()
 472         cfg.DefaultStorage = badStorage{}
 473         cl, err := NewClient(cfg)
 474         require.NoError(t, err)
 475         defer cl.Close()
 476         info := metainfo.Info{
 477                 PieceLength: 15,
 478                 Pieces:      make([]byte, 20),
 479                 Files: []metainfo.FileInfo{
 480                         {Path: []string{"greeting"}, Length: 13},
 481                 },
 482         }
 483         b, err := bencode.Marshal(info)
 484         require.NoError(t, err)
 485         tt, new, err := cl.AddTorrentSpec(&TorrentSpec{
 486                 InfoBytes: b,
 487                 InfoHash:  metainfo.HashBytes(b),
 488         })
 489         require.NoError(t, err)
 490         defer tt.Drop()
 491         assert.True(t, new)
 492         r := tt.NewReader()
 493         defer r.Close()
 494         b, err = ioutil.ReadAll(r)
 495         assert.Len(t, b, 13)
 496         assert.NoError(t, err)
 497 }
 498
 499 func BenchmarkAddLargeTorrent(b *testing.B) {
 500         cfg := TestingConfig()
 501         cfg.DisableTCP = true
 502         cfg.DisableUTP = true
 503         cl, err := NewClient(cfg)
 504         require.NoError(b, err)
 505         defer cl.Close()
 506         b.ReportAllocs()
 507         for range iter.N(b.N) {
 508                 t, err := cl.AddTorrentFromFile("testdata/bootstrap.dat.torrent")
 509                 if err != nil {
 510                         b.Fatal(err)
 511                 }
 512                 t.Drop()
 513         }
 514 }
 515
 516 func TestResponsive(t *testing.T) {
 517         seederDataDir, mi := testutil.GreetingTestTorrent()
 518         defer os.RemoveAll(seederDataDir)
 519         cfg := TestingConfig()
 520         cfg.Seed = true
 521         cfg.DataDir = seederDataDir
 522         seeder, err := NewClient(cfg)
 523         require.Nil(t, err)
 524         defer seeder.Close()
 525         seederTorrent, _, _ := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 526         seederTorrent.VerifyData()
 527         leecherDataDir, err := ioutil.TempDir("", "")
 528         require.Nil(t, err)
 529         defer os.RemoveAll(leecherDataDir)
 530         cfg = TestingConfig()
 531         cfg.DataDir = leecherDataDir
 532         leecher, err := NewClient(cfg)
 533         require.Nil(t, err)
 534         defer leecher.Close()
 535         leecherTorrent, _, _ := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 536                 ret = TorrentSpecFromMetaInfo(mi)
 537                 ret.ChunkSize = 2
 538                 return
 539         }())
 540         leecherTorrent.AddClientPeer(seeder)
 541         reader := leecherTorrent.NewReader()
 542         defer reader.Close()
 543         reader.SetReadahead(0)
 544         reader.SetResponsive()
 545         b := make([]byte, 2)
 546         _, err = reader.Seek(3, io.SeekStart)
 547         require.NoError(t, err)
 548         _, err = io.ReadFull(reader, b)
 549         assert.Nil(t, err)
 550         assert.EqualValues(t, "lo", string(b))
 551         _, err = reader.Seek(11, io.SeekStart)
 552         require.NoError(t, err)
 553         n, err := io.ReadFull(reader, b)
 554         assert.Nil(t, err)
 555         assert.EqualValues(t, 2, n)
 556         assert.EqualValues(t, "d\n", string(b))
 557 }
 558
 559 func TestTorrentDroppedDuringResponsiveRead(t *testing.T) {
 560         seederDataDir, mi := testutil.GreetingTestTorrent()
 561         defer os.RemoveAll(seederDataDir)
 562         cfg := TestingConfig()
 563         cfg.Seed = true
 564         cfg.DataDir = seederDataDir
 565         seeder, err := NewClient(cfg)
 566         require.Nil(t, err)
 567         defer seeder.Close()
 568         seederTorrent, _, _ := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 569         seederTorrent.VerifyData()
 570         leecherDataDir, err := ioutil.TempDir("", "")
 571         require.Nil(t, err)
 572         defer os.RemoveAll(leecherDataDir)
 573         cfg = TestingConfig()
 574         cfg.DataDir = leecherDataDir
 575         leecher, err := NewClient(cfg)
 576         require.Nil(t, err)
 577         defer leecher.Close()
 578         leecherTorrent, _, _ := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 579                 ret = TorrentSpecFromMetaInfo(mi)
 580                 ret.ChunkSize = 2
 581                 return
 582         }())
 583         leecherTorrent.AddClientPeer(seeder)
 584         reader := leecherTorrent.NewReader()
 585         defer reader.Close()
 586         reader.SetReadahead(0)
 587         reader.SetResponsive()
 588         b := make([]byte, 2)
 589         _, err = reader.Seek(3, io.SeekStart)
 590         require.NoError(t, err)
 591         _, err = io.ReadFull(reader, b)
 592         assert.Nil(t, err)
 593         assert.EqualValues(t, "lo", string(b))
 594         go leecherTorrent.Drop()
 595         _, err = reader.Seek(11, io.SeekStart)
 596         require.NoError(t, err)
 597         n, err := reader.Read(b)
 598         assert.EqualError(t, err, "torrent closed")
 599         assert.EqualValues(t, 0, n)
 600 }
 601
 602 func TestDHTInheritBlocklist(t *testing.T) {
 603         ipl := iplist.New(nil)
 604         require.NotNil(t, ipl)
 605         cfg := TestingConfig()
 606         cfg.IPBlocklist = ipl
 607         cfg.NoDHT = false
 608         cl, err := NewClient(cfg)
 609         require.NoError(t, err)
 610         defer cl.Close()
 611         numServers := 0
 612         cl.eachDhtServer(func(s *dht.Server) {
 613                 assert.Equal(t, ipl, s.IPBlocklist())
 614                 numServers++
 615         })
 616         assert.EqualValues(t, 2, numServers)
 617 }
 618
 619 // Check that stuff is merged in subsequent AddTorrentSpec for the same
 620 // infohash.
 621 func TestAddTorrentSpecMerging(t *testing.T) {
 622         cl, err := NewClient(TestingConfig())
 623         require.NoError(t, err)
 624         defer cl.Close()
 625         dir, mi := testutil.GreetingTestTorrent()
 626         defer os.RemoveAll(dir)
 627         tt, new, err := cl.AddTorrentSpec(&TorrentSpec{
 628                 InfoHash: mi.HashInfoBytes(),
 629         })
 630         require.NoError(t, err)
 631         require.True(t, new)
 632         require.Nil(t, tt.Info())
 633         _, new, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 634         require.NoError(t, err)
 635         require.False(t, new)
 636         require.NotNil(t, tt.Info())
 637 }
 638
 639 func TestTorrentDroppedBeforeGotInfo(t *testing.T) {
 640         dir, mi := testutil.GreetingTestTorrent()
 641         os.RemoveAll(dir)
 642         cl, _ := NewClient(TestingConfig())
 643         defer cl.Close()
 644         tt, _, _ := cl.AddTorrentSpec(&TorrentSpec{
 645                 InfoHash: mi.HashInfoBytes(),
 646         })
 647         tt.Drop()
 648         assert.EqualValues(t, 0, len(cl.Torrents()))
 649         select {
 650         case <-tt.GotInfo():
 651                 t.FailNow()
 652         default:
 653         }
 654 }
 655
 656 func writeTorrentData(ts *storage.Torrent, info metainfo.Info, b []byte) {
 657         for i := range iter.N(info.NumPieces()) {
 658                 p := info.Piece(i)
 659                 ts.Piece(p).WriteAt(b[p.Offset():p.Offset()+p.Length()], 0)
 660         }
 661 }
 662
 663 func testAddTorrentPriorPieceCompletion(t *testing.T, alreadyCompleted bool, csf func(*filecache.Cache) storage.ClientImpl) {
 664         fileCacheDir, err := ioutil.TempDir("", "")
 665         require.NoError(t, err)
 666         defer os.RemoveAll(fileCacheDir)
 667         fileCache, err := filecache.NewCache(fileCacheDir)
 668         require.NoError(t, err)
 669         greetingDataTempDir, greetingMetainfo := testutil.GreetingTestTorrent()
 670         defer os.RemoveAll(greetingDataTempDir)
 671         filePieceStore := csf(fileCache)
 672         defer filePieceStore.Close()
 673         info, err := greetingMetainfo.UnmarshalInfo()
 674         require.NoError(t, err)
 675         ih := greetingMetainfo.HashInfoBytes()
 676         greetingData, err := storage.NewClient(filePieceStore).OpenTorrent(&info, ih)
 677         require.NoError(t, err)
 678         writeTorrentData(greetingData, info, []byte(testutil.GreetingFileContents))
 679         // require.Equal(t, len(testutil.GreetingFileContents), written)
 680         // require.NoError(t, err)
 681         for i := 0; i < info.NumPieces(); i++ {
 682                 p := info.Piece(i)
 683                 if alreadyCompleted {
 684                         require.NoError(t, greetingData.Piece(p).MarkComplete())
 685                 }
 686         }
 687         cfg := TestingConfig()
 688         // TODO: Disable network option?
 689         cfg.DisableTCP = true
 690         cfg.DisableUTP = true
 691         cfg.DefaultStorage = filePieceStore
 692         cl, err := NewClient(cfg)
 693         require.NoError(t, err)
 694         defer cl.Close()
 695         tt, err := cl.AddTorrent(greetingMetainfo)
 696         require.NoError(t, err)
 697         psrs := tt.PieceStateRuns()
 698         assert.Len(t, psrs, 1)
 699         assert.EqualValues(t, 3, psrs[0].Length)
 700         assert.Equal(t, alreadyCompleted, psrs[0].Complete)
 701         if alreadyCompleted {
 702                 r := tt.NewReader()
 703                 b, err := ioutil.ReadAll(r)
 704                 assert.NoError(t, err)
 705                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 706         }
 707 }
 708
 709 func TestAddTorrentPiecesAlreadyCompleted(t *testing.T) {
 710         testAddTorrentPriorPieceCompletion(t, true, fileCachePieceResourceStorage)
 711 }
 712
 713 func TestAddTorrentPiecesNotAlreadyCompleted(t *testing.T) {
 714         testAddTorrentPriorPieceCompletion(t, false, fileCachePieceResourceStorage)
 715 }
 716
 717 func TestAddMetainfoWithNodes(t *testing.T) {
 718         cfg := TestingConfig()
 719         cfg.ListenHost = func(string) string { return "" }
 720         cfg.NoDHT = false
 721         cfg.DhtStartingNodes = func() ([]dht.Addr, error) { return nil, nil }
 722         // For now, we want to just jam the nodes into the table, without
 723         // verifying them first. Also the DHT code doesn't support mixing secure
 724         // and insecure nodes if security is enabled (yet).
 725         // cfg.DHTConfig.NoSecurity = true
 726         cl, err := NewClient(cfg)
 727         require.NoError(t, err)
 728         defer cl.Close()
 729         sum := func() (ret int64) {
 730                 cl.eachDhtServer(func(s *dht.Server) {
 731                         ret += s.Stats().OutboundQueriesAttempted
 732                 })
 733                 return
 734         }
 735         assert.EqualValues(t, 0, sum())
 736         tt, err := cl.AddTorrentFromFile("metainfo/testdata/issue_65a.torrent")
 737         require.NoError(t, err)
 738         // Nodes are not added or exposed in Torrent's metainfo. We just randomly
 739         // check if the announce-list is here instead. TODO: Add nodes.
 740         assert.Len(t, tt.metainfo.AnnounceList, 5)
 741         // There are 6 nodes in the torrent file.
 742         for sum() != int64(6*len(cl.dhtServers)) {
 743                 time.Sleep(time.Millisecond)
 744         }
 745 }
 746
 747 type testDownloadCancelParams struct {
 748         SetLeecherStorageCapacity bool
 749         LeecherStorageCapacity    int64
 750         Cancel                    bool
 751 }
 752
 753 func testDownloadCancel(t *testing.T, ps testDownloadCancelParams) {
 754         greetingTempDir, mi := testutil.GreetingTestTorrent()
 755         defer os.RemoveAll(greetingTempDir)
 756         cfg := TestingConfig()
 757         cfg.Seed = true
 758         cfg.DataDir = greetingTempDir
 759         seeder, err := NewClient(cfg)
 760         require.NoError(t, err)
 761         defer seeder.Close()
 762         defer testutil.ExportStatusWriter(seeder, "s")()
 763         seederTorrent, _, _ := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 764         seederTorrent.VerifyData()
 765         leecherDataDir, err := ioutil.TempDir("", "")
 766         require.NoError(t, err)
 767         defer os.RemoveAll(leecherDataDir)
 768         fc, err := filecache.NewCache(leecherDataDir)
 769         require.NoError(t, err)
 770         if ps.SetLeecherStorageCapacity {
 771                 fc.SetCapacity(ps.LeecherStorageCapacity)
 772         }
 773         cfg.DefaultStorage = storage.NewResourcePieces(fc.AsResourceProvider())
 774         cfg.DataDir = leecherDataDir
 775         leecher, err := NewClient(cfg)
 776         require.NoError(t, err)
 777         defer leecher.Close()
 778         defer testutil.ExportStatusWriter(leecher, "l")()
 779         leecherGreeting, new, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 780                 ret = TorrentSpecFromMetaInfo(mi)
 781                 ret.ChunkSize = 2
 782                 return
 783         }())
 784         require.NoError(t, err)
 785         assert.True(t, new)
 786         psc := leecherGreeting.SubscribePieceStateChanges()
 787         defer psc.Close()
 788
 789         leecherGreeting.cl.lock()
 790         leecherGreeting.downloadPiecesLocked(0, leecherGreeting.numPieces())
 791         if ps.Cancel {
 792                 leecherGreeting.cancelPiecesLocked(0, leecherGreeting.NumPieces())
 793         }
 794         leecherGreeting.cl.unlock()
 795         done := make(chan struct{})
 796         defer close(done)
 797         go leecherGreeting.AddClientPeer(seeder)
 798         completes := make(map[int]bool, 3)
 799         expected := func() map[int]bool {
 800                 if ps.Cancel {
 801                         return map[int]bool{0: false, 1: false, 2: false}
 802                 } else {
 803                         return map[int]bool{0: true, 1: true, 2: true}
 804                 }
 805         }()
 806         for !reflect.DeepEqual(completes, expected) {
 807                 _v := <-psc.Values
 808                 v := _v.(PieceStateChange)
 809                 completes[v.Index] = v.Complete
 810         }
 811 }
 812
 813 func TestTorrentDownloadAll(t *testing.T) {
 814         testDownloadCancel(t, testDownloadCancelParams{})
 815 }
 816
 817 func TestTorrentDownloadAllThenCancel(t *testing.T) {
 818         testDownloadCancel(t, testDownloadCancelParams{
 819                 Cancel: true,
 820         })
 821 }
 822
 823 // Ensure that it's an error for a peer to send an invalid have message.
 824 func TestPeerInvalidHave(t *testing.T) {
 825         cl, err := NewClient(TestingConfig())
 826         require.NoError(t, err)
 827         defer cl.Close()
 828         info := metainfo.Info{
 829                 PieceLength: 1,
 830                 Pieces:      make([]byte, 20),
 831                 Files:       []metainfo.FileInfo{{Length: 1}},
 832         }
 833         infoBytes, err := bencode.Marshal(info)
 834         require.NoError(t, err)
 835         tt, _new, err := cl.AddTorrentSpec(&TorrentSpec{
 836                 InfoBytes: infoBytes,
 837                 InfoHash:  metainfo.HashBytes(infoBytes),
 838                 Storage:   badStorage{},
 839         })
 840         require.NoError(t, err)
 841         assert.True(t, _new)
 842         defer tt.Drop()
 843         cn := &connection{
 844                 t: tt,
 845         }
 846         assert.NoError(t, cn.peerSentHave(0))
 847         assert.Error(t, cn.peerSentHave(1))
 848 }
 849
 850 func TestPieceCompletedInStorageButNotClient(t *testing.T) {
 851         greetingTempDir, greetingMetainfo := testutil.GreetingTestTorrent()
 852         defer os.RemoveAll(greetingTempDir)
 853         cfg := TestingConfig()
 854         cfg.DataDir = greetingTempDir
 855         seeder, err := NewClient(TestingConfig())
 856         require.NoError(t, err)
 857         seeder.AddTorrentSpec(&TorrentSpec{
 858                 InfoBytes: greetingMetainfo.InfoBytes,
 859         })
 860 }
 861
 862 // Check that when the listen port is 0, all the protocols listened on have
 863 // the same port, and it isn't zero.
 864 func TestClientDynamicListenPortAllProtocols(t *testing.T) {
 865         cl, err := NewClient(TestingConfig())
 866         require.NoError(t, err)
 867         defer cl.Close()
 868         port := cl.LocalPort()
 869         assert.NotEqual(t, 0, port)
 870         cl.eachListener(func(s socket) bool {
 871                 assert.Equal(t, port, missinggo.AddrPort(s.Addr()))
 872                 return true
 873         })
 874 }
 875
 876 func TestClientDynamicListenTCPOnly(t *testing.T) {
 877         cfg := TestingConfig()
 878         cfg.DisableUTP = true
 879         cfg.DisableTCP = false
 880         cl, err := NewClient(cfg)
 881         require.NoError(t, err)
 882         defer cl.Close()
 883         assert.NotEqual(t, 0, cl.LocalPort())
 884 }
 885
 886 func TestClientDynamicListenUTPOnly(t *testing.T) {
 887         cfg := TestingConfig()
 888         cfg.DisableTCP = true
 889         cfg.DisableUTP = false
 890         cl, err := NewClient(cfg)
 891         require.NoError(t, err)
 892         defer cl.Close()
 893         assert.NotEqual(t, 0, cl.LocalPort())
 894 }
 895
 896 func totalConns(tts []*Torrent) (ret int) {
 897         for _, tt := range tts {
 898                 tt.cl.lock()
 899                 ret += len(tt.conns)
 900                 tt.cl.unlock()
 901         }
 902         return
 903 }
 904
 905 func TestSetMaxEstablishedConn(t *testing.T) {
 906         var tts []*Torrent
 907         ih := testutil.GreetingMetaInfo().HashInfoBytes()
 908         cfg := TestingConfig()
 909         cfg.DisableAcceptRateLimiting = true
 910         cfg.dropDuplicatePeerIds = true
 911         for i := range iter.N(3) {
 912                 cl, err := NewClient(cfg)
 913                 require.NoError(t, err)
 914                 defer cl.Close()
 915                 tt, _ := cl.AddTorrentInfoHash(ih)
 916                 tt.SetMaxEstablishedConns(2)
 917                 defer testutil.ExportStatusWriter(cl, fmt.Sprintf("%d", i))()
 918                 tts = append(tts, tt)
 919         }
 920         addPeers := func() {
 921                 for _, tt := range tts {
 922                         for _, _tt := range tts {
 923                                 // if tt != _tt {
 924                                 tt.AddClientPeer(_tt.cl)
 925                                 // }
 926                         }
 927                 }
 928         }
 929         waitTotalConns := func(num int) {
 930                 for totalConns(tts) != num {
 931                         addPeers()
 932                         time.Sleep(time.Millisecond)
 933                 }
 934         }
 935         addPeers()
 936         waitTotalConns(6)
 937         tts[0].SetMaxEstablishedConns(1)
 938         waitTotalConns(4)
 939         tts[0].SetMaxEstablishedConns(0)
 940         waitTotalConns(2)
 941         tts[0].SetMaxEstablishedConns(1)
 942         addPeers()
 943         waitTotalConns(4)
 944         tts[0].SetMaxEstablishedConns(2)
 945         addPeers()
 946         waitTotalConns(6)
 947 }
 948
 949 // Creates a file containing its own name as data. Make a metainfo from that, adds it to the given
 950 // client, and returns a magnet link.
 951 func makeMagnet(t *testing.T, cl *Client, dir string, name string) string {
 952         os.MkdirAll(dir, 0770)
 953         file, err := os.Create(filepath.Join(dir, name))
 954         require.NoError(t, err)
 955         file.Write([]byte(name))
 956         file.Close()
 957         mi := metainfo.MetaInfo{}
 958         mi.SetDefaults()
 959         info := metainfo.Info{PieceLength: 256 * 1024}
 960         err = info.BuildFromFilePath(filepath.Join(dir, name))
 961         require.NoError(t, err)
 962         mi.InfoBytes, err = bencode.Marshal(info)
 963         require.NoError(t, err)
 964         magnet := mi.Magnet(name, mi.HashInfoBytes()).String()
 965         tr, err := cl.AddTorrent(&mi)
 966         require.NoError(t, err)
 967         require.True(t, tr.Seeding())
 968         tr.VerifyData()
 969         return magnet
 970 }
 971
 972 // https://github.com/anacrolix/torrent/issues/114
 973 func TestMultipleTorrentsWithEncryption(t *testing.T) {
 974         testSeederLeecherPair(
 975                 t,
 976                 func(cfg *ClientConfig) {
 977                         cfg.HeaderObfuscationPolicy.Preferred = true
 978                         cfg.HeaderObfuscationPolicy.RequirePreferred = true
 979                 },
 980                 func(cfg *ClientConfig) {
 981                         cfg.HeaderObfuscationPolicy.RequirePreferred = false
 982                 },
 983         )
 984 }
 985
 986 // Test that the leecher can download a torrent in its entirety from the seeder. Note that the
 987 // seeder config is done first.
 988 func testSeederLeecherPair(t *testing.T, seeder func(*ClientConfig), leecher func(*ClientConfig)) {
 989         cfg := TestingConfig()
 990         cfg.Seed = true
 991         cfg.DataDir = filepath.Join(cfg.DataDir, "server")
 992         os.Mkdir(cfg.DataDir, 0755)
 993         seeder(cfg)
 994         server, err := NewClient(cfg)
 995         require.NoError(t, err)
 996         defer server.Close()
 997         defer testutil.ExportStatusWriter(server, "s")()
 998         magnet1 := makeMagnet(t, server, cfg.DataDir, "test1")
 999         // Extra torrents are added to test the seeder having to match incoming obfuscated headers
1000         // against more than one torrent. See issue #114
1001         makeMagnet(t, server, cfg.DataDir, "test2")
1002         for i := 0; i < 100; i++ {
1003                 makeMagnet(t, server, cfg.DataDir, fmt.Sprintf("test%d", i+2))
1004         }
1005         cfg = TestingConfig()
1006         cfg.DataDir = filepath.Join(cfg.DataDir, "client")
1007         leecher(cfg)
1008         client, err := NewClient(cfg)
1009         require.NoError(t, err)
1010         defer client.Close()
1011         defer testutil.ExportStatusWriter(client, "c")()
1012         tr, err := client.AddMagnet(magnet1)
1013         require.NoError(t, err)
1014         tr.AddClientPeer(server)
1015         <-tr.GotInfo()
1016         tr.DownloadAll()
1017         client.WaitAll()
1018 }
1019
1020 // This appears to be the situation with the S3 BitTorrent client.
1021 func TestObfuscatedHeaderFallbackSeederDisallowsLeecherPrefers(t *testing.T) {
1022         // Leecher prefers obfuscation, but the seeder does not allow it.
1023         testSeederLeecherPair(
1024                 t,
1025                 func(cfg *ClientConfig) {
1026                         cfg.HeaderObfuscationPolicy.Preferred = false
1027                         cfg.HeaderObfuscationPolicy.RequirePreferred = true
1028                 },
1029                 func(cfg *ClientConfig) {
1030                         cfg.HeaderObfuscationPolicy.Preferred = true
1031                         cfg.HeaderObfuscationPolicy.RequirePreferred = false
1032                 },
1033         )
1034 }
1035
1036 func TestObfuscatedHeaderFallbackSeederRequiresLeecherPrefersNot(t *testing.T) {
1037         // Leecher prefers no obfuscation, but the seeder enforces it.
1038         testSeederLeecherPair(
1039                 t,
1040                 func(cfg *ClientConfig) {
1041                         cfg.HeaderObfuscationPolicy.Preferred = true
1042                         cfg.HeaderObfuscationPolicy.RequirePreferred = true
1043                 },
1044                 func(cfg *ClientConfig) {
1045                         cfg.HeaderObfuscationPolicy.Preferred = false
1046                         cfg.HeaderObfuscationPolicy.RequirePreferred = false
1047                 },
1048         )
1049 }
1050
1051 func TestClientAddressInUse(t *testing.T) {
1052         s, _ := NewUtpSocket("udp", ":50007", nil)
1053         if s != nil {
1054                 defer s.Close()
1055         }
1056         cfg := TestingConfig().SetListenAddr(":50007")
1057         cl, err := NewClient(cfg)
1058         require.Error(t, err)
1059         require.Nil(t, cl)
1060 }
1061
1062 func TestClientHasDhtServersWhenUtpDisabled(t *testing.T) {
1063         cc := TestingConfig()
1064         cc.DisableUTP = true
1065         cc.NoDHT = false
1066         cl, err := NewClient(cc)
1067         require.NoError(t, err)
1068         defer cl.Close()
1069         assert.NotEmpty(t, cl.DhtServers())
1070 }