tests/testschedular.py

   1 #!/usr/bin/env python
   2 import sys
   3 from basetest import BaseTest
   4 import unittest
   5
   6 sys.path.insert(0, '..')
   7
   8 from zeroinstall.injector import scheduler
   9 from zeroinstall.support import tasks
  10
  11 real_spawn = scheduler._spawn_thread
  12
  13 class DummyDownload:
  14         def __init__(self, url, downloads):
  15                 self._aborted = tasks.Blocker("abort " + url)
  16                 self.url = url
  17                 self.downloads = downloads
  18
  19         def _finish(self, status):
  20                 self.downloads[self.url] = status
  21
  22 class TestSchedular(BaseTest):
  23         def setUp(self):
  24                 BaseTest.setUp(self)
  25
  26         def tearDown(self):
  27                 BaseTest.tearDown(self)
  28                 scheduler._spawn_thread = real_spawn
  29
  30         def testQueuing(self):
  31                 #import logging; logging.getLogger().setLevel(logging.DEBUG)
  32
  33                 site = scheduler.Site()
  34
  35                 steps = []
  36
  37                 downloads = {}
  38
  39                 @tasks.async
  40                 def dummy_spawn_thread(step):
  41                         resume = tasks.Blocker('complete ' + step.url)
  42                         downloads[step.url] = resume
  43                         yield resume
  44                         try:
  45                                 tasks.check(resume)
  46                         except Exception:
  47                                 step.status = "fail"
  48                         else:
  49                                 step.status = "ok"
  50
  51                 scheduler._spawn_thread = dummy_spawn_thread
  52                 scheduler.MAX_DOWNLOADS_PER_SITE = 2
  53
  54                 for i in range(4):
  55                         dl = DummyDownload("http://step/" + str(i), downloads)
  56
  57                         s = scheduler.DownloadStep()
  58                         s.url = dl.url
  59                         s.dl = dl
  60                         steps.append(site.download(s))
  61
  62                 @tasks.async
  63                 def collect():
  64                         # Let the first two downloads start
  65                         for x in range(10): yield
  66                         self.assertEqual(2, len(downloads))
  67
  68                         # Let one of them complete
  69                         url, blocker = list(downloads.items())[0]
  70                         blocker.trigger()
  71                         for x in range(10): yield
  72                         self.assertEqual(3, len(downloads))
  73
  74                         # Check it was successful
  75                         self.assertEqual("ok", downloads[url])
  76                         del downloads[url]
  77
  78                         # Let the next one fail
  79                         url, blocker = list(downloads.items())[0]
  80                         blocker.trigger(exception = (Exception("test"), None))
  81                         for x in range(10): yield
  82                         self.assertEqual(3, len(downloads))
  83
  84                         # Check it failed
  85                         self.assertEqual("fail", downloads[url])
  86                         del downloads[url]
  87
  88                         # The last two should both be in progress now.
  89                         # Allow them both to finish.
  90                         blockers = list(downloads.values())
  91                         blockers[0].trigger()
  92                         blockers[1].trigger()
  93                         for x in range(10): yield
  94                         results = list(downloads.values())
  95                         self.assertEqual(["ok", "ok"], results)
  96
  97                 tasks.wait_for_blocker(collect())
  98
  99 if __name__ == '__main__':
 100         unittest.main()