tests/testifacecache.py

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