2 from basetest
import BaseTest
3 import sys
, tempfile
, os
, time
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
13 class TestIfaceCache(BaseTest
):
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
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
:
34 PendingFeed(feed_url
, src
)
36 except model
.SafeException
:
39 with tempfile
.TemporaryFile(mode
= 'w+b') as stream
:
40 stream
.write(data
.thomas_key
)
43 gpg
.import_key(stream
)
47 src
.write(data
.foo_signed_xml
)
51 pending
= PendingFeed(feed_url
, src
)
52 assert iface_cache
.update_feed_if_trusted(feed_url
, pending
.sigs
, pending
.new_xml
)
54 self
.assertEqual(['http://foo'],
55 iface_cache
.list_all_interfaces())
57 feed
= iface_cache
.get_feed(feed_url
)
59 self
.assertEqual(1154850229, feed
.last_modified
)
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
)
68 gpg
.import_key(stream
)
70 iface
= iface_cache
.get_interface('http://foo')
71 with tempfile
.TemporaryFile() as src
:
72 src
.write(data
.foo_signed_xml
)
74 pending
= PendingFeed(iface
.uri
, src
)
75 assert iface_cache
.update_feed_if_trusted(iface
.uri
, pending
.sigs
, pending
.new_xml
)
77 iface_cache
.__init
__()
78 feed
= iface_cache
.get_feed('http://foo')
79 assert feed
.last_modified
== 1154850229
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)
87 iface_cache
.__init
__()
88 feed
= iface_cache
.get_feed('http://foo')
89 assert feed
.last_modified
> 1154850229
91 with tempfile
.TemporaryFile() as src
:
92 src
.write(data
.new_foo_signed_xml
)
95 pending
= PendingFeed(feed
.url
, src
)
96 assert iface_cache
.update_feed_if_trusted(feed
.url
, pending
.sigs
, pending
.new_xml
)
98 # Can't 'update' to an older copy
99 with tempfile
.TemporaryFile() as src
:
100 src
.write(data
.foo_signed_xml
)
103 pending
= PendingFeed(feed
.url
, src
)
104 assert iface_cache
.update_feed_if_trusted(feed
.url
, pending
.sigs
, pending
.new_xml
)
107 except model
.SafeException
:
111 iface_cache
= self
.config
.iface_cache
112 with tempfile
.TemporaryFile() as stream
:
113 stream
.write(data
.thomas_key
)
115 gpg
.import_key(stream
)
117 upstream_dir
= basedir
.save_cache_path(config_site
, 'interfaces')
118 cached
= os
.path
.join(upstream_dir
, model
.escape('http://foo'))
120 with
open(cached
, 'wb') as stream
:
121 stream
.write(data
.foo_signed_xml
)
123 signed
= iface_cache
._get
_signature
_date
('http://foo')
124 assert signed
== None
126 trust
.trust_db
.trust_key(
127 '92429807C9853C0744A68B9AAE07828059A53CC1')
129 signed
= iface_cache
._get
_signature
_date
('http://foo')
130 self
.assertEqual(1154850229, signed
)
132 with
open(cached
, 'w+b') as stream
:
134 stream
.write(b
'Hello')
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
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"))
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")
150 assert last_check
is not None
151 assert last_check
>= start_time
, (last_check
, start_time
)
153 self
.assertEqual(None, iface_cache
.get_last_check_attempt("http://foo/bar2.xml"))
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
164 assert iface_cache
.is_stale(feed
, 1) == False
166 if __name__
== '__main__':