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                 iface_cache.add_pending(pending)
  54                 assert iface_cache.update_interface_if_trusted(iface, pending.sigs, pending.new_xml)
  55
  56                 self.assertEquals(['http://foo'],
  57                                 iface_cache.list_all_interfaces())
  58
  59                 self.assertEquals(1116788178,  iface.last_modified)
  60
  61         def testXMLupdate(self):
  62                 trust.trust_db.trust_key(
  63                         '92429807C9853C0744A68B9AAE07828059A53CC1')
  64                 stream = tempfile.TemporaryFile()
  65                 stream.write(data.thomas_key)
  66                 stream.seek(0)
  67                 gpg.import_key(stream)
  68
  69                 iface = iface_cache.get_interface('http://foo')
  70                 src = tempfile.TemporaryFile()
  71                 src.write(data.foo_signed_xml)
  72                 src.seek(0)
  73                 pending = PendingFeed(iface.uri, src)
  74                 iface_cache.add_pending(pending)
  75                 assert iface_cache.update_interface_if_trusted(iface, pending.sigs, pending.new_xml)
  76
  77                 iface_cache.__init__()
  78                 iface = iface_cache.get_interface('http://foo')
  79                 assert iface.last_modified == 1154850229
  80
  81                 # mtimes are unreliable because copying often changes them -
  82                 # check that we extract the time from the signature when upgrading
  83                 upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
  84                 cached = os.path.join(upstream_dir, model.escape(iface.uri))
  85                 os.utime(cached, None)
  86
  87                 iface_cache.__init__()
  88                 iface = iface_cache.get_interface('http://foo')
  89                 assert iface.last_modified > 1154850229
  90
  91                 src = tempfile.TemporaryFile()
  92                 src.write(data.new_foo_signed_xml)
  93                 src.seek(0)
  94
  95                 pending = PendingFeed(iface.uri, src)
  96                 iface_cache.add_pending(pending)
  97                 assert iface_cache.update_interface_if_trusted(iface, 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(iface.uri, src)
 105                         iface_cache.add_pending(pending)
 106                         assert iface_cache.update_interface_if_trusted(iface, pending.sigs, pending.new_xml)
 107
 108                         assert 0
 109                 except model.SafeException:
 110                         pass
 111
 112         def testTimes(self):
 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 = file(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 = file(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                 self.assertEquals(None, iface_cache.get_last_check_attempt("http://foo/bar.xml"))
 147
 148                 start_time = time.time() - 5    # Seems to be some odd rounding here
 149                 iface_cache.mark_as_checking("http://foo/bar.xml")
 150                 last_check = iface_cache.get_last_check_attempt("http://foo/bar.xml")
 151
 152                 assert last_check is not None
 153                 assert last_check >= start_time, (last_check, start_time)
 154
 155                 self.assertEquals(None, iface_cache.get_last_check_attempt("http://foo/bar2.xml"))
 156
 157
 158 suite = unittest.makeSuite(TestIfaceCache)
 159 if __name__ == '__main__':
 160         sys.argv.append('-v')
 161         unittest.main()