#!/usr/bin/env python # Unit Test Suites for pyTracker Project # # Using standard unit test framework coming with Python # To execute the test cases, type: # >>python test_pyTracker.py # # Written by Alex Dong ( alex.dong at gmail.com) import urllib2 import unittest from bencode import bencode, bdecode from urllib import quote from models import peer_db import web as db from logger import logger logger = logger('TEST_TRACKER') ''' ===== Section: Test Code Structure ========== 1. There will be two annouce and scrape server pairs to be tested. One will be public available track server and the other one will be the local pyTracker. 2. Every test case will be executed twise. The first time executes against the public track server to make sure the test case design and implementation are correct. The second time against the local pyTracker server. 3. The test code has two layers: 1) meta-instruction layer: This layer contains all the low-level instructions to deal with tracker server. For example, announce('start'), _status_assertions, etc. 2) test case layer: This layer will be built on top of the meta-instruction layer and it contains the main test case scenarios and assertions. ''' def requestURL(url): fp = urllib2.urlopen(url) r = bdecode(fp.read()) fp.close() return r # Global Constants. Used to build test cases. public_tracker_url = 'http://tracker.torrentbox.com:2710/' local_tracker_url = 'http://127.0.0.1:8080/tracker/' target_tracker_url = local_tracker_url + "announce" target_scraper_url = local_tracker_url + "scrape" info_hash = '839be8624144b3f993118240d9332d673bff7523' peer_id = 'M4-20-2-e918a1127f32' port = 6881 size = 93989333 torrent_name = 'Madonna - Confessions on a dance floor' class MetaInstruction(unittest.TestCase): def __init__(self, url, info_hash, size, peer_id=peer_id, port=port): self._url = url self._info_hash = info_hash.decode('hex') self._size = size self._peer_id = peer_id self._port = port def status_assert(self, cur_status, expectation=None): (cur_complete, cur_incomplete) = cur_status if expectation: if expectation['complete']=='down': self.assert_(self._last_complete - 1 == cur_complete) elif expectation['complete']=='up': self.assert_(self._last_complete + 1 == cur_complete) else: self.assert_(self._last_complete == cur_complete) if expectation['incomplete']=='down': self.assert_(self._last_incomplete - 1 == cur_incomplete) elif expectation['incomplete']=='up': self.assert_(self._last_incomplete + 1 == cur_incomplete) else: self.assert_(self._last_incomplete == cur_incomplete) (self._last_complete, self._last_incomplete) = cur_status def scrape(self): r = requestURL("%s?info_hash=%s" % (target_scraper_url, quote(self._info_hash))) r = r['files'][self._info_hash] return (r['complete'], r['incomplete']) def announce(self, event, downloaded, uploaded, left): # Make URL url = "%s?info_hash=%s&peer_id=%s&compact=1&port=%s&event=%s&downloaded=%s&uploaded=%s&left=%s" % \ (self._url, quote(self._info_hash), quote(self._peer_id), str(self._port), event, downloaded, uploaded, left) r = requestURL(url) return (r['complete'], r['incomplete']) class ExceptionalTestCases(unittest.TestCase): def _has_failure(self, url): r = requestURL(url) self.assert_(r['failure reason:'] != None) def testMissingInfoHash(self): self._has_failure("%s?peer_id=%s" % (target_tracker_url, peer_id)) def testNonExistenceInfoHash(self): self._has_failure("%s?info_hash=839be8624144b3f993118240d9332d673bf32322&peer_id=%s&port=%s" % (target_tracker_url, peer_id, port)) def testInvalidPort(self): self._has_failure("%s?info_hash=%s&peer_id=%s&port=68810976" % (target_tracker_url, info_hash, peer_id)) def testInvalidEVent(self): self._has_failure("%s?info_hash=%s&peer_id=%s&port=%s&event=Unknown" % (target_tracker_url, info_hash, peer_id, port)) def testInvalidLeftAndEvent(self): self._has_failure("%s?info_hash=%s&peer_id=%s&port=%s&event=Completed&left=25" % (target_tracker_url, info_hash, peer_id, port)) class MultipleClientsDownloadMultipleTorrentsTestCases(unittest.TestCase): def setUp(self): self._ut_tor_1 = MetaInstruction(target_tracker_url, '101c9d63211c3c570ffbadd49c5649d3fb497273',93989333, "UT000000000000000001", 21056) self._ut_tor_2 = MetaInstruction(target_tracker_url, '09bb5bb38283e111a5b830243525659e077d7f24',93989333, "UT000000000000000001", 21056) self._ml_tor_1 = MetaInstruction(target_tracker_url, '101c9d63211c3c570ffbadd49c5649d3fb497273', 93989333, "Mainline000000000001", 61099) self._ml_tor_2 = MetaInstruction(target_tracker_url, '09bb5bb38283e111a5b830243525659e077d7f24', 93989333, "Mainline000000000001", 61099) peer_db.truncate() def testOneClientMultipleTorrent(self): pass def testTwoClientsOneTorrent(self): """'Mainline' seeding and 'UT' downloading""" self._ml_tor_1.status_assert(self._ml_tor_1.announce('started', 93989333, 93989333, 0)) self._ml_tor_1.status_assert(self._ut_tor_1.announce('started', 93989333, 2342, 234231), {'complete':'equal', 'incomplete':'up'}) self._ml_tor_1.status_assert(self._ut_tor_1.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) def testTwoClientsReceiveCorrectLocationInfo(self): pass class OneClientDownloadsMultipleTorrentsTestCases(unittest.TestCase): def setUp(self): self._mi_1 = MetaInstruction(target_tracker_url, '101c9d63211c3c570ffbadd49c5649d3fb497273',93989333) self._mi_2 = MetaInstruction(target_tracker_url, '09bb5bb38283e111a5b830243525659e077d7f24', 93989333) def testTwoTorrentsAtSameOrder(self): self._mi_1.status_assert(self._mi_1.announce('started', 93989333, 93989333, 0)) self._mi_2.status_assert(self._mi_2.announce('started', 93989333, 93989333, 0)) self._mi_2.status_assert(self._mi_2.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) self._mi_1.status_assert(self._mi_1.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testTwoTorrentsRequests(self): self._mi_1.status_assert(self._mi_1.announce('started', 93989333, 93989333, 0)) self._mi_2.status_assert(self._mi_2.announce('started', 93989333, 93989333, 0)) self._mi_2.status_assert(self._mi_2.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) self._mi_1.status_assert(self._mi_1.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testTwoTorrentsRandomOrder(self): self._mi_1.status_assert(self._mi_1.announce('started', 93989333, 93989333, 233345)) self._mi_1.status_assert(self._mi_1.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) self._mi_2.status_assert(self._mi_2.announce('started', 93989333, 93989333, 334234)) self._mi_2.status_assert(self._mi_2.announce('stopped', 93989333, 93989333, 0), {'complete':'equal', 'incomplete':'down'}) self._mi_2.status_assert(self._mi_2.announce('started', 93989333, 93989333, 234234), {'complete':'equal', 'incomplete':'up'}) self._mi_1.status_assert(self._mi_1.announce('completed', 93989333, 93989333, 0), {'complete':'equal', 'incomplete':'equal'}) self._mi_1.status_assert(self._mi_1.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) self._mi_2.status_assert(self._mi_2.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) self._mi_2.status_assert(self._mi_2.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) class SmokeTestCases(unittest.TestCase): def setUp(self): self._url = target_tracker_url self._info_hash = info_hash self._size = size self._torrent_name = torrent_name self._mi = MetaInstruction(self._url, self._info_hash, self._size) peer_db.truncate() def testInitialIncompleteStartup(self): init = self._mi.scrape() curr = self._mi.announce('started', 93989333, 93989333, 333) self.assert_(init[0]==curr[0]) self.assert_(init[1]+1==curr[1]) def testInitialCompleteStartup(self): init = self._mi.scrape() curr = self._mi.announce('started', 93989333, 93989333, 0) self.assert_(init[0]+1==curr[0]) # complete self.assert_(init[1]==curr[1]) # incomplete def testInitialSeeding(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 0)) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testDownloaderRequests(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 32424)) self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 322), {'complete':'equal', 'incomplete':'equal'}) def testDuplicatedCompeteRequests(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 32424)) self._mi.status_assert(self._mi.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) self._mi.status_assert(self._mi.announce('completed', 93989333, 93989333, 0), {'complete':'equal', 'incomplete':'equal'}) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testDuplicatedStartedRequests(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 0)) self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 0), {'complete':'equal', 'incomplete':'equal'}) self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 0), {'complete':'equal', 'incomplete':'equal'}) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testDuplicatedStoppedRequests(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 23420)) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 2342), {'complete':'equal', 'incomplete':'down'}) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 2342), {'complete':'equal', 'incomplete':'equal'}) def testResumeWorkflow(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 234320)) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 234230), {'complete':'equal', 'incomplete':'down'}) self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 234230), {'complete':'equal', 'incomplete':'up'}) self._mi.status_assert(self._mi.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) def testNormalWorkflow(self): self._mi.status_assert(self._mi.announce('started', 93989333, 93989333, 234230)) self._mi.status_assert(self._mi.announce('completed', 93989333, 93989333, 0), {'complete':'up', 'incomplete':'down'}) self._mi.status_assert(self._mi.announce('stopped', 93989333, 93989333, 0), {'complete':'down', 'incomplete':'equal'}) if __name__ == '__main__': db.db_parameters = dict(dbn='mysql', user='root', pw='root', db='bitspace') db.load() unittest.main()