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