@@ -0,0 +1,3193 @@
+import threading%0Afrom subprocess import Popen%0A%0Aimport mock%0Aimport pytest%0Afrom watchdog.events import FileSystemEvent, DirModifiedEvent%0A%0Afrom chalice.cli import reloader%0Afrom chalice.local import LocalDevServer%0A%0A%0A# NOTE: Most of the reloader module relies on threads, subprocesses,%0A# and process exiting with specific return codes.  This is quite difficult%0A# to unit test, so the more realistic tests are over in function/test_local.py.%0Aclass RecordingPopen(object):%0A    def __init__(self, process, return_codes=None):%0A        self.process = process%0A        self.recorded_args = %5B%5D%0A        if return_codes is None:%0A            return_codes = %5B%5D%0A        self.return_codes = return_codes%0A%0A    def __call__(self, *args, **kwargs):%0A        self.recorded_args.append((args, kwargs))%0A        if self.return_codes:%0A            rc = self.return_codes.pop(0)%0A            self.process.returncode = rc%0A        return self.process%0A%0A%0Adef test_restarter_triggers_event():%0A    restart_event = threading.Event()%0A    restarter = reloader.Restarter(restart_event)%0A    app_modified = FileSystemEvent(src_path='./app.py')%0A    restarter.on_any_event(app_modified)%0A    assert restart_event.is_set()%0A%0A%0Adef test_directory_events_ignored():%0A    restart_event = threading.Event()%0A    restarter = reloader.Restarter(restart_event)%0A    app_modified = DirModifiedEvent(src_path='./')%0A    restarter.on_any_event(app_modified)%0A    assert not restart_event.is_set()%0A%0A%0Adef test_http_server_thread_starts_server_and_shutsdown():%0A    server = mock.Mock(spec=LocalDevServer)%0A    thread = reloader.HTTPServerThread(lambda: server)%0A    thread.run()%0A    thread.shutdown()%0A    server.serve_forever.assert_called_with()%0A    server.shutdown.assert_called_with()%0A%0A%0Adef test_shutdown_noop_if_server_not_started():%0A    server = mock.Mock(spec=LocalDevServer)%0A    thread = reloader.HTTPServerThread(lambda: server)%0A    thread.shutdown()%0A    assert not server.shutdown.called%0A%0A%0Adef test_parent_process_starts_child_with_worker_env_var():%0A    process = mock.Mock(spec=Popen)%0A    process.returncode = 0%0A    popen = RecordingPopen(process)%0A    env = %7B'original-env': 'foo'%7D%0A    parent = reloader.ParentProcess(env, popen)%0A%0A    parent.main()%0A%0A    assert len(popen.recorded_args) == 1%0A    kwargs = popen.recorded_args%5B-1%5D%5B1%5D%0A    assert kwargs == %7B'env': %7B'original-env': 'foo',%0A                              'CHALICE_WORKER': 'true'%7D%7D%0A%0A%0Adef test_assert_child_restarted_until_not_restart_rc():%0A    process = mock.Mock(spec=Popen)%0A    popen = RecordingPopen(%0A        process, return_codes=%5Breloader.RESTART_REQUEST_RC, 0%5D)%0A    parent = reloader.ParentProcess(%7B%7D, popen)%0A%0A    parent.main()%0A%0A    # The child process should have been invoked twice, the first one%0A    # was with RESTART_REQUEST_RC so that should trigger a restart,%0A    # then second one was rc 0 the process should just exit.%0A    assert len(popen.recorded_args) == 2%0A%0A%0Adef test_ctrl_c_kill_child_process():%0A    process = mock.Mock(spec=Popen)%0A    process.communicate.side_effect = KeyboardInterrupt%0A    popen = RecordingPopen(process)%0A    parent = reloader.ParentProcess(%7B%7D, popen)%0A%0A    with pytest.raises(KeyboardInterrupt):%0A        parent.main()%0A%0A    assert process.terminate.called%0A

Add back tests for reloader