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