client_test.go

   1 package torrent
   2
   3 import (
   4         "encoding/binary"
   5         "errors"
   6         "fmt"
   7         "io"
   8         "io/ioutil"
   9         "log"
  10         "math/rand"
  11         "net"
  12         "os"
  13         "path/filepath"
  14         "strings"
  15         "sync"
  16         "testing"
  17         "time"
  18
  19         _ "github.com/anacrolix/envpprof"
  20         "github.com/anacrolix/missinggo"
  21         "github.com/anacrolix/missinggo/filecache"
  22         "github.com/anacrolix/utp"
  23         "github.com/bradfitz/iter"
  24         "github.com/stretchr/testify/assert"
  25         "github.com/stretchr/testify/require"
  26
  27         "github.com/anacrolix/torrent/bencode"
  28         "github.com/anacrolix/torrent/dht"
  29         "github.com/anacrolix/torrent/internal/testutil"
  30         "github.com/anacrolix/torrent/iplist"
  31         "github.com/anacrolix/torrent/metainfo"
  32         "github.com/anacrolix/torrent/storage"
  33 )
  34
  35 func init() {
  36         log.SetFlags(log.LstdFlags | log.Llongfile)
  37 }
  38
  39 var TestingConfig = Config{
  40         ListenAddr:           "localhost:0",
  41         NoDHT:                true,
  42         DisableTrackers:      true,
  43         DisableMetainfoCache: true,
  44         DataDir:              "/dev/null",
  45         DHTConfig: dht.ServerConfig{
  46                 NoDefaultBootstrap: true,
  47         },
  48 }
  49
  50 func TestClientDefault(t *testing.T) {
  51         cl, err := NewClient(&TestingConfig)
  52         require.NoError(t, err)
  53         cl.Close()
  54 }
  55
  56 func TestAddDropTorrent(t *testing.T) {
  57         cl, err := NewClient(&TestingConfig)
  58         require.NoError(t, err)
  59         defer cl.Close()
  60         dir, mi := testutil.GreetingTestTorrent()
  61         defer os.RemoveAll(dir)
  62         tt, new, err := cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
  63         require.NoError(t, err)
  64         assert.True(t, new)
  65         tt.Drop()
  66 }
  67
  68 func TestAddTorrentNoSupportedTrackerSchemes(t *testing.T) {
  69         t.SkipNow()
  70 }
  71
  72 func TestAddTorrentNoUsableURLs(t *testing.T) {
  73         t.SkipNow()
  74 }
  75
  76 func TestAddPeersToUnknownTorrent(t *testing.T) {
  77         t.SkipNow()
  78 }
  79
  80 func TestPieceHashSize(t *testing.T) {
  81         if pieceHash.Size() != 20 {
  82                 t.FailNow()
  83         }
  84 }
  85
  86 func TestTorrentInitialState(t *testing.T) {
  87         dir, mi := testutil.GreetingTestTorrent()
  88         defer os.RemoveAll(dir)
  89         tor := newTorrent(func() (ih metainfo.Hash) {
  90                 missinggo.CopyExact(ih[:], mi.Info.Hash)
  91                 return
  92         }())
  93         tor.chunkSize = 2
  94         tor.storageOpener = storage.NewFile(dir)
  95         // Needed to lock for asynchronous piece verification.
  96         tor.cl = new(Client)
  97         err := tor.setMetadata(&mi.Info.Info, mi.Info.Bytes)
  98         require.NoError(t, err)
  99         require.Len(t, tor.pieces, 3)
 100         tor.pendAllChunkSpecs(0)
 101         assert.EqualValues(t, 3, tor.pieceNumPendingChunks(0))
 102         assert.EqualValues(t, chunkSpec{4, 1}, chunkIndexSpec(2, tor.pieceLength(0), tor.chunkSize))
 103 }
 104
 105 func TestUnmarshalPEXMsg(t *testing.T) {
 106         var m peerExchangeMessage
 107         if err := bencode.Unmarshal([]byte("d5:added12:\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0ce"), &m); err != nil {
 108                 t.Fatal(err)
 109         }
 110         if len(m.Added) != 2 {
 111                 t.FailNow()
 112         }
 113         if m.Added[0].Port != 0x506 {
 114                 t.FailNow()
 115         }
 116 }
 117
 118 func TestReducedDialTimeout(t *testing.T) {
 119         for _, _case := range []struct {
 120                 Max             time.Duration
 121                 HalfOpenLimit   int
 122                 PendingPeers    int
 123                 ExpectedReduced time.Duration
 124         }{
 125                 {nominalDialTimeout, 40, 0, nominalDialTimeout},
 126                 {nominalDialTimeout, 40, 1, nominalDialTimeout},
 127                 {nominalDialTimeout, 40, 39, nominalDialTimeout},
 128                 {nominalDialTimeout, 40, 40, nominalDialTimeout / 2},
 129                 {nominalDialTimeout, 40, 80, nominalDialTimeout / 3},
 130                 {nominalDialTimeout, 40, 4000, nominalDialTimeout / 101},
 131         } {
 132                 reduced := reducedDialTimeout(_case.Max, _case.HalfOpenLimit, _case.PendingPeers)
 133                 expected := _case.ExpectedReduced
 134                 if expected < minDialTimeout {
 135                         expected = minDialTimeout
 136                 }
 137                 if reduced != expected {
 138                         t.Fatalf("expected %s, got %s", _case.ExpectedReduced, reduced)
 139                 }
 140         }
 141 }
 142
 143 func TestUTPRawConn(t *testing.T) {
 144         l, err := utp.NewSocket("udp", "")
 145         if err != nil {
 146                 t.Fatal(err)
 147         }
 148         defer l.Close()
 149         go func() {
 150                 for {
 151                         _, err := l.Accept()
 152                         if err != nil {
 153                                 break
 154                         }
 155                 }
 156         }()
 157         // Connect a UTP peer to see if the RawConn will still work.
 158         s, _ := utp.NewSocket("udp", "")
 159         defer s.Close()
 160         utpPeer, err := s.Dial(fmt.Sprintf("localhost:%d", missinggo.AddrPort(l.Addr())))
 161         if err != nil {
 162                 t.Fatalf("error dialing utp listener: %s", err)
 163         }
 164         defer utpPeer.Close()
 165         peer, err := net.ListenPacket("udp", ":0")
 166         if err != nil {
 167                 t.Fatal(err)
 168         }
 169         defer peer.Close()
 170
 171         msgsReceived := 0
 172         // How many messages to send. I've set this to double the channel buffer
 173         // size in the raw packetConn.
 174         const N = 200
 175         readerStopped := make(chan struct{})
 176         // The reader goroutine.
 177         go func() {
 178                 defer close(readerStopped)
 179                 b := make([]byte, 500)
 180                 for i := 0; i < N; i++ {
 181                         n, _, err := l.ReadFrom(b)
 182                         if err != nil {
 183                                 t.Fatalf("error reading from raw conn: %s", err)
 184                         }
 185                         msgsReceived++
 186                         var d int
 187                         fmt.Sscan(string(b[:n]), &d)
 188                         if d != i {
 189                                 log.Printf("got wrong number: expected %d, got %d", i, d)
 190                         }
 191                 }
 192         }()
 193         udpAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("localhost:%d", missinggo.AddrPort(l.Addr())))
 194         if err != nil {
 195                 t.Fatal(err)
 196         }
 197         for i := 0; i < N; i++ {
 198                 _, err := peer.WriteTo([]byte(fmt.Sprintf("%d", i)), udpAddr)
 199                 if err != nil {
 200                         t.Fatal(err)
 201                 }
 202                 time.Sleep(time.Microsecond)
 203         }
 204         select {
 205         case <-readerStopped:
 206         case <-time.After(time.Second):
 207                 t.Fatal("reader timed out")
 208         }
 209         if msgsReceived != N {
 210                 t.Fatalf("messages received: %d", msgsReceived)
 211         }
 212 }
 213
 214 func TestTwoClientsArbitraryPorts(t *testing.T) {
 215         for i := 0; i < 2; i++ {
 216                 cl, err := NewClient(&TestingConfig)
 217                 if err != nil {
 218                         t.Fatal(err)
 219                 }
 220                 defer cl.Close()
 221         }
 222 }
 223
 224 func TestAddDropManyTorrents(t *testing.T) {
 225         cl, err := NewClient(&TestingConfig)
 226         require.NoError(t, err)
 227         defer cl.Close()
 228         for i := range iter.N(1000) {
 229                 var spec TorrentSpec
 230                 binary.PutVarint(spec.InfoHash[:], int64(i))
 231                 tt, new, err := cl.AddTorrentSpec(&spec)
 232                 assert.NoError(t, err)
 233                 assert.True(t, new)
 234                 defer tt.Drop()
 235         }
 236 }
 237
 238 func TestClientTransferDefault(t *testing.T) {
 239         testClientTransfer(t, testClientTransferParams{
 240                 ExportClientStatus: true,
 241         })
 242 }
 243
 244 func TestClientTransferSmallCache(t *testing.T) {
 245         testClientTransfer(t, testClientTransferParams{
 246                 SetLeecherStorageCapacity: true,
 247                 // Going below the piece length means it can't complete a piece so
 248                 // that it can be hashed.
 249                 LeecherStorageCapacity: 5,
 250                 SetReadahead:           true,
 251                 // Can't readahead too far or the cache will thrash and drop data we
 252                 // thought we had.
 253                 Readahead:          0,
 254                 ExportClientStatus: true,
 255         })
 256 }
 257
 258 func TestClientTransferVarious(t *testing.T) {
 259         for _, ss := range []func(string) storage.I{
 260                 storage.NewFile,
 261                 storage.NewMMap,
 262         } {
 263                 for _, responsive := range []bool{false, true} {
 264                         testClientTransfer(t, testClientTransferParams{
 265                                 Responsive:    responsive,
 266                                 SeederStorage: ss,
 267                         })
 268                         for _, readahead := range []int64{-1, 0, 1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 13, 14, 15, 20} {
 269                                 testClientTransfer(t, testClientTransferParams{
 270                                         SeederStorage: ss,
 271                                         Responsive:    responsive,
 272                                         SetReadahead:  true,
 273                                         Readahead:     readahead,
 274                                 })
 275                         }
 276                 }
 277         }
 278 }
 279
 280 type testClientTransferParams struct {
 281         Responsive                bool
 282         Readahead                 int64
 283         SetReadahead              bool
 284         ExportClientStatus        bool
 285         SetLeecherStorageCapacity bool
 286         LeecherStorageCapacity    int64
 287         SeederStorage             func(string) storage.I
 288 }
 289
 290 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
 291         greetingTempDir, mi := testutil.GreetingTestTorrent()
 292         defer os.RemoveAll(greetingTempDir)
 293         cfg := TestingConfig
 294         cfg.Seed = true
 295         if ps.SeederStorage != nil {
 296                 cfg.DefaultStorage = ps.SeederStorage(greetingTempDir)
 297         } else {
 298                 cfg.DataDir = greetingTempDir
 299         }
 300         seeder, err := NewClient(&cfg)
 301         require.NoError(t, err)
 302         defer seeder.Close()
 303         if ps.ExportClientStatus {
 304                 testutil.ExportStatusWriter(seeder, "s")
 305         }
 306         _, new, err := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 307         require.NoError(t, err)
 308         assert.True(t, new)
 309         leecherDataDir, err := ioutil.TempDir("", "")
 310         require.NoError(t, err)
 311         defer os.RemoveAll(leecherDataDir)
 312         fc, err := filecache.NewCache(leecherDataDir)
 313         require.NoError(t, err)
 314         if ps.SetLeecherStorageCapacity {
 315                 fc.SetCapacity(ps.LeecherStorageCapacity)
 316         }
 317         cfg.DefaultStorage = storage.NewPieceFileStorage(fc.AsFileStore())
 318         leecher, err := NewClient(&cfg)
 319         require.NoError(t, err)
 320         defer leecher.Close()
 321         if ps.ExportClientStatus {
 322                 testutil.ExportStatusWriter(leecher, "l")
 323         }
 324         leecherGreeting, new, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 325                 ret = TorrentSpecFromMetaInfo(mi)
 326                 ret.ChunkSize = 2
 327                 ret.Storage = storage.NewFile(leecherDataDir)
 328                 return
 329         }())
 330         require.NoError(t, err)
 331         assert.True(t, new)
 332         leecherGreeting.AddPeers([]Peer{
 333                 Peer{
 334                         IP:   missinggo.AddrIP(seeder.ListenAddr()),
 335                         Port: missinggo.AddrPort(seeder.ListenAddr()),
 336                 },
 337         })
 338         r := leecherGreeting.NewReader()
 339         defer r.Close()
 340         if ps.Responsive {
 341                 r.SetResponsive()
 342         }
 343         if ps.SetReadahead {
 344                 r.SetReadahead(ps.Readahead)
 345         }
 346         for range iter.N(2) {
 347                 pos, err := r.Seek(0, os.SEEK_SET)
 348                 assert.NoError(t, err)
 349                 assert.EqualValues(t, 0, pos)
 350                 _greeting, err := ioutil.ReadAll(r)
 351                 assert.NoError(t, err)
 352                 assert.EqualValues(t, testutil.GreetingFileContents, _greeting)
 353         }
 354 }
 355
 356 // Check that after completing leeching, a leecher transitions to a seeding
 357 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
 358 func TestSeedAfterDownloading(t *testing.T) {
 359         greetingTempDir, mi := testutil.GreetingTestTorrent()
 360         defer os.RemoveAll(greetingTempDir)
 361         cfg := TestingConfig
 362         cfg.Seed = true
 363         cfg.DataDir = greetingTempDir
 364         seeder, err := NewClient(&cfg)
 365         require.NoError(t, err)
 366         defer seeder.Close()
 367         testutil.ExportStatusWriter(seeder, "s")
 368         seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 369         cfg.DataDir, err = ioutil.TempDir("", "")
 370         require.NoError(t, err)
 371         defer os.RemoveAll(cfg.DataDir)
 372         leecher, err := NewClient(&cfg)
 373         require.NoError(t, err)
 374         defer leecher.Close()
 375         testutil.ExportStatusWriter(leecher, "l")
 376         cfg.Seed = false
 377         // cfg.TorrentDataOpener = nil
 378         cfg.DataDir, err = ioutil.TempDir("", "")
 379         require.NoError(t, err)
 380         defer os.RemoveAll(cfg.DataDir)
 381         leecherLeecher, _ := NewClient(&cfg)
 382         defer leecherLeecher.Close()
 383         testutil.ExportStatusWriter(leecherLeecher, "ll")
 384         leecherGreeting, _, _ := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 385                 ret = TorrentSpecFromMetaInfo(mi)
 386                 ret.ChunkSize = 2
 387                 return
 388         }())
 389         llg, _, _ := leecherLeecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 390                 ret = TorrentSpecFromMetaInfo(mi)
 391                 ret.ChunkSize = 3
 392                 return
 393         }())
 394         // Simultaneously DownloadAll in Leecher, and read the contents
 395         // consecutively in LeecherLeecher. This non-deterministically triggered a
 396         // case where the leecher wouldn't unchoke the LeecherLeecher.
 397         var wg sync.WaitGroup
 398         wg.Add(1)
 399         go func() {
 400                 defer wg.Done()
 401                 r := llg.NewReader()
 402                 defer r.Close()
 403                 b, err := ioutil.ReadAll(r)
 404                 require.NoError(t, err)
 405                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 406         }()
 407         leecherGreeting.AddPeers([]Peer{
 408                 Peer{
 409                         IP:   missinggo.AddrIP(seeder.ListenAddr()),
 410                         Port: missinggo.AddrPort(seeder.ListenAddr()),
 411                 },
 412                 Peer{
 413                         IP:   missinggo.AddrIP(leecherLeecher.ListenAddr()),
 414                         Port: missinggo.AddrPort(leecherLeecher.ListenAddr()),
 415                 },
 416         })
 417         wg.Add(1)
 418         go func() {
 419                 defer wg.Done()
 420                 leecherGreeting.DownloadAll()
 421                 leecher.WaitAll()
 422         }()
 423         wg.Wait()
 424 }
 425
 426 func TestReadaheadPieces(t *testing.T) {
 427         for _, case_ := range []struct {
 428                 readaheadBytes, pieceLength int64
 429                 readaheadPieces             int
 430         }{
 431                 {5 * 1024 * 1024, 256 * 1024, 19},
 432                 {5 * 1024 * 1024, 5 * 1024 * 1024, 1},
 433                 {5*1024*1024 - 1, 5 * 1024 * 1024, 1},
 434                 {5 * 1024 * 1024, 5*1024*1024 - 1, 2},
 435                 {0, 5 * 1024 * 1024, 0},
 436                 {5 * 1024 * 1024, 1048576, 4},
 437         } {
 438                 pieces := readaheadPieces(case_.readaheadBytes, case_.pieceLength)
 439                 assert.Equal(t, case_.readaheadPieces, pieces, "%v", case_)
 440         }
 441 }
 442
 443 func TestMergingTrackersByAddingSpecs(t *testing.T) {
 444         cl, err := NewClient(&TestingConfig)
 445         require.NoError(t, err)
 446         defer cl.Close()
 447         spec := TorrentSpec{}
 448         T, new, _ := cl.AddTorrentSpec(&spec)
 449         if !new {
 450                 t.FailNow()
 451         }
 452         spec.Trackers = [][]string{{"http://a"}, {"udp://b"}}
 453         _, new, _ = cl.AddTorrentSpec(&spec)
 454         if new {
 455                 t.FailNow()
 456         }
 457         assert.EqualValues(t, T.trackers[0][0], "http://a")
 458         assert.EqualValues(t, T.trackers[1][0], "udp://b")
 459 }
 460
 461 type badStorage struct{}
 462
 463 func (me badStorage) OpenTorrent(*metainfo.InfoEx) (storage.Torrent, error) {
 464         return me, nil
 465 }
 466
 467 func (me badStorage) Close() error {
 468         return nil
 469 }
 470
 471 func (me badStorage) Piece(p metainfo.Piece) storage.Piece {
 472         return badStoragePiece{p}
 473 }
 474
 475 type badStoragePiece struct {
 476         p metainfo.Piece
 477 }
 478
 479 func (me badStoragePiece) WriteAt(b []byte, off int64) (int, error) {
 480         return 0, nil
 481 }
 482
 483 func (me badStoragePiece) GetIsComplete() bool {
 484         return true
 485 }
 486
 487 func (me badStoragePiece) MarkComplete() error {
 488         return errors.New("psyyyyyyyche")
 489 }
 490
 491 func (me badStoragePiece) randomlyTruncatedDataString() string {
 492         return "hello, world\n"[:rand.Intn(14)]
 493 }
 494
 495 func (me badStoragePiece) ReadAt(b []byte, off int64) (n int, err error) {
 496         r := strings.NewReader(me.randomlyTruncatedDataString())
 497         return r.ReadAt(b, off+me.p.Offset())
 498 }
 499
 500 // We read from a piece which is marked completed, but is missing data.
 501 func TestCompletedPieceWrongSize(t *testing.T) {
 502         cfg := TestingConfig
 503         cfg.DefaultStorage = badStorage{}
 504         cl, _ := NewClient(&cfg)
 505         defer cl.Close()
 506         tt, new, err := cl.AddTorrentSpec(&TorrentSpec{
 507                 Info: &metainfo.InfoEx{
 508                         Info: metainfo.Info{
 509                                 PieceLength: 15,
 510                                 Pieces:      make([]byte, 20),
 511                                 Files: []metainfo.FileInfo{
 512                                         metainfo.FileInfo{Path: []string{"greeting"}, Length: 13},
 513                                 },
 514                         },
 515                 },
 516         })
 517         require.NoError(t, err)
 518         defer tt.Drop()
 519         assert.True(t, new)
 520         r := tt.NewReader()
 521         defer r.Close()
 522         b, err := ioutil.ReadAll(r)
 523         assert.Len(t, b, 13)
 524         assert.NoError(t, err)
 525 }
 526
 527 func BenchmarkAddLargeTorrent(b *testing.B) {
 528         cfg := TestingConfig
 529         cfg.DisableTCP = true
 530         cfg.DisableUTP = true
 531         cfg.ListenAddr = "redonk"
 532         cl, _ := NewClient(&cfg)
 533         defer cl.Close()
 534         for range iter.N(b.N) {
 535                 t, err := cl.AddTorrentFromFile("testdata/bootstrap.dat.torrent")
 536                 if err != nil {
 537                         b.Fatal(err)
 538                 }
 539                 t.Drop()
 540         }
 541 }
 542
 543 func TestResponsive(t *testing.T) {
 544         seederDataDir, mi := testutil.GreetingTestTorrent()
 545         defer os.RemoveAll(seederDataDir)
 546         cfg := TestingConfig
 547         cfg.Seed = true
 548         cfg.DataDir = seederDataDir
 549         seeder, err := NewClient(&cfg)
 550         require.Nil(t, err)
 551         defer seeder.Close()
 552         seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 553         leecherDataDir, err := ioutil.TempDir("", "")
 554         require.Nil(t, err)
 555         defer os.RemoveAll(leecherDataDir)
 556         cfg = TestingConfig
 557         cfg.DataDir = leecherDataDir
 558         leecher, err := NewClient(&cfg)
 559         require.Nil(t, err)
 560         defer leecher.Close()
 561         leecherTorrent, _, _ := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 562                 ret = TorrentSpecFromMetaInfo(mi)
 563                 ret.ChunkSize = 2
 564                 return
 565         }())
 566         leecherTorrent.AddPeers([]Peer{
 567                 Peer{
 568                         IP:   missinggo.AddrIP(seeder.ListenAddr()),
 569                         Port: missinggo.AddrPort(seeder.ListenAddr()),
 570                 },
 571         })
 572         reader := leecherTorrent.NewReader()
 573         defer reader.Close()
 574         reader.SetReadahead(0)
 575         reader.SetResponsive()
 576         b := make([]byte, 2)
 577         _, err = reader.Seek(3, os.SEEK_SET)
 578         require.NoError(t, err)
 579         _, err = io.ReadFull(reader, b)
 580         assert.Nil(t, err)
 581         assert.EqualValues(t, "lo", string(b))
 582         _, err = reader.Seek(11, os.SEEK_SET)
 583         require.NoError(t, err)
 584         n, err := io.ReadFull(reader, b)
 585         assert.Nil(t, err)
 586         assert.EqualValues(t, 2, n)
 587         assert.EqualValues(t, "d\n", string(b))
 588 }
 589
 590 func TestTorrentDroppedDuringResponsiveRead(t *testing.T) {
 591         seederDataDir, mi := testutil.GreetingTestTorrent()
 592         defer os.RemoveAll(seederDataDir)
 593         cfg := TestingConfig
 594         cfg.Seed = true
 595         cfg.DataDir = seederDataDir
 596         seeder, err := NewClient(&cfg)
 597         require.Nil(t, err)
 598         defer seeder.Close()
 599         seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 600         leecherDataDir, err := ioutil.TempDir("", "")
 601         require.Nil(t, err)
 602         defer os.RemoveAll(leecherDataDir)
 603         cfg = TestingConfig
 604         cfg.DataDir = leecherDataDir
 605         leecher, err := NewClient(&cfg)
 606         require.Nil(t, err)
 607         defer leecher.Close()
 608         leecherTorrent, _, _ := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 609                 ret = TorrentSpecFromMetaInfo(mi)
 610                 ret.ChunkSize = 2
 611                 return
 612         }())
 613         leecherTorrent.AddPeers([]Peer{
 614                 Peer{
 615                         IP:   missinggo.AddrIP(seeder.ListenAddr()),
 616                         Port: missinggo.AddrPort(seeder.ListenAddr()),
 617                 },
 618         })
 619         reader := leecherTorrent.NewReader()
 620         defer reader.Close()
 621         reader.SetReadahead(0)
 622         reader.SetResponsive()
 623         b := make([]byte, 2)
 624         _, err = reader.Seek(3, os.SEEK_SET)
 625         require.NoError(t, err)
 626         _, err = io.ReadFull(reader, b)
 627         assert.Nil(t, err)
 628         assert.EqualValues(t, "lo", string(b))
 629         go leecherTorrent.Drop()
 630         _, err = reader.Seek(11, os.SEEK_SET)
 631         require.NoError(t, err)
 632         n, err := reader.Read(b)
 633         assert.EqualError(t, err, "torrent closed")
 634         assert.EqualValues(t, 0, n)
 635 }
 636
 637 func TestDHTInheritBlocklist(t *testing.T) {
 638         ipl := iplist.New(nil)
 639         require.NotNil(t, ipl)
 640         cfg := TestingConfig
 641         cfg.IPBlocklist = ipl
 642         cfg.NoDHT = false
 643         cl, err := NewClient(&cfg)
 644         require.NoError(t, err)
 645         defer cl.Close()
 646         require.Equal(t, ipl, cl.DHT().IPBlocklist())
 647 }
 648
 649 // Check that stuff is merged in subsequent AddTorrentSpec for the same
 650 // infohash.
 651 func TestAddTorrentSpecMerging(t *testing.T) {
 652         cl, err := NewClient(&TestingConfig)
 653         require.NoError(t, err)
 654         defer cl.Close()
 655         dir, mi := testutil.GreetingTestTorrent()
 656         defer os.RemoveAll(dir)
 657         var ts TorrentSpec
 658         missinggo.CopyExact(&ts.InfoHash, mi.Info.Hash)
 659         tt, new, err := cl.AddTorrentSpec(&ts)
 660         require.NoError(t, err)
 661         require.True(t, new)
 662         require.Nil(t, tt.Info())
 663         _, new, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 664         require.NoError(t, err)
 665         require.False(t, new)
 666         require.NotNil(t, tt.Info())
 667 }
 668
 669 // Check that torrent Info is obtained from the metainfo file cache.
 670 func TestAddTorrentMetainfoInCache(t *testing.T) {
 671         cfg := TestingConfig
 672         cfg.DisableMetainfoCache = false
 673         cfg.ConfigDir, _ = ioutil.TempDir(os.TempDir(), "")
 674         defer os.RemoveAll(cfg.ConfigDir)
 675         cl, err := NewClient(&cfg)
 676         require.NoError(t, err)
 677         defer cl.Close()
 678         dir, mi := testutil.GreetingTestTorrent()
 679         defer os.RemoveAll(dir)
 680         tt, new, err := cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 681         require.NoError(t, err)
 682         require.True(t, new)
 683         require.NotNil(t, tt.Info())
 684         _, err = os.Stat(filepath.Join(cfg.ConfigDir, "torrents", fmt.Sprintf("%x.torrent", mi.Info.Hash.Bytes())))
 685         require.NoError(t, err)
 686         // Contains only the infohash.
 687         var ts TorrentSpec
 688         missinggo.CopyExact(&ts.InfoHash, mi.Info.Hash)
 689         _, ok := cl.Torrent(ts.InfoHash)
 690         require.True(t, ok)
 691         tt.Drop()
 692         _, ok = cl.Torrent(ts.InfoHash)
 693         require.False(t, ok)
 694         tt, new, err = cl.AddTorrentSpec(&ts)
 695         require.NoError(t, err)
 696         require.True(t, new)
 697         // Obtained from the metainfo cache.
 698         require.NotNil(t, tt.Info())
 699 }
 700
 701 func TestTorrentDroppedBeforeGotInfo(t *testing.T) {
 702         dir, mi := testutil.GreetingTestTorrent()
 703         os.RemoveAll(dir)
 704         cl, _ := NewClient(&TestingConfig)
 705         defer cl.Close()
 706         var ts TorrentSpec
 707         missinggo.CopyExact(&ts.InfoHash, mi.Info.Hash)
 708         tt, _, _ := cl.AddTorrentSpec(&ts)
 709         tt.Drop()
 710         assert.EqualValues(t, 0, len(cl.Torrents()))
 711         select {
 712         case <-tt.GotInfo():
 713                 t.FailNow()
 714         default:
 715         }
 716 }
 717
 718 func writeTorrentData(ts storage.Torrent, info *metainfo.InfoEx, b []byte) {
 719         for i := range iter.N(info.NumPieces()) {
 720                 n, _ := ts.Piece(info.Piece(i)).WriteAt(b, 0)
 721                 b = b[n:]
 722         }
 723 }
 724
 725 func testAddTorrentPriorPieceCompletion(t *testing.T, alreadyCompleted bool) {
 726         fileCacheDir, err := ioutil.TempDir("", "")
 727         require.NoError(t, err)
 728         defer os.RemoveAll(fileCacheDir)
 729         fileCache, err := filecache.NewCache(fileCacheDir)
 730         require.NoError(t, err)
 731         greetingDataTempDir, greetingMetainfo := testutil.GreetingTestTorrent()
 732         defer os.RemoveAll(greetingDataTempDir)
 733         filePieceStore := storage.NewPieceFileStorage(fileCache.AsFileStore())
 734         greetingData, err := filePieceStore.OpenTorrent(&greetingMetainfo.Info)
 735         require.NoError(t, err)
 736         writeTorrentData(greetingData, &greetingMetainfo.Info, []byte(testutil.GreetingFileContents))
 737         // require.Equal(t, len(testutil.GreetingFileContents), written)
 738         // require.NoError(t, err)
 739         for i := 0; i < greetingMetainfo.Info.NumPieces(); i++ {
 740                 p := greetingMetainfo.Info.Piece(i)
 741                 if alreadyCompleted {
 742                         err := greetingData.Piece(p).MarkComplete()
 743                         assert.NoError(t, err)
 744                 }
 745         }
 746         cfg := TestingConfig
 747         // TODO: Disable network option?
 748         cfg.DisableTCP = true
 749         cfg.DisableUTP = true
 750         cfg.DefaultStorage = filePieceStore
 751         cl, err := NewClient(&cfg)
 752         require.NoError(t, err)
 753         defer cl.Close()
 754         tt, err := cl.AddTorrent(greetingMetainfo)
 755         require.NoError(t, err)
 756         psrs := tt.PieceStateRuns()
 757         assert.Len(t, psrs, 1)
 758         assert.EqualValues(t, 3, psrs[0].Length)
 759         assert.Equal(t, alreadyCompleted, psrs[0].Complete)
 760         if alreadyCompleted {
 761                 r := tt.NewReader()
 762                 b, err := ioutil.ReadAll(r)
 763                 assert.NoError(t, err)
 764                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 765         }
 766 }
 767
 768 func TestAddTorrentPiecesAlreadyCompleted(t *testing.T) {
 769         testAddTorrentPriorPieceCompletion(t, true)
 770 }
 771
 772 func TestAddTorrentPiecesNotAlreadyCompleted(t *testing.T) {
 773         testAddTorrentPriorPieceCompletion(t, false)
 774 }
 775
 776 func TestAddMetainfoWithNodes(t *testing.T) {
 777         cfg := TestingConfig
 778         cfg.NoDHT = false
 779         // For now, we want to just jam the nodes into the table, without
 780         // verifying them first. Also the DHT code doesn't support mixing secure
 781         // and insecure nodes if security is enabled (yet).
 782         cfg.DHTConfig.NoSecurity = true
 783         cl, err := NewClient(&cfg)
 784         require.NoError(t, err)
 785         defer cl.Close()
 786         assert.EqualValues(t, cl.DHT().NumNodes(), 0)
 787         tt, err := cl.AddTorrentFromFile("metainfo/testdata/issue_65a.torrent")
 788         require.NoError(t, err)
 789         assert.Len(t, tt.trackers, 5)
 790         assert.EqualValues(t, 6, cl.DHT().NumNodes())
 791 }
 792
 793 type testDownloadCancelParams struct {
 794         Responsive                bool
 795         Readahead                 int64
 796         SetReadahead              bool
 797         ExportClientStatus        bool
 798         SetLeecherStorageCapacity bool
 799         LeecherStorageCapacity    int64
 800         Cancel                    bool
 801 }
 802
 803 func testDownloadCancel(t *testing.T, ps testDownloadCancelParams) {
 804         greetingTempDir, mi := testutil.GreetingTestTorrent()
 805         defer os.RemoveAll(greetingTempDir)
 806         cfg := TestingConfig
 807         cfg.Seed = true
 808         cfg.DataDir = greetingTempDir
 809         seeder, err := NewClient(&cfg)
 810         require.NoError(t, err)
 811         defer seeder.Close()
 812         if ps.ExportClientStatus {
 813                 testutil.ExportStatusWriter(seeder, "s")
 814         }
 815         seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
 816         leecherDataDir, err := ioutil.TempDir("", "")
 817         require.NoError(t, err)
 818         defer os.RemoveAll(leecherDataDir)
 819         fc, err := filecache.NewCache(leecherDataDir)
 820         require.NoError(t, err)
 821         if ps.SetLeecherStorageCapacity {
 822                 fc.SetCapacity(ps.LeecherStorageCapacity)
 823         }
 824         cfg.DefaultStorage = storage.NewPieceFileStorage(fc.AsFileStore())
 825         cfg.DataDir = leecherDataDir
 826         leecher, _ := NewClient(&cfg)
 827         defer leecher.Close()
 828         if ps.ExportClientStatus {
 829                 testutil.ExportStatusWriter(leecher, "l")
 830         }
 831         leecherGreeting, new, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
 832                 ret = TorrentSpecFromMetaInfo(mi)
 833                 ret.ChunkSize = 2
 834                 return
 835         }())
 836         require.NoError(t, err)
 837         assert.True(t, new)
 838         psc := leecherGreeting.SubscribePieceStateChanges()
 839         defer psc.Close()
 840         leecherGreeting.DownloadAll()
 841         if ps.Cancel {
 842                 leecherGreeting.CancelPieces(0, leecherGreeting.NumPieces())
 843         }
 844         leecherGreeting.AddPeers([]Peer{
 845                 Peer{
 846                         IP:   missinggo.AddrIP(seeder.ListenAddr()),
 847                         Port: missinggo.AddrPort(seeder.ListenAddr()),
 848                 },
 849         })
 850         completes := make(map[int]bool, 3)
 851 values:
 852         for {
 853                 // started := time.Now()
 854                 select {
 855                 case _v := <-psc.Values:
 856                         // log.Print(time.Since(started))
 857                         v := _v.(PieceStateChange)
 858                         completes[v.Index] = v.Complete
 859                 case <-time.After(100 * time.Millisecond):
 860                         break values
 861                 }
 862         }
 863         if ps.Cancel {
 864                 assert.EqualValues(t, map[int]bool{0: false, 1: false, 2: false}, completes)
 865         } else {
 866                 assert.EqualValues(t, map[int]bool{0: true, 1: true, 2: true}, completes)
 867         }
 868
 869 }
 870
 871 func TestTorrentDownloadAll(t *testing.T) {
 872         testDownloadCancel(t, testDownloadCancelParams{})
 873 }
 874
 875 func TestTorrentDownloadAllThenCancel(t *testing.T) {
 876         testDownloadCancel(t, testDownloadCancelParams{
 877                 Cancel: true,
 878         })
 879 }
 880
 881 // Ensure that it's an error for a peer to send an invalid have message.
 882 func TestPeerInvalidHave(t *testing.T) {
 883         cl, err := NewClient(&TestingConfig)
 884         require.NoError(t, err)
 885         defer cl.Close()
 886         tt, _new, err := cl.AddTorrentSpec(&TorrentSpec{
 887                 Info: &metainfo.InfoEx{
 888                         Info: metainfo.Info{
 889                                 PieceLength: 1,
 890                                 Pieces:      make([]byte, 20),
 891                                 Files:       []metainfo.FileInfo{{Length: 1}},
 892                         },
 893                 },
 894         })
 895         require.NoError(t, err)
 896         assert.True(t, _new)
 897         defer tt.Drop()
 898         cn := &connection{
 899                 t: tt,
 900         }
 901         assert.NoError(t, cn.peerSentHave(0))
 902         assert.Error(t, cn.peerSentHave(1))
 903 }
 904
 905 func TestPieceCompletedInStorageButNotClient(t *testing.T) {
 906         greetingTempDir, greetingMetainfo := testutil.GreetingTestTorrent()
 907         defer os.RemoveAll(greetingTempDir)
 908         cfg := TestingConfig
 909         cfg.DataDir = greetingTempDir
 910         seeder, err := NewClient(&TestingConfig)
 911         require.NoError(t, err)
 912         seeder.AddTorrentSpec(&TorrentSpec{
 913                 Info: &greetingMetainfo.Info,
 914         })
 915 }