def file_like_objects(self):
     fd = BytesIO()
     fd.write(b"yup\n")
     result = self.c.put(local=fd, remote=self.remote)
     eq_(open(self.remote).read(), "yup\n")
     eq_(result.remote, self.remote)
     ok_(result.local is fd)

 def handles_invalid_kwargs_like_any_other_function(self):
     try:
         self._run(_, nope_noway_nohow='as if')
     except TypeError as e:
         ok_('got an unexpected keyword argument' in str(e))
     else:
         assert False, "Invalid run() kwarg didn't raise TypeError"

 def exited_is_None(self):
     try:
         self._watcher_error()
     except Failure as e:
         exited = e.result.exited
         err = "Expected None, got {0!r}".format(exited)
         ok_(exited is None, err)

 def stringrep_notes_exit_status(self):
     try:
         self._regular_error()
     except Failure as e:
         ok_("exited with status 1" in str(e.result))
     else:
         assert False, "Did not raise Failure!"

 def exited_is_integer(self):
     try:
         self._regular_error()
     except Failure as e:
         ok_(isinstance(e.result.exited, int))
     else:
         assert False, "Did not raise Failure!"

 def preserves_basic_members(self):
     c1 = Config(
         defaults={'key': 'default'},
         overrides={'key': 'override'},
         system_prefix='global',
         user_prefix='user',
         project_home='project',
         runtime_path='runtime.yaml',
     )
     c2 = c1.clone()
     # NOTE: expecting identical defaults also implicitly tests that
     # clone() passes in defaults= instead of doing an empty init +
     # copy. (When that is not the case, we end up with
     # global_defaults() being rerun and re-added to _defaults...)
     eq_(c2._defaults, c1._defaults)
     ok_(c2._defaults is not c1._defaults)
     eq_(c2._overrides, c1._overrides)
     ok_(c2._overrides is not c1._overrides)
     eq_(c2._system_prefix, c1._system_prefix)
     eq_(c2._user_prefix, c1._user_prefix)
     eq_(c2._project_home, c1._project_home)
     eq_(c2.prefix, c1.prefix)
     eq_(c2.file_prefix, c1.file_prefix)
     eq_(c2.env_prefix, c1.env_prefix)
     eq_(c2._runtime_path, c1._runtime_path)

 def modifications_on_clone_do_not_alter_original(self):
     # Setup
     orig = Call(
         self.task,
         called_as='foo',
         args=[1, 2, 3],
         kwargs={'key': 'val'}
     )
     context = Context()
     context['setting'] = 'value'
     orig.context = context
     # Clone & tweak
     clone = orig.clone()
     newtask = Task(Mock(__name__='meh'))
     clone.task = newtask
     clone.called_as = 'notfoo'
     clone.args[0] = 7
     clone.kwargs['key'] = 'notval'
     clone.context['setting'] = 'notvalue'
     # Compare
     ok_(clone.task is not orig.task)
     eq_(orig.called_as, 'foo')
     eq_(clone.called_as, 'notfoo')
     eq_(orig.args, [1, 2, 3])
     eq_(clone.args, [7, 2, 3])
     eq_(orig.kwargs['key'], 'val')
     eq_(clone.kwargs['key'], 'notval')
     eq_(orig.context['setting'], 'value')
     eq_(clone.context['setting'], 'notvalue')

 def can_be_pickled(self):
     c = Context()
     c.foo = {'bar': {'biz': ['baz', 'buzz']}}
     c2 = pickle.loads(pickle.dumps(c))
     eq_(c, c2)
     ok_(c is not c2)
     ok_(c.foo.bar.biz is not c2.foo.bar.biz)

 def can_clone_into_a_subclass(self):
     orig = Call(self.task)
     class MyCall(Call):
         pass
     clone = orig.clone(into=MyCall)
     eq_(clone, orig)
     ok_(isinstance(clone, MyCall))

 def base_case(self):
     queue = Queue()
     t = EHThread(target=self.worker, args=[queue])
     t.start()
     t.join()
     eq_(queue.get(block=False), 7)
     ok_(queue.empty())

 def prefixes_command_with_sudo(self, Local):
     runner = Local.return_value
     Context().sudo('whoami')
     # NOTE: implicitly tests default sudo.prompt conf value
     cmd = "sudo -S -p '[sudo] password: ' whoami"
     ok_(runner.run.called, "sudo() never called runner.run()!")
     eq_(runner.run.call_args[0][0], cmd)

 def comparison_looks_at_merged_config(self):
     c1 = Config(defaults={'foo': {'bar': 'biz'}})
     # Empty defaults to suppress global_defaults
     c2 = Config(defaults={}, overrides={'foo': {'bar': 'biz'}})
     ok_(c1 is not c2)
     ok_(c1._defaults != c2._defaults)
     eq_(c1, c2)

 def is_exception_when_WatcherError_raised_internally(self):
     try:
         self._watcher_error()
     except Failure as e:
         ok_(isinstance(e.reason, WatcherError))
     else:
         assert False, "Failed to raise Failure!"

 def optional_prevents_bool_defaults_from_affecting_kind(self):
     # Re #416. See notes in the function under test for rationale.
     @task(optional=['myarg'])
     def mytask(c, myarg=False):
         pass
     arg = mytask.get_arguments()[0]
     ok_(arg.kind is str) # not bool!

 def dict_value_merges_are_not_references(self):
     core = {}
     coll = {'foo': {'bar': {'biz': 'coll value'}}}
     proj = {'foo': {'bar': {'biz': 'proj value'}}}
     # Initial merge - when bug present, this sets core['foo'] to the entire
     # 'foo' dict in 'proj' as a reference - meaning it 'links' back to the
     # 'proj' dict whenever other things are merged into it
     merge_dicts(core, proj)
     eq_(core, {'foo': {'bar': {'biz': 'proj value'}}})
     eq_(proj['foo']['bar']['biz'], 'proj value')
     # Identity tests can also prove the bug early
     ok_(core['foo'] is not proj['foo'], "Core foo is literally proj foo!")
     # Subsequent merge - just overwrites leaf values this time (thus no
     # real change, but this is what real config merge code does, so why
     # not)
     merge_dicts(core, proj)
     eq_(core, {'foo': {'bar': {'biz': 'proj value'}}})
     eq_(proj['foo']['bar']['biz'], 'proj value')
     # The problem merge - when bug present, core['foo'] references 'foo'
     # inside 'proj', so this ends up tweaking "core" but it actually
     # affects "proj" as well!
     merge_dicts(core, coll)
     # Expect that the core dict got the update from 'coll'...
     eq_(core, {'foo': {'bar': {'biz': 'coll value'}}})
     # BUT that 'proj' remains UNTOUCHED
     eq_(proj['foo']['bar']['biz'], 'proj value')

 def returns_run_result(self, Local):
     runner = Local.return_value
     expected = runner.run.return_value
     result = Context().sudo('whoami')
     ok_(
         result is expected,
         "sudo() did not return run()'s return value!",
     )

 def hide_unknown_vals_mention_value_given_in_error(self):
     value = "penguinmints"
     try:
         run("command", hide=value)
     except ValueError, e:
         msg = "Error from run(hide=xxx) did not tell user what the bad value was!"
         msg += "\nException msg: %s" % e
         ok_(value in str(e), msg)

 def returns_deep_copy_of_given_dict(self):
     # NOTE: not actual deepcopy...
     source = {'foo': {'bar': {'biz': 'baz'}}}
     copy = copy_dict(source)
     eq_(copy['foo']['bar'], source['foo']['bar'])
     ok_(copy['foo']['bar'] is not source['foo']['bar'])
     copy['foo']['bar']['biz'] = 'notbaz'
     eq_(source['foo']['bar']['biz'], 'baz')

 def kwargs(self):
     k = {'foo': 'bar'}
     self.executor.execute(('task1', k))
     args = self.task1.body.call_args[0]
     kwargs = self.task1.body.call_args[1]
     ok_(isinstance(args[0], Context))
     eq_(len(args), 1)
     eq_(kwargs['foo'], 'bar')

 def task_has_no_help_shows_per_task_help(self):
     task1 = Context('mytask')
     init = Context(args=[Argument('help', optional=True)])
     parser = Parser(initial=init, contexts=[task1])
     result = parser.parse_argv(['mytask', '--help'])
     eq_(len(result), 2)
     eq_(result[0].args.help.value, 'mytask')
     ok_('help' not in result[1].args)