def test_shortname_default():
    r = remote.Remote(
        name='[email protected]',
        ssh=fudge.Fake('SSHConnection'),
        )
    eq(r.shortname, '[email protected]')
    eq(str(r), '[email protected]')

def test_init_templates():
    tmp = maketemp()
    path = os.path.join(tmp, 'repo.git')
    templatedir = os.path.join(
        os.path.dirname(__file__),
        'mocktemplates',
        )
    repository.init(path, template=templatedir)
    repository.init(path)
    got = readFile(os.path.join(path, 'no-confusion'))
    eq(got, 'i should show up\n')
    check_mode(
        os.path.join(path, 'hooks', 'post-update'),
        0755,
        is_file=True,
        )
    got = readFile(os.path.join(path, 'hooks', 'post-update'))
    eq(got, '#!/bin/sh\n# i can override standard templates\n')
    # standard templates are there, too
    assert (
        # compatibility with git <1.6.0
        os.path.isfile(os.path.join(path, 'hooks', 'pre-rebase'))
        # for git >=1.6.0
        or os.path.isfile(os.path.join(path, 'hooks', 'pre-rebase.sample'))
        )

    def test_constructor(self):

        # missing data arg
        with self.assertRaises(TypeError):
            # noinspection PyArgumentList
            AlleleCountsArray()

        # data has wrong dtype
        data = 'foo bar'
        with self.assertRaises(TypeError):
            AlleleCountsArray(data)

        # data has wrong dtype
        data = [4., 5., 3.7]
        with self.assertRaises(TypeError):
            AlleleCountsArray(data)

        # data has wrong dimensions
        data = [1, 2, 3]
        with self.assertRaises(TypeError):
            AlleleCountsArray(data)

        # data has wrong dimensions
        data = diploid_genotype_data
        with self.assertRaises(TypeError):
            AlleleCountsArray(data)

        # valid data (typed)
        ac = AlleleCountsArray(allele_counts_data, dtype='u1')
        aeq(allele_counts_data, ac)
        eq(np.uint8, ac.dtype)

def test_read_yes_map_wouldHaveWritable():
    cfg = RawConfigParser()
    cfg.add_section('group fooers')
    cfg.set('group fooers', 'members', 'jdoe')
    cfg.set('group fooers', 'map writable foo/bar', 'quux/thud')
    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
       None)

def test_has_initial_commit_fail_notAGitDir():
    tmp = maketemp()
    e = assert_raises(
        repository.GitRevParseError,
        repository.has_initial_commit,
        git_dir=tmp)
    eq(str(e), 'rev-parse failed: exit status 128')

 def test_list_objects(self):
     self.ioctx.write('a', b'')
     self.ioctx.write('b', b'foo')
     self.ioctx.write_full('c', b'bar')
     self.ioctx.append('d', b'jazz')
     object_names = [obj.key for obj in self.ioctx.list_objects()]
     eq(sorted(object_names), ['a', 'b', 'c', 'd'])

def test_write_no_simple_wouldHaveReadonly():
    cfg = RawConfigParser()
    cfg.add_section('group fooers')
    cfg.set('group fooers', 'members', 'jdoe')
    cfg.set('group fooers', 'readonly', 'foo/bar')
    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
       None)

def test_getsize():
    store = dict()
    store['foo'] = b'aaa'
    store['bar'] = b'bbbb'
    store['baz/quux'] = b'ccccc'
    eq(7, getsize(store))
    eq(5, getsize(store, 'baz'))

 def test_update(self):
     store = self.create_store()
     assert 'foo' not in store
     assert 'baz' not in store
     store.update(foo=b'bar', baz=b'quux')
     eq(b'bar', store['foo'])
     eq(b'quux', store['baz'])

    def test_init_group_overwrite_path(self):
        # setup
        path = 'foo/bar'
        store = self.create_store()
        meta = dict(shape=(2000,),
                    chunks=(200,),
                    dtype=np.dtype('u1'),
                    compressor=None,
                    fill_value=0,
                    order='F',
                    filters=None)
        store[array_meta_key] = encode_array_metadata(meta)
        store[path + '/' + array_meta_key] = encode_array_metadata(meta)

        # don't overwrite
        with assert_raises(ValueError):
            init_group(store, path=path)

        # do overwrite
        try:
            init_group(store, overwrite=True, path=path)
        except NotImplementedError:
            pass
        else:
            assert array_meta_key not in store
            assert group_meta_key in store
            assert (path + '/' + array_meta_key) not in store
            assert (path + '/' + group_meta_key) in store
            # should have been overwritten
            meta = decode_group_metadata(store[path + '/' + group_meta_key])
            eq(ZARR_FORMAT, meta['zarr_format'])

def setdel_hierarchy_checks(store):
    # these tests are for stores that are aware of hierarchy levels; this
    # behaviour is not stricly required by Zarr but these tests are included
    # to define behaviour of DictStore and DirectoryStore classes

    # check __setitem__ and __delitem__ blocked by leaf

    store['a/b'] = b'aaa'
    with assert_raises(KeyError):
        store['a/b/c'] = b'xxx'
    with assert_raises(KeyError):
        del store['a/b/c']

    store['d'] = b'ddd'
    with assert_raises(KeyError):
        store['d/e/f'] = b'xxx'
    with assert_raises(KeyError):
        del store['d/e/f']

    # test __setitem__ overwrite level
    store['x/y/z'] = b'xxx'
    store['x/y'] = b'yyy'
    eq(b'yyy', store['x/y'])
    assert 'x/y/z' not in store
    store['x'] = b'zzz'
    eq(b'zzz', store['x'])
    assert 'x/y' not in store

    # test __delitem__ overwrite level
    store['r/s/t'] = b'xxx'
    del store['r/s']
    assert 'r/s/t' not in store
    store['r/s'] = b'xxx'
    del store['r']
    assert 'r/s' not in store

    def test_init_group_overwrite(self):
        # setup
        store = self.create_store()
        store[array_meta_key] = encode_array_metadata(
            dict(shape=(2000,),
                 chunks=(200,),
                 dtype=np.dtype('u1'),
                 compressor=None,
                 fill_value=0,
                 order='F',
                 filters=None)
        )

        # don't overwrite array (default)
        with assert_raises(ValueError):
            init_group(store)

        # do overwrite
        try:
            init_group(store, overwrite=True)
        except NotImplementedError:
            pass
        else:
            assert array_meta_key not in store
            assert group_meta_key in store
            meta = decode_group_metadata(store[group_meta_key])
            eq(ZARR_FORMAT, meta['zarr_format'])

        # don't overwrite group
        with assert_raises(ValueError):
            init_group(store)

    def test_get_set_del_contains(self):
        store = self.create_store()

        # test __contains__, __getitem__, __setitem__
        assert 'foo' not in store
        with assert_raises(KeyError):
            # noinspection PyStatementEffect
            store['foo']
        store['foo'] = b'bar'
        assert 'foo' in store
        eq(b'bar', store['foo'])

        # test __delitem__ (optional)
        try:
            del store['foo']
        except NotImplementedError:
            pass
        else:
            assert 'foo' not in store
            with assert_raises(KeyError):
                # noinspection PyStatementEffect
                store['foo']
            with assert_raises(KeyError):
                # noinspection PyStatementEffect
                del store['foo']

 def test_comments(self):
     now = datetime.datetime.now()
     now_ts = int(time.mktime(now.timetuple()))
     before_ts = int(time.mktime((now - datetime.timedelta(minutes=15)).timetuple()))
     message = 'test message ' + str(now_ts)
     comment_id = dog.Comment.create(handle=TEST_USER, message=message)['comment']['id']
     time.sleep(self.wait_time)
     event = dog.Event.get(comment_id)
     eq(event['event']['text'], message)
     dog.Comment.update(comment_id, handle=TEST_USER, message=message + ' updated')
     time.sleep(self.wait_time)
     event = dog.Event.get(comment_id)
     eq(event['event']['text'], message + ' updated')
     reply_id = dog.Comment.create(handle=TEST_USER, message=message + ' reply',
                                   related_event_id=comment_id)['comment']['id']
     time.sleep(3)
     stream = dog.Event.query(start=before_ts, end=now_ts + 100)['events']
     ok(stream is not None, msg="No events found in stream")
     ok(isinstance(stream, list), msg="Event stream is not a list")
     ok(len(stream) > 0, msg="No events found in stream")
     comment_ids = [x['id'] for x in stream[0]['comments']]
     ok(reply_id in comment_ids,
        msg="Should find {0} in {1}".format(reply_id, comment_ids))
     # Delete the reply
     dog.Comment.delete(reply_id)
     # Then the post itself
     dog.Comment.delete(comment_id)
     time.sleep(self.wait_time)
     try:
         dog.Event.get(comment_id)
     except:
         pass
     else:
         assert False

def test_route_error():
    app = natrix.Application([
        ("/500", lambda x: x.response(None.None)),
    ])

    @app.route(":error-404")
    def error_404(x):
        x.response("Custom error 404")

    @app.route(":error-500")
    def error_500(x):
        x.response("Custom error 500")
    # endfold

    testapp = webtest.TestApp(app)

    response = testapp.get("/", status=404)
    eq(response.normal_body, "Custom error 404")

    def _error(*args, **kwargs):
        pass
    # endfold

    natrix_error = natrix.error
    natrix.error = _error
    response = testapp.get("/500", status=500)
    eq(response.normal_body, "Custom error 500")
    natrix.error = natrix_error

    def test_applications(self):
        cmd = {"prefix":"osd dump", "format":"json"}
        ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'')
        eq(ret, 0)
        assert len(buf) > 0
        release = json.loads(buf.decode("utf-8")).get("require_osd_release",
                                                      None)
        if not release or release[0] < 'l':
            raise SkipTest

        eq([], self.ioctx.application_list())

        self.ioctx.application_enable("app1")
        assert_raises(Error, self.ioctx.application_enable, "app2")
        self.ioctx.application_enable("app2", True)

        assert_raises(Error, self.ioctx.application_metadata_list, "dne")
        eq([], self.ioctx.application_metadata_list("app1"))

        assert_raises(Error, self.ioctx.application_metadata_set, "dne", "key",
                      "key")
        self.ioctx.application_metadata_set("app1", "key1", "val1")
        self.ioctx.application_metadata_set("app1", "key2", "val2")
        self.ioctx.application_metadata_set("app2", "key1", "val1")

        eq([("key1", "val1"), ("key2", "val2")],
           self.ioctx.application_metadata_list("app1"))

        self.ioctx.application_metadata_remove("app1", "key1")
        eq([("key2", "val2")], self.ioctx.application_metadata_list("app1"))

 def cb(arg, line, who, sec, nsec, seq, level, msg):
     # NOTE(sileht): the old pyrados API was received the pointer as int
     # instead of the value of arg
     eq(arg, "arg")
     with lock:
         lock.notify()
     return 0

def put_and_compare_file(size, content_func):
    """
    Create file with `size` and content generated by `content_func`.
    Use CLI to PUT and GET that file. Compare afterwards
    """

    obj = random_id()
    in_file = prepare_input_file(size, content_func)
    out_file = prepare_output_file()

    ret = call(["./veintidos.py",
                "--pool", POOL_NAME,
                "put", obj,
                in_file])
    eq(0, ret)

    ret = call(["./veintidos.py",
                "--pool", POOL_NAME,
                "get", obj,
                out_file])
    eq(0, ret)

    eq_file(in_file, out_file)

    os.unlink(in_file)
    os.unlink(out_file)

 def test_write_ops(self):
     with WriteOpCtx(self.ioctx) as write_op:
         write_op.new(0)
         self.ioctx.operate_write_op(write_op, "write_ops")
         eq(self.ioctx.read('write_ops'), b'')
         write_op.write_full(b'1')
         write_op.append(b'2')
         self.ioctx.operate_write_op(write_op, "write_ops")
         eq(self.ioctx.read('write_ops'), b'12')
         write_op.write_full(b'12345')
         write_op.write(b'x', 2)
         self.ioctx.operate_write_op(write_op, "write_ops")
         eq(self.ioctx.read('write_ops'), b'12x45')
         write_op.write_full(b'12345')
         write_op.zero(2, 2)
         self.ioctx.operate_write_op(write_op, "write_ops")
         eq(self.ioctx.read('write_ops'), b'12\x00\x005')
         write_op.write_full(b'12345')
         write_op.truncate(2)
         self.ioctx.operate_write_op(write_op, "write_ops")
         eq(self.ioctx.read('write_ops'), b'12')
         write_op.remove()
         self.ioctx.operate_write_op(write_op, "write_ops")
         with assert_raises(ObjectNotFound):
             self.ioctx.read('write_ops')

    def test_aio_stat(self):
        lock = threading.Condition()
        count = [0]
        def cb(_, size, mtime):
            with lock:
                count[0] += 1
                lock.notify()

        comp = self.ioctx.aio_stat("foo", cb)
        comp.wait_for_complete()
        with lock:
            while count[0] < 1:
                lock.wait()
        eq(comp.get_return_value(), -2)

        self.ioctx.write("foo", b"bar")

        comp = self.ioctx.aio_stat("foo", cb)
        comp.wait_for_complete()
        with lock:
            while count[0] < 2:
                lock.wait()
        eq(comp.get_return_value(), 0)

        [i.remove() for i in self.ioctx.list_objects()]