from __future__ import absolute_import, with_statement import unittest from datetime import datetime from hashlib import md5 import mox import xmlrpclib from abl.installer.installer import Updater from abl.util import Bunch from abl.vpath.base import URI from abl.vpath.base.fs import CONNECTION_REGISTRY from abl.webconnector import DownloadsNotFinished, RPCConnector, WrongChecksum from abl.webconnector.auto_updates import DownloadTask, PrepareAutoUpdateTask from abl.webconnector.connector import ConnectorBase, InstallerInvoker from abl.webconnector.reporting import EventTypes as et from cStringIO import StringIO from .common import FakeTransport, FakeUsageReporter, create_file class MockTempDir(object): def __init__(self): self.temp_dir = URI("memory:///temp") self.temp_dir.mkdir() def get_dir(self): return self.temp_dir class MockPopenFactory(object): class MockPopen(object): def communicate(self): return None, None def poll(self): return None def __init__(self): self.captured_args = [] self.captured_kwargs = [] def spawn(self, *args, **kwargs): self.captured_args.append(args) self.captured_kwargs.append(kwargs) return self.MockPopen() class SuccessfulUpdater(Updater): def prepare_auto_update(self, v_from, v_to, cleanup=True): pass class FailingUpdater(Updater): def prepare_auto_update(self, v_from, v_to, cleanup=True): raise Exception("Hi, I'm an exception, but I'm completely expected.") class UpdaterFailingInInit(Updater): def __init__(self, app_base_path, copy_app_base_path): raise Exception("No worries, I'm an expected exception.") class UpdaterFailingInClone(Updater): def clone_installation(self): """This function only creates the base folder of the "clone", so it will mismatch the original, which has sub-folders.""" self.copy_app_base_path.mkdir() class AutoUpdateTests(unittest.TestCase): def tearDown(self): CONNECTION_REGISTRY.cleanup(force=True) super(AutoUpdateTests, self).tearDown() def test_info_db(self): app_base_path = URI("memory:///app_base_path") app_base_path.mkdir() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() connector = ConnectorBase( Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), ) ) self.assertEqual(connector.load_update_info(), {}) data = {"foo": "bar"} connector.save_update_info(data) self.assertEqual(connector.load_update_info(), data) # ------------------------------------------------------------------------------------- def test_update_retrieval_no_update_available(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn({}) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) assert not res assert not connector.pending_update_available() mocker.VerifyAll() assert len(fake_reporter.events) == 2 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.NO_UPDATES_AVAILABLE in fake_reporter.events # ------------------------------------------------------------------------------------- def test_update_retrieval_update_available(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) self.assertEqual(len(res), 2) update_info = connector.load_update_info() # The update isn't "pending" yet because we didn't execute the "tasks" assert not connector.pending_update_available() self.assertEqual( update_info, { "prepared_version": None, "versions": { "from": {"to": {"deltas": [delta_filename], "status": "downloading"}} }, "delta_infos": {delta_filename: dict(checksum=checksum, version=1)}, }, ) assert len(fake_reporter.events) == 2 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events # ------------------------------------------------------------------------------------- def test_download_and_prepare_deltas(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(SuccessfulUpdater) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) for task in res: task.perform(fake_reporter) assert connector.pending_update_available() update_info = connector.load_update_info() self.assertEqual(update_info["versions"]["from"]["to"]["status"], "prepared") self.assertEqual(update_info["prepared_version"], ["from", "to"]) assert len(fake_reporter.events) == 9 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.COPIED_INSTALL in fake_reporter.events assert et.UPDATE_COPY in fake_reporter.events assert et.UPDATED_COPY in fake_reporter.events # ------------------------------------------------------------------------------------- def test_update_retrieval_and_preparation(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( autoupdates_dir_base_path=str(autoupdates_dir_base_path), pid=5678, app_base_path=str(app_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() # now we find an update, # and download it. data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(SuccessfulUpdater) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) # download the delta files # and prepare for task in res: task.perform(fake_reporter) assert connector.pending_update_available() # now swap the two prepared copies connector.apply_available_auto_update(fake_reporter) # assert that the swapper was correctly called self.assertEqual(len(mock_popen_factory.captured_args), 1) self.assertEqual(len(mock_popen_factory.captured_kwargs), 1) swapper_cmd = " ".join(mock_popen_factory.captured_args[0][0]) assert str(mock_temp_dir.temp_dir / swapper_path.basename()) in swapper_cmd assert "--live=" + str(app_base_path) in swapper_cmd assert "--copy" in swapper_cmd assert "--usagedata" in swapper_cmd assert "pillepalle" in swapper_cmd assert "foobarbaz.swapper" in swapper_cmd assert "--parentpid=5678" in swapper_cmd assert mock_popen_factory.captured_kwargs[0]["cwd"] == str(mock_temp_dir.temp_dir) # check that the swapper was really copied to the temp location assert (mock_temp_dir.temp_dir / swapper_path.basename()).exists() # finally, make sure all calls # have been made. mocker.VerifyAll() assert len(fake_reporter.events) == 10 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.COPIED_INSTALL in fake_reporter.events assert et.UPDATE_COPY in fake_reporter.events assert et.UPDATED_COPY in fake_reporter.events assert et.INVOKING_SWAPPER in fake_reporter.events # ------------------------------------------------------------------------------------- def test_update_failed_dont_swap(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() # now we find an update, # and download it. data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(FailingUpdater) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) # download the delta files # and prepare for task in res: task.perform(fake_reporter) # at this point, we would check that the pending update is available, and, if so, # ask the connector to apply it. In this case, we'll just check that the connector # claims that no pending update is available. assert not connector.pending_update_available() # finally, make sure all calls # have been made. mocker.VerifyAll() assert len(fake_reporter.events) == 9 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.COPIED_INSTALL in fake_reporter.events assert et.UPDATE_COPY in fake_reporter.events assert et.EXCEPTION in fake_reporter.events # ------------------------------------------------------------------------------------- def test_copying_install_fails(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() # now we find an update, # and download it. data = "a" * 100000 delta_filename = "delta.asu" checksum = md5(data).hexdigest() cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(UpdaterFailingInInit) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) # download the delta files # and prepare for task in res: task.perform(fake_reporter) # at this point, we would check that the pending update is available, and, if so, # ask the connector to apply it. In this case, we'll just check that the connector # claims that no pending update is available. assert not connector.pending_update_available() # finally, make sure all calls # have been made. mocker.VerifyAll() assert len(fake_reporter.events) == 7 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.EXCEPTION in fake_reporter.events # ------------------------------------------------------------------------------------- def test_sanity_check_clone_failed(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() # now we find an update, # and download it. data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(UpdaterFailingInClone) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) # download the delta files # and prepare for task in res: task.perform(fake_reporter) # at this point, we would check that the pending update is available, and, if so, # ask the connector to apply it. In this case, we'll just check that the connector # claims that no pending update is available. assert not connector.pending_update_available() # this should assert # finally, make sure all calls # have been made. mocker.VerifyAll() assert len(fake_reporter.events) == 7 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.EXCEPTION in fake_reporter.events assert ( fake_reporter.params[fake_reporter.events.index(et.EXCEPTION)].find( "failed to clone Live" ) != -1 ) # ------------------------------------------------------------------------------------- def test_copying_with_symlinks(self): app_base_path = URI("memory:///app_base_path") / "live_beta.app" app_base_path.makedirs() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() resources_path = app_base_path / "Resources" resources_path.makedirs() # create a symlink in the app folder to an existing script (e.g. oxygen8_path = resources_path / "MIDI Remote Scripts" / "oxygen8" oxygen8_path.makedirs() create_file(oxygen8_path / "__init__.pyc") controller_path = resources_path / "MIDI Remote Scripts" / "humptydumpty" oxygen8_path.symlink(controller_path) # create another symlink to an "external" file external_path = URI("memory:///") / "Library" / "Scripts" external_path.makedirs() external_controller_path = external_path / "mycontroller.py" create_file(external_controller_path) controller2_path = resources_path / "MIDI Remote Scripts" / "acontroller.py" external_controller_path.symlink(controller2_path) # create a broken symlink - a loop to itself loop_path = resources_path / "MIDI Remote Scripts" / "ixwick.plue" loop_path.symlink(loop_path) # and another broken symlink - now one over multiple levels loop_dir_path = resources_path / "MIDI Remote Scripts" / "foo" resources_path.symlink(loop_dir_path) live_path = app_base_path / "live.exe" live_path.mkdir() swapper_dir = app_base_path / "Extensions" swapper_dir.mkdir() swapper_path = swapper_dir / "Swapper" usage_report_prefix = "foobarbaz" aip_usage_data_dir = "/pillepalle" config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", app_exe_path=str(live_path), logging=dict(log_level="DEBUG", file_path="/tmp/logfile"), usage_report_prefix=usage_report_prefix, aip_usage_data_dir=aip_usage_data_dir, swapper_exe=str(swapper_path), ) tags = [] version = "from" # Make a dummy file for the swapper create_file(swapper_path, content="foobar") mock_popen_factory = MockPopenFactory() mock_temp_dir = MockTempDir() # now we find an update, # and download it. data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mock_conn.download_link_for_delta(delta_filename).AndReturn(cdn_link) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, temp_dir_accessor=mock_temp_dir.get_dir, ) connector.invoker = InstallerInvoker(SuccessfulUpdater) fake_reporter = FakeUsageReporter() res = connector.check_and_process_auto_updates(tags, version, fake_reporter) # download the delta files and prepare for task in res: task.perform(fake_reporter) assert connector.pending_update_available() # now swap the two prepared copies connector.apply_available_auto_update(fake_reporter) # check that the swapper was really copied to the temp location assert (mock_temp_dir.temp_dir / swapper_path.basename()).exists() # finally, make sure all calls # have been made. mocker.VerifyAll() assert len(fake_reporter.events) == 10 assert et.CHECK_FOR_UPDATES in fake_reporter.events assert et.UPDATE_FOUND in fake_reporter.events assert et.RETRIEVE_DELTA in fake_reporter.events assert et.RETRIEVED_DELTA in fake_reporter.events assert et.PREPARE in fake_reporter.events assert et.COPY_INSTALL in fake_reporter.events assert et.COPIED_INSTALL in fake_reporter.events assert et.UPDATE_COPY in fake_reporter.events assert et.UPDATED_COPY in fake_reporter.events assert et.INVOKING_SWAPPER in fake_reporter.events assert oxygen8_path.isdir() copied_path = URI("memory:///app_base_path/.live_beta_updated.app") copied_internal_path = ( copied_path / "Resources" / "MIDI Remote Scripts" / "humptydumpty" ) assert copied_internal_path.isdir() assert copied_internal_path.islink() # the copied link points still into the original copy. After # swapping everything will be fine. assert copied_internal_path.readlink() == oxygen8_path copied_external_path = ( copied_path / "Resources" / "MIDI Remote Scripts" / "acontroller.py" ) assert copied_external_path.readlink() == external_controller_path # even a broken (looped) symlink will be preserved copied_loop_path = ( copied_path / "Resources" / "MIDI Remote Scripts" / "ixwick.plue" ) assert copied_loop_path.readlink() == loop_path copied_loop_dir_path = copied_path / "Resources" / "MIDI Remote Scripts" / "foo" assert copied_loop_dir_path.readlink() == resources_path # ------------------------------------------------------------------------------------- def test_version_skipping(self): app_base_path = URI("memory:///app_base_path") app_base_path.mkdir() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() config = Bunch( app_base_path=str(app_base_path), autoupdates_dir_base_path=str(autoupdates_dir_base_path), auto_updater_exe="auto_updater_exe", auto_updater_main="auto_updater_main.py", ) tags = [] version = "from" mock_popen_factory = MockPopenFactory() data = "a" * 100000 checksum = md5(data).hexdigest() delta_filename = "delta.asu" cdn_link = "http://cdn.ableton.com/%s/%s" % (checksum, delta_filename) mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.spawn, ) res = connector.check_and_process_auto_updates(tags, version, FakeUsageReporter()) mocker.VerifyAll() self.assertEqual(len(res), 2) # we do *not* execute the download # and prepare commands so that # skipping the version actually makes sense. connector.skip_version("from", "to") update_info = connector.load_update_info() self.assertEqual(update_info["versions"]["from"]["to"]["status"], "skipped") # now invoke the check_and_process_auto_updates # command again - this should result in *no* # action taken! mocker = mox.Mox() mock_conn = mocker.CreateMockAnything() mock_conn.check_for_auto_updates(tags, version, None).AndReturn( {"to": [dict(name=delta_filename, checksum=checksum, version=1)]} ) mocker.ReplayAll() connector = RPCConnector( config, RPCConnector.DEFAULT_ENDPOINT, transport=FakeTransport(mock_conn), urlopen=lambda url: StringIO(data), popen_factory=mock_popen_factory.captured_args, ) res = connector.check_and_process_auto_updates(tags, version, FakeUsageReporter()) mocker.VerifyAll() self.assertEqual(len(res), 0) def test_checksums_are_checked_by_download_tasks(self): base = URI("memory:///base") base.mkdir() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() delta = base / "delta.asu" checksum = "aaaa" config = Bunch( app_base_path=base, autoupdates_dir_base_path=autoupdates_dir_base_path ) class TestConnector(ConnectorBase): def retrieve_delta_into(self, delta_name, dest): create_file(dest, content=delta_name) connector = TestConnector(config) dt = DownloadTask(connector, delta, checksum) self.failUnlessRaises(WrongChecksum, dt.perform, FakeUsageReporter()) assert not dt.finished def test_prepare_auto_update_task_validates_downloads(self): pt = PrepareAutoUpdateTask(None, None, None, [Bunch(finished=False)]) self.failUnlessRaises(DownloadsNotFinished, pt.perform, FakeUsageReporter()) def test_existing_deltas_are_checked_for_proper_checksum(self): app_base_path = URI("memory:///app_base_path") app_base_path.mkdir() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() autoupdates = autoupdates_dir_base_path / "_autoupdates" autoupdates.mkdir() (autoupdates / "deltas").mkdir() delta = autoupdates / "deltas" / "delta.asu" create_file(delta, content="foobar") config = Bunch( app_base_path=app_base_path, autoupdates_dir_base_path=autoupdates_dir_base_path, ) def test(checksum): class TestConnector(ConnectorBase): def check_for_auto_updates( self, tags, current_version, update_to_version ): return {"to": [dict(name="delta.asu", checksum=checksum, version=1)]} def load_update_info(self): return { "delta_infos": {"delta.asu": dict(checksum=checksum, version=1)} } connector = TestConnector(config) res = connector.check_and_process_auto_updates( ["win64"], "from", FakeUsageReporter() ) return res res = test("bbbbb") # we need two tasks, because the checksum differed # and thus a download is created, and the old # delta is removed. self.assertEqual(len(res), 2) assert not delta.exists() # again, this time with the proper checksum create_file(delta, content="foobar") res = test(delta.md5()) # now, only one task should be there, and # the delta still exist! self.assertEqual(len(res), 1) assert delta.exists() def test_residue_is_removed(self): app_base_path = URI("memory:///app_base_path") app_base_path.mkdir() autoupdates_dir_base_path = URI("memory:///autoupdates_dir_base_path") autoupdates_dir_base_path.mkdir() autoupdates = autoupdates_dir_base_path / "_autoupdates" autoupdates.mkdir() extraction_dir = autoupdates / "pending_update" extraction_dir.mkdir() update_commands = extraction_dir / "update_commands" create_file(update_commands, content="foobar") config = Bunch( app_base_path=app_base_path, autoupdates_dir_base_path=autoupdates_dir_base_path, ) class TestConnector(ConnectorBase): def check_for_auto_updates(self, tags, current_version, update_to_version): return {} connector = TestConnector(config) res = connector.check_and_process_auto_updates( ["win64"], "from", FakeUsageReporter() ) assert not extraction_dir.exists() # TODO-dir: to test # test that delta infos are asserted # test that update_infos are pre-populated working # test that declined version isn't processed further # test that partial deltas are discarded