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