tests/testifacecache.py

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