tests/testifacecache.py

   1 #!/usr/bin/env python
   2 from basetest import BaseTest
   3 import sys, tempfile, os, time
   4 import unittest
   5 import data
   6
   7 sys.path.insert(0, '..')
   8 from zeroinstall.injector import model, gpg, trust
   9 from zeroinstall.injector.namespaces import config_site
  10 from zeroinstall.injector.iface_cache import PendingFeed
  11 from zeroinstall.support import basedir
  12
  13 class TestIfaceCache(BaseTest):
  14         def testList(self):
  15                 iface_cache = self.config.iface_cache
  16                 self.assertEqual([], iface_cache.list_all_interfaces())
  17                 iface_dir = basedir.save_cache_path(config_site, 'interfaces')
  18                 open(os.path.join(iface_dir, 'http%3a%2f%2ffoo'), 'w').close()
  19                 self.assertEqual(['http://foo'],
  20                                 iface_cache.list_all_interfaces())
  21                 # TODO: test overrides
  22
  23         def testCheckSigned(self):
  24                 iface_cache = self.config.iface_cache
  25                 trust.trust_db.trust_key(
  26                         '92429807C9853C0744A68B9AAE07828059A53CC1')
  27                 feed_url = 'http://foo'
  28                 src = tempfile.TemporaryFile()
  29
  30                 # Unsigned
  31                 src.write("hello")
  32                 src.flush()
  33                 src.seek(0)
  34                 try:
  35                         PendingFeed(feed_url, src)
  36                         assert 0
  37                 except model.SafeException:
  38                         pass
  39
  40                 stream = tempfile.TemporaryFile()
  41                 stream.write(data.thomas_key)
  42                 stream.seek(0)
  43
  44                 gpg.import_key(stream)
  45
  46                 # Signed
  47                 src.seek(0)
  48                 src.write(data.foo_signed_xml)
  49                 src.flush()
  50                 src.seek(0)
  51
  52                 pending = PendingFeed(feed_url, src)
  53                 assert iface_cache.update_feed_if_trusted(feed_url, pending.sigs, pending.new_xml)
  54
  55                 self.assertEqual(['http://foo'],
  56                                 iface_cache.list_all_interfaces())
  57
  58                 feed = iface_cache.get_feed(feed_url)
  59
  60                 self.assertEqual(1154850229, feed.last_modified)
  61
  62         def testXMLupdate(self):
  63                 iface_cache = self.config.iface_cache
  64                 trust.trust_db.trust_key(
  65                         '92429807C9853C0744A68B9AAE07828059A53CC1')
  66                 stream = tempfile.TemporaryFile()
  67                 stream.write(data.thomas_key)
  68                 stream.seek(0)
  69                 gpg.import_key(stream)
  70
  71                 iface = iface_cache.get_interface('http://foo')
  72                 src = tempfile.TemporaryFile()
  73                 src.write(data.foo_signed_xml)
  74                 src.seek(0)
  75                 pending = PendingFeed(iface.uri, src)
  76                 assert iface_cache.update_feed_if_trusted(iface.uri, pending.sigs, pending.new_xml)
  77
  78                 iface_cache.__init__()
  79                 feed = iface_cache.get_feed('http://foo')
  80                 assert feed.last_modified == 1154850229
  81
  82                 # mtimes are unreliable because copying often changes them -
  83                 # check that we extract the time from the signature when upgrading
  84                 upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
  85                 cached = os.path.join(upstream_dir, model.escape(feed.url))
  86                 os.utime(cached, None)
  87
  88                 iface_cache.__init__()
  89                 feed = iface_cache.get_feed('http://foo')
  90                 assert feed.last_modified > 1154850229
  91
  92                 src = tempfile.TemporaryFile()
  93                 src.write(data.new_foo_signed_xml)
  94                 src.seek(0)
  95
  96                 pending = PendingFeed(feed.url, src)
  97                 assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)
  98
  99                 # Can't 'update' to an older copy
 100                 src = tempfile.TemporaryFile()
 101                 src.write(data.foo_signed_xml)
 102                 src.seek(0)
 103                 try:
 104                         pending = PendingFeed(feed.url, src)
 105                         assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)
 106
 107                         assert 0
 108                 except model.SafeException:
 109                         pass
 110
 111         def testTimes(self):
 112                 iface_cache = self.config.iface_cache
 113                 stream = tempfile.TemporaryFile()
 114                 stream.write(data.thomas_key)
 115                 stream.seek(0)
 116                 gpg.import_key(stream)
 117
 118                 upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
 119                 cached = os.path.join(upstream_dir, model.escape('http://foo'))
 120
 121                 stream = open(cached, 'w')
 122                 stream.write(data.foo_signed_xml)
 123                 stream.close()
 124
 125                 signed = iface_cache._get_signature_date('http://foo')
 126                 assert signed == None
 127
 128                 trust.trust_db.trust_key(
 129                         '92429807C9853C0744A68B9AAE07828059A53CC1')
 130
 131                 signed = iface_cache._get_signature_date('http://foo')
 132                 assert signed == 1154850229
 133
 134                 stream = open(cached, 'w+')
 135                 stream.seek(0)
 136                 stream.write('Hello')
 137                 stream.close()
 138
 139                 # When the signature is invalid, we just return None.
 140                 # This is because versions < 0.22 used to corrupt the signatue
 141                 # by adding an attribute to the XML
 142                 signed = iface_cache._get_signature_date('http://foo')
 143                 assert signed == None
 144
 145         def testCheckAttempt(self):
 146                 iface_cache = self.config.iface_cache
 147                 self.assertEqual(None, iface_cache.get_last_check_attempt("http://foo/bar.xml"))
 148
 149                 start_time = time.time() - 5    # Seems to be some odd rounding here
 150                 iface_cache.mark_as_checking("http://foo/bar.xml")
 151                 last_check = iface_cache.get_last_check_attempt("http://foo/bar.xml")
 152
 153                 assert last_check is not None
 154                 assert last_check >= start_time, (last_check, start_time)
 155
 156                 self.assertEqual(None, iface_cache.get_last_check_attempt("http://foo/bar2.xml"))
 157
 158         def testIsStale(self):
 159                 iface_cache = self.config.iface_cache
 160                 feed = self.import_feed('http://localhost:8000/Hello', 'Hello')
 161                 assert iface_cache.is_stale(feed.url, 1) == True
 162                 assert iface_cache.is_stale(feed.url, time.time() + 1) == False
 163                 iface_cache.mark_as_checking(feed.url)
 164                 assert iface_cache.is_stale(feed.url, 1) == False
 165
 166                 # Old API
 167                 assert iface_cache.is_stale(feed, 1) == False
 168
 169 if __name__ == '__main__':
 170         unittest.main()