tests/testdownload.py

   1 #!/usr/bin/env python
   2 from __future__ import with_statement
   3 from basetest import BaseTest
   4 import sys, tempfile, os
   5 from StringIO import StringIO
   6 import unittest, signal
   7 from logging import getLogger, WARN, ERROR
   8 from contextlib import contextmanager
   9
  10 sys.path.insert(0, '..')
  11
  12 os.environ["http_proxy"] = "localhost:8000"
  13
  14 from zeroinstall import helpers
  15 from zeroinstall.injector import model, gpg, download, trust, background, arch, selections, qdom, run
  16 from zeroinstall.injector.requirements import Requirements
  17 from zeroinstall.injector.driver import Driver
  18 from zeroinstall.zerostore import Store, NotStored; Store._add_with_helper = lambda *unused: False
  19 from zeroinstall.support import basedir, tasks, ro_rmtree
  20 from zeroinstall.injector import fetch
  21 import data
  22 import my_dbus
  23
  24 import server
  25
  26 ran_gui = False
  27 def raise_gui(*args):
  28         global ran_gui
  29         ran_gui = True
  30 background._detach = lambda: False
  31
  32 local_hello = """<?xml version="1.0" ?>
  33 <selections command="run" interface="http://example.com:8000/Hello.xml" xmlns="http://zero-install.sourceforge.net/2004/injector/interface">
  34   <selection id="." local-path='.' interface="http://example.com:8000/Hello.xml" version="0.1"><command name="run" path="foo"/></selection>
  35 </selections>"""
  36
  37 @contextmanager
  38 def output_suppressed():
  39         old_stdout = sys.stdout
  40         old_stderr = sys.stderr
  41         try:
  42                 sys.stdout = StringIO()
  43                 sys.stderr = StringIO()
  44                 try:
  45                         yield
  46                 except Exception:
  47                         raise
  48                 except BaseException as ex:
  49                         # Don't abort unit-tests if someone raises SystemExit
  50                         raise Exception(str(type(ex)) + " " + str(ex))
  51         finally:
  52                 sys.stdout = old_stdout
  53                 sys.stderr = old_stderr
  54
  55 @contextmanager
  56 def trapped_exit(expected_exit_status):
  57         pid = os.getpid()
  58         old_exit = os._exit
  59         def my_exit(code):
  60                 # The background handler runs in the same process
  61                 # as the tests, so don't let it abort.
  62                 if os.getpid() == pid:
  63                         raise SystemExit(code)
  64                 # But, child download processes are OK
  65                 old_exit(code)
  66         os._exit = my_exit
  67         try:
  68                 try:
  69                         yield
  70                         assert False
  71                 except SystemExit as ex:
  72                         assert ex.code == expected_exit_status
  73         finally:
  74                 os._exit = old_exit
  75
  76 class Reply:
  77         def __init__(self, reply):
  78                 self.reply = reply
  79
  80         def readline(self):
  81                 return self.reply
  82
  83 def download_and_execute(driver, prog_args, main = None):
  84         downloaded = driver.solve_and_download_impls()
  85         if downloaded:
  86                 tasks.wait_for_blocker(downloaded)
  87         run.execute_selections(driver.solver.selections, prog_args, stores = driver.config.stores, main = main)
  88
  89 class NetworkManager:
  90         def state(self):
  91                 return 3        # NM_STATUS_CONNECTED
  92
  93 server_process = None
  94 def kill_server_process():
  95         global server_process
  96         if server_process is not None:
  97                 os.kill(server_process, signal.SIGTERM)
  98                 os.waitpid(server_process, 0)
  99                 server_process = None
 100
 101 def run_server(*args):
 102         global server_process
 103         assert server_process is None
 104         server_process = server.handle_requests(*args)
 105
 106 real_get_selections_gui = helpers.get_selections_gui
 107
 108 class TestDownload(BaseTest):
 109         def setUp(self):
 110                 BaseTest.setUp(self)
 111
 112                 self.config.handler.allow_downloads = True
 113                 self.config.key_info_server = 'http://localhost:3333/key-info'
 114
 115                 self.config.fetcher = fetch.Fetcher(self.config)
 116
 117                 stream = tempfile.TemporaryFile()
 118                 stream.write(data.thomas_key)
 119                 stream.seek(0)
 120                 gpg.import_key(stream)
 121                 stream.close()
 122
 123                 trust.trust_db.watchers = []
 124
 125                 helpers.get_selections_gui = raise_gui
 126
 127                 global ran_gui
 128                 ran_gui = False
 129
 130         def tearDown(self):
 131                 helpers.get_selections_gui = real_get_selections_gui
 132                 BaseTest.tearDown(self)
 133                 kill_server_process()
 134
 135         def testRejectKey(self):
 136                 with output_suppressed():
 137                         run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
 138                         driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
 139                         assert driver.need_download()
 140                         sys.stdin = Reply("N\n")
 141                         try:
 142                                 download_and_execute(driver, ['Hello'])
 143                                 assert 0
 144                         except model.SafeException as ex:
 145                                 if "has no usable implementations" not in str(ex):
 146                                         raise ex
 147                                 if "Not signed with a trusted key" not in str(self.config.handler.ex):
 148                                         raise ex
 149                                 self.config.handler.ex = None
 150
 151         def testRejectKeyXML(self):
 152                 with output_suppressed():
 153                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
 154                         driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
 155                         assert driver.need_download()
 156                         sys.stdin = Reply("N\n")
 157                         try:
 158                                 download_and_execute(driver, ['Hello'])
 159                                 assert 0
 160                         except model.SafeException as ex:
 161                                 if "has no usable implementations" not in str(ex):
 162                                         raise ex
 163                                 if "Not signed with a trusted key" not in str(self.config.handler.ex):
 164                                         raise
 165                                 self.config.handler.ex = None
 166
 167         def testImport(self):
 168                 from zeroinstall.injector import cli
 169
 170                 rootLogger = getLogger()
 171                 rootLogger.disabled = True
 172                 try:
 173                         try:
 174                                 cli.main(['--import', '-v', 'NO-SUCH-FILE'], config = self.config)
 175                                 assert 0
 176                         except model.SafeException as ex:
 177                                 assert 'NO-SUCH-FILE' in str(ex)
 178                 finally:
 179                         rootLogger.disabled = False
 180                         rootLogger.setLevel(WARN)
 181
 182                 hello = self.config.iface_cache.get_feed('http://localhost:8000/Hello')
 183                 self.assertEqual(None, hello)
 184
 185                 with output_suppressed():
 186                         run_server('6FCF121BE2390E0B.gpg')
 187                         sys.stdin = Reply("Y\n")
 188
 189                         assert not trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 190                         cli.main(['--import', 'Hello'], config = self.config)
 191                         assert trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 192
 193                         # Check we imported the interface after trusting the key
 194                         hello = self.config.iface_cache.get_feed('http://localhost:8000/Hello', force = True)
 195                         self.assertEqual(1, len(hello.implementations))
 196
 197                         self.assertEqual(None, hello.local_path)
 198
 199                         # Shouldn't need to prompt the second time
 200                         sys.stdin = None
 201                         cli.main(['--import', 'Hello'], config = self.config)
 202
 203         def testSelections(self):
 204                 from zeroinstall.injector import cli
 205                 root = qdom.parse(open("selections.xml"))
 206                 sels = selections.Selections(root)
 207                 class Options: dry_run = False
 208
 209                 with output_suppressed():
 210                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 211                         sys.stdin = Reply("Y\n")
 212                         try:
 213                                 self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
 214                                 assert False
 215                         except NotStored:
 216                                 pass
 217                         cli.main(['--download-only', 'selections.xml'], config = self.config)
 218                         path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
 219                         assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
 220
 221                         assert sels.download_missing(self.config) is None
 222
 223         def testHelpers(self):
 224                 from zeroinstall import helpers
 225
 226                 with output_suppressed():
 227                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 228                         sys.stdin = Reply("Y\n")
 229                         sels = helpers.ensure_cached('http://example.com:8000/Hello.xml', config = self.config)
 230                         path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
 231                         assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
 232                         assert sels.download_missing(self.config) is None
 233
 234         def testSelectionsWithFeed(self):
 235                 from zeroinstall.injector import cli
 236                 root = qdom.parse(open("selections.xml"))
 237                 sels = selections.Selections(root)
 238
 239                 with output_suppressed():
 240                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 241                         sys.stdin = Reply("Y\n")
 242
 243                         tasks.wait_for_blocker(self.config.fetcher.download_and_import_feed('http://example.com:8000/Hello.xml', self.config.iface_cache))
 244
 245                         cli.main(['--download-only', 'selections.xml'], config = self.config)
 246                         path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
 247                         assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
 248
 249                         assert sels.download_missing(self.config) is None
 250
 251         def testAcceptKey(self):
 252                 with output_suppressed():
 253                         run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 254                         driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
 255                         assert driver.need_download()
 256                         sys.stdin = Reply("Y\n")
 257                         try:
 258                                 download_and_execute(driver, ['Hello'], main = 'Missing')
 259                                 assert 0
 260                         except model.SafeException as ex:
 261                                 if "HelloWorld/Missing" not in str(ex):
 262                                         raise
 263
 264         def testAutoAcceptKey(self):
 265                 self.config.auto_approve_keys = True
 266                 with output_suppressed():
 267                         run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 268                         driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
 269                         assert driver.need_download()
 270                         sys.stdin = Reply("")
 271                         try:
 272                                 download_and_execute(driver, ['Hello'], main = 'Missing')
 273                                 assert 0
 274                         except model.SafeException as ex:
 275                                 if "HelloWorld/Missing" not in str(ex):
 276                                         raise
 277
 278         def testDistro(self):
 279                 with output_suppressed():
 280                         native_url = 'http://example.com:8000/Native.xml'
 281
 282                         # Initially, we don't have the feed at all...
 283                         master_feed = self.config.iface_cache.get_feed(native_url)
 284                         assert master_feed is None, master_feed
 285
 286                         trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
 287                         run_server('Native.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
 288                         driver = Driver(requirements = Requirements(native_url), config = self.config)
 289                         assert driver.need_download()
 290
 291                         solve = driver.solve_with_downloads()
 292                         tasks.wait_for_blocker(solve)
 293                         tasks.check(solve)
 294
 295                         master_feed = self.config.iface_cache.get_feed(native_url)
 296                         assert master_feed is not None
 297                         assert master_feed.implementations == {}
 298
 299                         distro_feed_url = master_feed.get_distro_feed()
 300                         assert distro_feed_url is not None
 301                         distro_feed = self.config.iface_cache.get_feed(distro_feed_url)
 302                         assert distro_feed is not None
 303                         assert len(distro_feed.implementations) == 2, distro_feed.implementations
 304
 305         def testWrongSize(self):
 306                 with output_suppressed():
 307                         run_server('Hello-wrong-size', '6FCF121BE2390E0B.gpg',
 308                                                         '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
 309                         driver = Driver(requirements = Requirements('http://localhost:8000/Hello-wrong-size'), config = self.config)
 310                         assert driver.need_download()
 311                         sys.stdin = Reply("Y\n")
 312                         try:
 313                                 download_and_execute(driver, ['Hello'], main = 'Missing')
 314                                 assert 0
 315                         except model.SafeException as ex:
 316                                 if "Downloaded archive has incorrect size" not in str(ex):
 317                                         raise ex
 318
 319         def testRecipe(self):
 320                 old_out = sys.stdout
 321                 try:
 322                         sys.stdout = StringIO()
 323                         run_server(('HelloWorld.tar.bz2', 'redirect/dummy_1-1_all.deb', 'dummy_1-1_all.deb'))
 324                         driver = Driver(requirements = Requirements(os.path.abspath('Recipe.xml')), config = self.config)
 325                         try:
 326                                 download_and_execute(driver, [])
 327                                 assert False
 328                         except model.SafeException as ex:
 329                                 if "HelloWorld/Missing" not in str(ex):
 330                                         raise ex
 331                 finally:
 332                         sys.stdout = old_out
 333
 334         def testSymlink(self):
 335                 old_out = sys.stdout
 336                 try:
 337                         sys.stdout = StringIO()
 338                         run_server(('HelloWorld.tar.bz2', 'HelloSym.tgz'))
 339                         driver = Driver(requirements = Requirements(os.path.abspath('RecipeSymlink.xml')), config = self.config)
 340                         try:
 341                                 download_and_execute(driver, [])
 342                                 assert False
 343                         except model.SafeException as ex:
 344                                 if 'Attempt to unpack dir over symlink "HelloWorld"' not in str(ex):
 345                                         raise
 346                         self.assertEqual(None, basedir.load_first_cache('0install.net', 'implementations', 'main'))
 347                 finally:
 348                         sys.stdout = old_out
 349
 350         def testAutopackage(self):
 351                 old_out = sys.stdout
 352                 try:
 353                         sys.stdout = StringIO()
 354                         run_server('HelloWorld.autopackage')
 355                         driver = Driver(requirements = Requirements(os.path.abspath('Autopackage.xml')), config = self.config)
 356                         try:
 357                                 download_and_execute(driver, [])
 358                                 assert False
 359                         except model.SafeException as ex:
 360                                 if "HelloWorld/Missing" not in str(ex):
 361                                         raise
 362                 finally:
 363                         sys.stdout = old_out
 364
 365         def testRecipeFailure(self):
 366                 old_out = sys.stdout
 367                 try:
 368                         sys.stdout = StringIO()
 369                         run_server('*')
 370                         driver = Driver(requirements = Requirements(os.path.abspath('Recipe.xml')), config = self.config)
 371                         try:
 372                                 download_and_execute(driver, [])
 373                                 assert False
 374                         except download.DownloadError as ex:
 375                                 if "Connection" not in str(ex):
 376                                         raise
 377                 finally:
 378                         sys.stdout = old_out
 379
 380         def testMirrors(self):
 381                 old_out = sys.stdout
 382                 try:
 383                         sys.stdout = StringIO()
 384                         getLogger().setLevel(ERROR)
 385                         trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
 386                         run_server(server.Give404('/Hello.xml'),
 387                                         '/0mirror/feeds/http/example.com:8000/Hello.xml/latest.xml',
 388                                         '/0mirror/keys/6FCF121BE2390E0B.gpg')
 389                         driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
 390                         self.config.feed_mirror = 'http://example.com:8000/0mirror'
 391
 392                         refreshed = driver.solve_with_downloads()
 393                         tasks.wait_for_blocker(refreshed)
 394                         assert driver.solver.ready
 395                 finally:
 396                         sys.stdout = old_out
 397
 398         def testReplay(self):
 399                 old_out = sys.stdout
 400                 try:
 401                         sys.stdout = StringIO()
 402                         getLogger().setLevel(ERROR)
 403                         iface = self.config.iface_cache.get_interface('http://example.com:8000/Hello.xml')
 404                         mtime = int(os.stat('Hello-new.xml').st_mtime)
 405                         self.config.iface_cache.update_feed_from_network(iface.uri, open('Hello-new.xml').read(), mtime + 10000)
 406
 407                         trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
 408                         run_server(server.Give404('/Hello.xml'), 'latest.xml', '/0mirror/keys/6FCF121BE2390E0B.gpg', 'Hello.xml')
 409                         driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
 410                         self.config.feed_mirror = 'http://example.com:8000/0mirror'
 411
 412                         # Update from mirror (should ignore out-of-date timestamp)
 413                         refreshed = self.config.fetcher.download_and_import_feed(iface.uri, self.config.iface_cache)
 414                         tasks.wait_for_blocker(refreshed)
 415
 416                         # Update from upstream (should report an error)
 417                         refreshed = self.config.fetcher.download_and_import_feed(iface.uri, self.config.iface_cache)
 418                         try:
 419                                 tasks.wait_for_blocker(refreshed)
 420                                 raise Exception("Should have been rejected!")
 421                         except model.SafeException as ex:
 422                                 assert "New feed's modification time is before old version" in str(ex)
 423
 424                         # Must finish with the newest version
 425                         self.assertEqual(1235911552, self.config.iface_cache._get_signature_date(iface.uri))
 426                 finally:
 427                         sys.stdout = old_out
 428
 429         def testBackground(self, verbose = False):
 430                 r = Requirements('http://example.com:8000/Hello.xml')
 431                 d = Driver(requirements = r, config = self.config)
 432                 self.import_feed(r.interface_uri, 'Hello.xml')
 433                 self.config.freshness = 0
 434                 self.config.network_use = model.network_minimal
 435                 d.solver.solve(r.interface_uri, arch.get_host_architecture())
 436                 assert d.solver.ready, d.solver.get_failure_reason()
 437
 438                 @tasks.async
 439                 def choose_download(registed_cb, nid, actions):
 440                         try:
 441                                 assert actions == ['download', 'Download'], actions
 442                                 registed_cb(nid, 'download')
 443                         except:
 444                                 import traceback
 445                                 traceback.print_exc()
 446                         yield None
 447
 448                 global ran_gui
 449                 ran_gui = False
 450                 os.environ['DISPLAY'] = 'dummy'
 451                 old_out = sys.stdout
 452                 try:
 453                         sys.stdout = StringIO()
 454                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg')
 455                         my_dbus.system_services = {"org.freedesktop.NetworkManager": {"/org/freedesktop/NetworkManager": NetworkManager()}}
 456                         my_dbus.user_callback = choose_download
 457
 458                         with trapped_exit(1):
 459                                 from zeroinstall.injector import config
 460                                 key_info = config.DEFAULT_KEY_LOOKUP_SERVER
 461                                 config.DEFAULT_KEY_LOOKUP_SERVER = None
 462                                 try:
 463                                         background.spawn_background_update(d, verbose)
 464                                 finally:
 465                                         config.DEFAULT_KEY_LOOKUP_SERVER = key_info
 466                 finally:
 467                         sys.stdout = old_out
 468                 assert ran_gui
 469
 470         def testBackgroundVerbose(self):
 471                 self.testBackground(verbose = True)
 472
 473         def testBackgroundApp(self):
 474                 my_dbus.system_services = {"org.freedesktop.NetworkManager": {"/org/freedesktop/NetworkManager": NetworkManager()}}
 475
 476                 trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
 477
 478                 global ran_gui
 479
 480                 with output_suppressed():
 481                         # Select a version of Hello
 482                         run_server('Hello.xml', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
 483                         r = Requirements('http://example.com:8000/Hello.xml')
 484                         driver = Driver(requirements = r, config = self.config)
 485                         tasks.wait_for_blocker(driver.solve_with_downloads())
 486                         assert driver.solver.ready
 487                         kill_server_process()
 488
 489                         # Save it as an app
 490                         app = self.config.app_mgr.create_app('test-app', r)
 491                         app.set_selections(driver.solver.selections)
 492                         timestamp = os.path.join(app.path, 'last-checked')
 493                         last_check_attempt = os.path.join(app.path, 'last-check-attempt')
 494                         selections_path = os.path.join(app.path, 'selections.xml')
 495
 496                         def reset_timestamps():
 497                                 ran_gui = False
 498                                 os.utime(timestamp, (1, 1))             # 1970
 499                                 os.utime(selections_path, (1, 1))
 500                                 if os.path.exists(last_check_attempt):
 501                                         os.unlink(last_check_attempt)
 502
 503                         # Download the implementation
 504                         sels = app.get_selections()
 505                         run_server('HelloWorld.tgz')
 506                         tasks.wait_for_blocker(app.download_selections(sels))
 507                         kill_server_process()
 508
 509                         # Not time for a background update yet
 510                         self.config.freshness = 100
 511                         dl = app.download_selections(sels)
 512                         assert dl == None
 513                         assert not ran_gui
 514
 515                         # Trigger a background update - no updates found
 516                         reset_timestamps()
 517                         run_server('Hello.xml')
 518                         with trapped_exit(1):
 519                                 dl = app.download_selections(sels)
 520                                 assert dl == None
 521                         assert not ran_gui
 522                         self.assertNotEqual(1, os.stat(timestamp).st_mtime)
 523                         self.assertEqual(1, os.stat(selections_path).st_mtime)
 524                         kill_server_process()
 525
 526                         # Change the selections
 527                         sels_path = os.path.join(app.path, 'selections.xml')
 528                         with open(sels_path) as stream:
 529                                 old = stream.read()
 530                         with open(sels_path, 'w') as stream:
 531                                 stream.write(old.replace('Hello', 'Goodbye'))
 532
 533                         # Trigger another background update - metadata changes found
 534                         reset_timestamps()
 535                         run_server('Hello.xml')
 536                         with trapped_exit(1):
 537                                 dl = app.download_selections(sels)
 538                                 assert dl == None
 539                         assert not ran_gui
 540                         self.assertNotEqual(1, os.stat(timestamp).st_mtime)
 541                         self.assertNotEqual(1, os.stat(selections_path).st_mtime)
 542                         kill_server_process()
 543
 544                         # Trigger another background update - GUI needed now
 545
 546                         # Delete cached implementation so we need to download it again
 547                         stored = sels.selections['http://example.com:8000/Hello.xml'].get_path(self.config.stores)
 548                         assert os.path.basename(stored).startswith('sha1')
 549                         ro_rmtree(stored)
 550
 551                         # Replace with a valid local feed so we don't have to download immediately
 552                         with open(sels_path, 'w') as stream:
 553                                 stream.write(local_hello)
 554                         sels = app.get_selections()
 555
 556                         os.environ['DISPLAY'] = 'dummy'
 557                         reset_timestamps()
 558                         run_server('Hello.xml')
 559                         with trapped_exit(1):
 560                                 dl = app.download_selections(sels)
 561                                 assert dl == None
 562                         assert ran_gui  # (so doesn't actually update)
 563                         kill_server_process()
 564
 565                         # Now again with no DISPLAY
 566                         reset_timestamps()
 567                         del os.environ['DISPLAY']
 568                         run_server('Hello.xml', 'HelloWorld.tgz')
 569                         with trapped_exit(1):
 570                                 dl = app.download_selections(sels)
 571                                 assert dl == None
 572                         assert ran_gui  # (so doesn't actually update)
 573
 574                         self.assertNotEqual(1, os.stat(timestamp).st_mtime)
 575                         self.assertNotEqual(1, os.stat(selections_path).st_mtime)
 576                         kill_server_process()
 577
 578                         sels = app.get_selections()
 579                         sel, = sels.selections.values()
 580                         self.assertEqual("sha1=3ce644dc725f1d21cfcf02562c76f375944b266a", sel.id)
 581
 582                         # Untrust the key
 583                         trust.trust_db.untrust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
 584
 585                         os.environ['DISPLAY'] = 'dummy'
 586                         reset_timestamps()
 587                         run_server('Hello.xml')
 588                         with trapped_exit(1):
 589                                 #import logging; logging.getLogger().setLevel(logging.INFO)
 590                                 dl = app.download_selections(sels)
 591                                 assert dl == None
 592                         assert ran_gui
 593                         kill_server_process()
 594
 595                         # Update not triggered because of last-check-attempt
 596                         ran_gui = False
 597                         os.utime(timestamp, (1, 1))             # 1970
 598                         os.utime(selections_path, (1, 1))
 599                         dl = app.download_selections(sels)
 600                         assert dl == None
 601                         assert not ran_gui
 602
 603 if __name__ == '__main__':
 604         try:
 605                 unittest.main()
 606         finally:
 607                 kill_server_process()