Repo test for PR 2448

repo_test.py

from uuid import uuid4 as UUID
import tempfile
import pytest
import time
import sys
import os

libdir = os.path.abspath("dist/lib/python")
sys.path.insert(0, libdir)
assert "ICE_CONFIG" not in os.environ


import omero.cli

#
# omero.fs.repo.path=%user%_%userId%//%year%-%month%/%day%/%time%
#
# %user%       bob
# %userId%     4
# %group%      bobLab
# %groupId%    3
# %year%       2011
# %month%      01
# %monthname%  january
# %day%        01
# %time%       15-13-54.014
# /            path separator


tmp = None

def go(*args):
    if 0 != omero.cli.argv(["bin/omero"] + list(args)):
        raise Exception("bin/omero %s failed!" % " ".join(args))


uuid = str(UUID())
go("login", "root@localhost", "-w", "ome")
go("group", "add", uuid)
go("user", "add", "-P", "ome", uuid, "first", "last", uuid)


def execute(repopath, expected):
    global tmp
    go("config", "set", "omero.fs.repo.path", repopath)
    go("config", "get", "omero.fs.repo.path", repopath)
    go("admin", "restart")
    count = 3
    while count > 0:
        count -= 1
        time.sleep(1)
        try:
            go("login", uuid + "@localhost", "-w", "ome")
            count == 0  # Succeeded
        except:
            if count == 0:
                raise
    go("import", tmp)#"---file=/dev/null", "---errs=/dev/null", tmp)


def setup_module():
    global tmp
    tmp = tempfile.mktemp(suffix=".fake")
    open(tmp, "w").close()
    print "FILE:", tmp


@pytest.mark.parametrize(
    "repopath,expect_success", [
        ("//", False),
        ("//%time%", True),
        ("//uid%userId%/%time%", True),
        ("uid%userId%//%time%", True),
        ("drive-1//uid%userId%/%time%", True),
        ("drive-1/uid%userId%//%time%", True),
        ("drive-1//uid%userId%/%time%", True),
        ("gid%groupId%/uid%userId%//%time%", True),
        ("gid%groupId%//uid%userId%/%time%", True),
    ])
def test_repo(capsys, repopath, expect_success):
    try:
        execute(repopath, expect_success)
        if not expect_success:
            raise Exception("Should have failed!")
    except:
        out, err = capsys.readouterr()
        rv = ""
        rv += "=== out ===\n"
        rv += out
        rv += "=== err ===\n"
        rv += err
        if expect_success:
            raise Exception(rv)

def test_simple_giduid():
    execute("gid%groupId%/uid%userId%//%time%", True)


@pytest.mark.templatechanges
def test_swap_user_root():
    execute("gid%groupId%/uid%userId%//%time%", True)
    execute("gid%groupId%//uid%userId%/%time%", True)
    execute("gid%groupId%/uid%userId%//%time%", True)
    execute("gid%groupId%//uid%userId%/%time%", True)


@pytest.mark.templatechanges
def test_change_to_root_with_mount():
    uuid = str(UUID())
    execute("%s//a/%%time%%" % uuid, True)
    execute("%s/a//%%time%%" % uuid, True)


@pytest.mark.templatechanges
def test_change_to_user_with_mount():
    uuid = str(UUID())
    execute("%s/a//%%time%%" % uuid, True)
    execute("%s//a/%%time%%" % uuid, True)


def teardown_module():
    global tmp
    try:
        go("config", "set", "omero.fs.repo.path")
        go("admin", "restartasync")
    finally:
        os.remove(tmp)