status = 1
# Print the locator (uuid) of the new collection.
- stdout.write(output)
- if not output.endswith('\n'):
- stdout.write('\n')
+ try:
+ stdout.write(output)
+ if not output.endswith('\n'):
+ stdout.write('\n')
+ except NameError:
+ status = status or 1
for sigcode, orig_handler in orig_signal_handlers.items():
signal.signal(sigcode, orig_handler)
import arvados
import arvados.commands.put as arv_put
-from arvados_testutil import ArvadosBaseTestCase
+from arvados_testutil import ArvadosBaseTestCase, fake_httplib2_response
import run_test_server
class ArvadosPutResumeCacheTest(ArvadosBaseTestCase):
self.call_main_with_args,
['--project-uuid', self.Z_UUID, '--stream'])
+ def test_api_error_handling(self):
+ collections_mock = mock.Mock(name='arv.collections()')
+ coll_create_mock = collections_mock().create().execute
+ coll_create_mock.side_effect = arvados.errors.ApiError(
+ fake_httplib2_response(403), '{}')
+ arv_put.api_client = arvados.api('v1')
+ arv_put.api_client.collections = collections_mock
+ with self.assertRaises(SystemExit) as exc_test:
+ self.call_main_with_args(['/dev/null'])
+ self.assertLess(0, exc_test.exception.args[0])
+ self.assertLess(0, coll_create_mock.call_count)
+ self.assertEqual("", self.main_stdout.getvalue())
+
+
class ArvPutIntegrationTest(run_test_server.TestCaseWithServers,
ArvadosBaseTestCase):
def _getKeepServerConfig():