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
  47                 src.seek(0)
  48                 src.write(data.foo_signed)
  49                 src.flush()
  50                 src.seek(0)
  51
  52                 pending = PendingFeed(iface.uri, src)
  53                 assert iface_cache.update_interface_if_trusted(iface, pending.sigs, pending.new_xml)
  54
  55                 self.assertEquals(['http://foo'],
  56                                 iface_cache.list_all_interfaces())
  57
  58                 self.assertEquals(1116788178,  iface.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_interface_if_trusted(iface, pending.sigs, pending.new_xml)
  74
  75                 iface_cache.__init__()
  76                 iface = iface_cache.get_interface('http://foo')
  77                 assert iface.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(iface.uri))
  83                 os.utime(cached, None)
  84
  85                 iface_cache.__init__()
  86                 iface = iface_cache.get_interface('http://foo')
  87                 assert iface.last_modified > 1154850229
  88
  89                 src = tempfile.TemporaryFile()
  90                 src.write(data.new_foo_signed_xml)
  91                 src.seek(0)
  92
  93                 pending = PendingFeed(iface.uri, src)
  94                 assert iface_cache.update_interface_if_trusted(iface, 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(iface.uri, src)
 102                         assert iface_cache.update_interface_if_trusted(iface, 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 suite = unittest.makeSuite(TestIfaceCache)
 155 if __name__ == '__main__':
 156         sys.argv.append('-v')
 157         unittest.main()