fc6e58d9e6
Add SPDX header to python files with the 'py' extension in the test directory. Signed-off-by: Albert Esteve <aesteve@redhat.com>
157 lines
4.4 KiB
Python
157 lines
4.4 KiB
Python
# SPDX-FileCopyrightText: Red Hat, Inc.
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
|
|
import logging
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from ovirt_imageio import admin
|
|
from ovirt_imageio._internal import config
|
|
from ovirt_imageio._internal import server
|
|
|
|
from . import testutil
|
|
|
|
log = logging.getLogger("test")
|
|
|
|
|
|
@pytest.fixture(scope="module", params=["daemon", "proxy"])
|
|
def srv(request):
|
|
path = "test/conf/{}.conf".format(request.param)
|
|
cfg = config.load(path)
|
|
s = server.Server(cfg)
|
|
s.start()
|
|
yield s
|
|
s.stop()
|
|
|
|
|
|
@pytest.mark.parametrize("trans", ["unix", "tcp"])
|
|
def test_client_error(trans):
|
|
ticket = testutil.create_ticket(
|
|
ops=["read"], sparse=False, dirty=False, transfer_id="123")
|
|
|
|
class cfg:
|
|
class control:
|
|
transport = trans
|
|
socket = "/no/such/sock"
|
|
port = 64 * 1024
|
|
|
|
log.debug("Accessing stopped server")
|
|
with pytest.raises(admin.ClientError) as e:
|
|
with admin.Client(cfg, timeout=0.5) as c:
|
|
c.add_ticket(ticket)
|
|
log.debug("Error: %s", e.value)
|
|
|
|
|
|
def test_get_ticket_missing(srv):
|
|
with admin.Client(srv.config) as c:
|
|
ticket_id = str(uuid.uuid4())
|
|
|
|
with pytest.raises(admin.ServerError) as e:
|
|
log.debug("Getting ticket %s", ticket_id)
|
|
c.get_ticket(ticket_id)
|
|
|
|
log.debug("Error: %s", e.value)
|
|
assert e.value.code == 404
|
|
assert ticket_id in e.value.message
|
|
|
|
|
|
def test_add_ticket(srv):
|
|
ticket = testutil.create_ticket(
|
|
ops=["read"], sparse=False, dirty=False, transfer_id="123")
|
|
|
|
with admin.Client(srv.config) as c:
|
|
log.debug("Adding ticket %s", ticket)
|
|
c.add_ticket(ticket)
|
|
info = c.get_ticket(ticket["uuid"])
|
|
log.debug("Got ticket %s", info)
|
|
assert info["uuid"] == ticket["uuid"]
|
|
|
|
|
|
def test_get_ticket(srv, fake_time):
|
|
ticket = testutil.create_ticket(
|
|
ops=["read"], sparse=False, dirty=False, transfer_id="123")
|
|
|
|
with admin.Client(srv.config) as c:
|
|
log.debug("Adding ticket %s", ticket)
|
|
c.add_ticket(ticket)
|
|
info = c.get_ticket(ticket["uuid"])
|
|
log.debug("Got ticket %s", info)
|
|
|
|
assert info == {
|
|
# Ticket attributes
|
|
"ops": ticket["ops"],
|
|
"size": ticket["size"],
|
|
"sparse": ticket["sparse"],
|
|
"dirty": ticket["dirty"],
|
|
"inactivity_timeout": ticket["inactivity_timeout"],
|
|
"timeout": ticket["timeout"],
|
|
"url": ticket["url"],
|
|
"uuid": ticket["uuid"],
|
|
"transfer_id": ticket["transfer_id"],
|
|
# Ticket server status.
|
|
"active": False,
|
|
"canceled": False,
|
|
"connections": 0,
|
|
"expires": 300,
|
|
"idle_time": 0,
|
|
"transferred": 0
|
|
}
|
|
|
|
|
|
def test_mod_ticket(srv, fake_time):
|
|
ticket = testutil.create_ticket(
|
|
ops=["read"], sparse=False, dirty=False, transfer_id="123")
|
|
|
|
with admin.Client(srv.config) as c:
|
|
log.debug("Adding ticket %s", ticket)
|
|
c.add_ticket(ticket)
|
|
info = c.get_ticket(ticket["uuid"])
|
|
assert info["expires"] == 300
|
|
|
|
# Extend ticket.
|
|
c.mod_ticket(ticket["uuid"], {"timeout": 600})
|
|
info = c.get_ticket(ticket["uuid"])
|
|
assert info["expires"] == 600
|
|
|
|
# Expire ticket.
|
|
c.mod_ticket(ticket["uuid"], {"timeout": 0})
|
|
info = c.get_ticket(ticket["uuid"])
|
|
assert info["expires"] == 0
|
|
|
|
# Try to change read only value
|
|
with pytest.raises(admin.ServerError):
|
|
c.mod_ticket(ticket["uuid"], {"url": "file:///other"})
|
|
|
|
# Value must not change.
|
|
info = c.get_ticket(ticket["uuid"])
|
|
assert info["url"] == ticket["url"]
|
|
|
|
|
|
def test_del_ticket(srv):
|
|
ticket = testutil.create_ticket(
|
|
ops=["read"], sparse=False, dirty=False, transfer_id="123")
|
|
|
|
with admin.Client(srv.config) as c:
|
|
log.debug("Adding ticket %s", ticket)
|
|
c.add_ticket(ticket)
|
|
|
|
# Make sure ticket exists before removing it.
|
|
c.get_ticket(ticket["uuid"])
|
|
|
|
log.debug("Removing ticket %s", ticket["uuid"])
|
|
c.del_ticket(ticket["uuid"])
|
|
|
|
with pytest.raises(admin.ServerError) as e:
|
|
c.get_ticket(ticket["uuid"])
|
|
|
|
log.debug("Error: %s", e.value)
|
|
assert e.value.code == 404
|
|
|
|
|
|
def test_del_ticket_missing(srv):
|
|
with admin.Client(srv.config) as c:
|
|
ticket_id = str(uuid.uuid4())
|
|
# Shoudld succeed.
|
|
c.del_ticket(ticket_id)
|