1# Written by Arno Bakker
2# see LICENSE.txt for license information
3
4import unittest
5
6import os
7import sys
8import tempfile
9import random
10import shutil
11import time
12import subprocess
13import urllib2
14import string
15import binascii
16from traceback import print_exc
17
18from testasserver import TestAsServer
19from SwiftDef import SwiftDef
20from swiftconn import *
21
22DEBUG=False
23
24
25class TestDirSeedFramework(TestAsServer):
26    """
27    Framework for multi-swarm tests.
28    """
29
30    def setUpPreSession(self):
31        TestAsServer.setUpPreSession(self)
32        #self.cmdport = None
33        self.httpport = None
34        self.scandir = "seeder2"
35        self.progress = True
36
37        self.setUpScanDir()
38
39        try:
40            shutil.rmtree(self.scandir)
41        except:
42            pass
43        os.mkdir(self.scandir)
44
45        # Create content
46        for i in range(len(self.filelist)):
47            fn = self.filelist[i][0]
48            s = self.filelist[i][1]
49            f = open(fn,"wb")
50            data = '#' * s
51            f.write(data)
52            f.close()
53
54            # Pre hash check and checkpoint them
55            sdef = SwiftDef()
56            sdef.add_content(fn)
57            sdef.finalize(self.binpath)
58            self.filelist[i][2] = sdef.get_id() # save roothash
59
60        for i in range(len(self.filelist)):
61            print >>sys.stderr,"test: Configured from filelist",binascii.hexlify(self.filelist[i][2])
62
63
64    def setUpScanDir(self):
65        self.filelist = []
66        # Minimum 1 entry
67        # DO NOT MODIFY THESE ENTRIES without adjusting tests, e.g. requesttest.py
68        self.filelist.append([os.path.join(self.scandir,"anita.ts"), 1234, None])
69        self.filelist.append([os.path.join(self.scandir,"bill.ts"),  200487, None])
70        self.filelist.append([os.path.join(self.scandir,"claire.ts"),65535, None])
71
72    def setUpPostSession(self):
73        TestAsServer.setUpPostSession(self)
74
75    def tearDown(self):
76        TestAsServer.tearDown(self)
77        time.sleep(1)
78        try:
79            shutil.rmtree(self.scandir)
80        except:
81            pass
82
83
84class TestDirSeed(TestDirSeedFramework):
85    """
86    Basic test for the SwarmManager activation code to see if it found the
87    swarms in the scandir.
88    """
89
90    def test_connect_one(self):
91        myaddr = ("127.0.0.1",15353)
92        hisaddr = ("127.0.0.1",self.listenport)
93
94        # last
95        swarmid = self.filelist[len(self.filelist)-1][2]
96
97        s = SwiftConnection(myaddr,hisaddr,swarmid)
98        d = s.recv()
99        responded = False
100        while True:
101            msg = d.get_message()
102            if msg is None:
103                break
104            if msg.get_id() == MSG_ID_HANDSHAKE:
105                print >>sys.stderr,"test: Parsed",`msg`
106                responded = True
107
108        self.assertTrue(responded)
109
110        time.sleep(10)
111
112
113    def test_connect_reactivate(self):
114
115        self.test_connect_one()
116
117        print >>sys.stderr,"test: Now sleeping so swift deactivates swarm"
118        time.sleep(60)
119        # Should be deactivated
120
121        print >>sys.stderr,"test: Testing if reactivate works"
122        self.test_connect_one()
123
124
125def test_suite():
126    suite = unittest.TestSuite()
127    suite.addTest(unittest.makeSuite(TestDirSeed))
128
129    return suite
130
131
132def main():
133    unittest.main(defaultTest='test_suite',argv=[sys.argv[0]])
134
135if __name__ == "__main__":
136    main()
137
138