tests/testifacecache.py

   1 #!/usr/bin/env python
   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 PendingFeed
  11 from zeroinstall.support import basedir
  12
  13 class TestIfaceCache(BaseTest):
  14         def testList(self):
  15                 iface_cache = self.config.iface_cache
  16                 self.assertEqual([], iface_cache.list_all_interfaces())
  17                 iface_dir = basedir.save_cache_path(config_site, 'interfaces')
  18                 open(os.path.join(iface_dir, 'http%3a%2f%2ffoo'), 'w').close()
  19                 self.assertEqual(['http://foo'],
  20                                 iface_cache.list_all_interfaces())
  21                 # TODO: test overrides
  22
  23         def testCheckSigned(self):
  24                 iface_cache = self.config.iface_cache
  25                 trust.trust_db.trust_key(
  26                         '92429807C9853C0744A68B9AAE07828059A53CC1')
  27                 feed_url = 'http://foo'
  28                 with tempfile.TemporaryFile(mode = 'w+b') as src:
  29                         # Unsigned
  30                         src.write(b"hello")
  31                         src.flush()
  32                         src.seek(0)
  33                         try:
  34                                 PendingFeed(feed_url, src)
  35                                 assert 0
  36                         except model.SafeException:
  37                                 pass
  38
  39                         with tempfile.TemporaryFile(mode = 'w+b') as stream:
  40                                 stream.write(data.thomas_key)
  41                                 stream.seek(0)
  42
  43                                 gpg.import_key(stream)
  44
  45                         # Signed
  46                         src.seek(0)
  47                         src.write(data.foo_signed_xml)
  48                         src.flush()
  49                         src.seek(0)
  50
  51                         pending = PendingFeed(feed_url, src)
  52                         assert iface_cache.update_feed_if_trusted(feed_url, pending.sigs, pending.new_xml)
  53
  54                         self.assertEqual(['http://foo'],
  55                                         iface_cache.list_all_interfaces())
  56
  57                         feed = iface_cache.get_feed(feed_url)
  58
  59                         self.assertEqual(1154850229, feed.last_modified)
  60
  61         def testXMLupdate(self):
  62                 iface_cache = self.config.iface_cache
  63                 trust.trust_db.trust_key(
  64                         '92429807C9853C0744A68B9AAE07828059A53CC1')
  65                 with tempfile.TemporaryFile() as stream:
  66                         stream.write(data.thomas_key)
  67                         stream.seek(0)
  68                         gpg.import_key(stream)
  69
  70                 iface = iface_cache.get_interface('http://foo')
  71                 with tempfile.TemporaryFile() as src:
  72                         src.write(data.foo_signed_xml)
  73                         src.seek(0)
  74                         pending = PendingFeed(iface.uri, src)
  75                         assert iface_cache.update_feed_if_trusted(iface.uri, pending.sigs, pending.new_xml)
  76
  77                 iface_cache.__init__()
  78                 feed = iface_cache.get_feed('http://foo')
  79                 assert feed.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(feed.url))
  85                 os.utime(cached, None)
  86
  87                 iface_cache.__init__()
  88                 feed = iface_cache.get_feed('http://foo')
  89                 assert feed.last_modified > 1154850229
  90
  91                 with tempfile.TemporaryFile() as src:
  92                         src.write(data.new_foo_signed_xml)
  93                         src.seek(0)
  94
  95                         pending = PendingFeed(feed.url, src)
  96                         assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)
  97
  98                 # Can't 'update' to an older copy
  99                 with tempfile.TemporaryFile() as src:
 100                         src.write(data.foo_signed_xml)
 101                         src.seek(0)
 102                         try:
 103                                 pending = PendingFeed(feed.url, src)
 104                                 assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)
 105
 106                                 assert 0
 107                         except model.SafeException:
 108                                 pass
 109
 110         def testTimes(self):
 111                 iface_cache = self.config.iface_cache
 112                 with tempfile.TemporaryFile() as stream:
 113                         stream.write(data.thomas_key)
 114                         stream.seek(0)
 115                         gpg.import_key(stream)
 116
 117                 upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
 118                 cached = os.path.join(upstream_dir, model.escape('http://foo'))
 119
 120                 with open(cached, 'wb') as stream:
 121                         stream.write(data.foo_signed_xml)
 122
 123                 signed = iface_cache._get_signature_date('http://foo')
 124                 assert signed == None
 125
 126                 trust.trust_db.trust_key(
 127                         '92429807C9853C0744A68B9AAE07828059A53CC1')
 128
 129                 signed = iface_cache._get_signature_date('http://foo')
 130                 self.assertEqual(1154850229, signed)
 131
 132                 with open(cached, 'w+b') as stream:
 133                         stream.seek(0)
 134                         stream.write(b'Hello')
 135
 136                 # When the signature is invalid, we just return None.
 137                 # This is because versions < 0.22 used to corrupt the signatue
 138                 # by adding an attribute to the XML
 139                 signed = iface_cache._get_signature_date('http://foo')
 140                 assert signed == None
 141
 142         def testCheckAttempt(self):
 143                 iface_cache = self.config.iface_cache
 144                 self.assertEqual(None, iface_cache.get_last_check_attempt("http://foo/bar.xml"))
 145
 146                 start_time = time.time() - 5    # Seems to be some odd rounding here
 147                 iface_cache.mark_as_checking("http://foo/bar.xml")
 148                 last_check = iface_cache.get_last_check_attempt("http://foo/bar.xml")
 149
 150                 assert last_check is not None
 151                 assert last_check >= start_time, (last_check, start_time)
 152
 153                 self.assertEqual(None, iface_cache.get_last_check_attempt("http://foo/bar2.xml"))
 154
 155         def testIsStale(self):
 156                 iface_cache = self.config.iface_cache
 157                 feed = self.import_feed('http://localhost:8000/Hello', 'Hello')
 158                 assert iface_cache.is_stale(feed.url, 1) == True
 159                 assert iface_cache.is_stale(feed.url, time.time() + 1) == False
 160                 iface_cache.mark_as_checking(feed.url)
 161                 assert iface_cache.is_stale(feed.url, 1) == False
 162
 163                 # Old API
 164                 assert iface_cache.is_stale(feed, 1) == False
 165
 166 if __name__ == '__main__':
 167         unittest.main()