saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-16 09:40:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix event tests via suggestions

Browse source
This commit is contained in:
MKLeb 2022-02-22 13:33:54 -05:00 • committed by Megan Wilhite
parent baf388d0fe
commit e682d9de50
4 changed files with 24 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
doc/man/salt.7
View file

@ -61231,7 +61231,7 @@ import salt.utils.event
opts = salt.config.client_config("/etc/salt/master")
event = salt.utils.event.get_event(
"master", sock_dir=opts["sock_dir"], opts=opts
"master", sock_dir=opts["sock_dir"], transport=opts["transport"], opts=opts
)
data = event.get_event()
@ -61305,7 +61305,7 @@ import salt.utils.event
opts = salt.config.client_config("/etc/salt/master")
sevent = salt.utils.event.get_event(
"master", sock_dir=opts["sock_dir"], opts=opts
"master", sock_dir=opts["sock_dir"], transport=opts["transport"], opts=opts
)
while True:

21
tests/pytests/functional/utils/test_async_event_publisher.py
View file

@ -1,20 +1,8 @@
"""
:codeauthor: Pedro Algarvio (pedro@algarvio.me)
tests.unit.utils.event_test
~~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
import os
import shutil
import pytest
import salt.config
import salt.ext.tornado.ioloop
import salt.utils.event
import salt.utils.stringutils
from tests.support.runtests import RUNTIME_VARS
pytestmark = [pytest.mark.windows_whitelisted]
@ -31,12 +19,12 @@ class IOLoopContainer:
@pytest.fixture()
def io_cont(io_loop):
def io_cont(io_loop, tmp_path):
io_cont = IOLoopContainer()
io_cont.io_loop = io_loop
io_cont.sock_dir = os.path.join(RUNTIME_VARS.TMP, "test-socks")
if not os.path.exists(io_cont.sock_dir):
os.makedirs(io_cont.sock_dir)
socks_dir_path = tmp_path / "test-socks"
io_cont.sock_dir = str(socks_dir_path)
socks_dir_path.mkdir(exist_ok=True)
io_cont.opts = {"sock_dir": io_cont.sock_dir}
io_cont.publisher = salt.utils.event.AsyncEventPublisher(
io_cont.opts,
@ -48,7 +36,6 @@ def io_cont(io_loop):
io_cont.event.subscribe("")
io_cont.event.set_event_handler(io_cont._handle_publish)
yield io_cont
shutil.rmtree(io_cont.sock_dir, ignore_errors=True)
def test_event_subscription(io_cont):

55
tests/pytests/unit/utils/event/test_event.py
View file

@ -1,6 +1,4 @@
import hashlib
import os
import shutil
import time
import pytest
@ -11,7 +9,6 @@ import salt.utils.stringutils
import zmq
import zmq.eventloop.ioloop
from tests.support.events import eventpublisher_process, eventsender_process
from tests.support.runtests import RUNTIME_VARS
NO_LONG_IPC = False
if getattr(zmq, "IPC_PATH_MAX_LEN", 103) <= 103:
@ -26,17 +23,11 @@ pytestmark = [
]
@pytest.fixture()
def sock_dir():
return os.path.join(RUNTIME_VARS.TMP, "test-socks")
@pytest.fixture(autouse=True)
def setUp(sock_dir):
if not os.path.exists(sock_dir):
os.makedirs(sock_dir)
yield
shutil.rmtree(sock_dir, ignore_errors=True)
def sock_dir(tmp_path):
sock_dir_path = tmp_path / "test-socks"
sock_dir_path.mkdir(parents=True, exist_ok=True)
yield sock_dir_path
def _assert_got_event(evt, data, msg=None, expected_failure=False):
@ -53,24 +44,18 @@ def _assert_got_event(evt, data, msg=None, expected_failure=False):
def test_master_event(sock_dir):
with salt.utils.event.MasterEvent(sock_dir, listen=False) as me:
assert me.puburi == "{}".format(os.path.join(sock_dir, "master_event_pub.ipc"))
assert me.pulluri == "{}".format(
os.path.join(sock_dir, "master_event_pull.ipc")
)
assert me.puburi == str(sock_dir / "master_event_pub.ipc")
assert me.pulluri == str(sock_dir / "master_event_pull.ipc")
def test_minion_event(sock_dir):
opts = dict(id="foo", sock_dir=sock_dir)
opts = dict(id="foo", sock_dir=str(sock_dir))
id_hash = hashlib.sha256(salt.utils.stringutils.to_bytes(opts["id"])).hexdigest()[
:10
]
with salt.utils.event.MinionEvent(opts, listen=False) as me:
assert me.puburi == "{}".format(
os.path.join(sock_dir, "minion_event_{}_pub.ipc".format(id_hash))
)
assert me.pulluri == "{}".format(
os.path.join(sock_dir, "minion_event_{}_pull.ipc".format(id_hash))
)
assert me.puburi == str(sock_dir / "minion_event_{}_pub.ipc".format(id_hash))
assert me.pulluri == str(sock_dir, "minion_event_{}_pull.ipc".format(id_hash))
def test_minion_event_tcp_ipc_mode():
@ -81,21 +66,17 @@ def test_minion_event_tcp_ipc_mode():
def test_minion_event_no_id(sock_dir):
with salt.utils.event.MinionEvent(dict(sock_dir=sock_dir), listen=False) as me:
with salt.utils.event.MinionEvent(dict(sock_dir=str(sock_dir)), listen=False) as me:
id_hash = hashlib.sha256(salt.utils.stringutils.to_bytes("")).hexdigest()[:10]
assert me.puburi == "{}".format(
os.path.join(sock_dir, "minion_event_{}_pub.ipc".format(id_hash))
)
assert me.pulluri == "{}".format(
os.path.join(sock_dir, "minion_event_{}_pull.ipc".format(id_hash))
)
assert me.puburi == str(sock_dir / "minion_event_{}_pub.ipc".format(id_hash))
assert me.pulluri == str(sock_dir / "minion_event_{}_pull.ipc".format(id_hash))
@pytest.mark.slow_test
def test_event_single(sock_dir):
"""Test a single event is received"""
with eventpublisher_process(sock_dir):
with salt.utils.event.MasterEvent(sock_dir, listen=True) as me:
with eventpublisher_process(str(sock_dir)):
with salt.utils.event.MasterEvent(str(sock_dir), listen=True) as me:
me.fire_event({"data": "foo1"}, "evt1")
evt1 = me.get_event(tag="evt1")
_assert_got_event(evt1, {"data": "foo1"})
@ -104,8 +85,8 @@ def test_event_single(sock_dir):
@pytest.mark.slow_test
def test_event_single_no_block(sock_dir):
"""Test a single event is received, no block"""
with eventpublisher_process(sock_dir):
with salt.utils.event.MasterEvent(sock_dir, listen=True) as me:
with eventpublisher_process(str(sock_dir)):
with salt.utils.event.MasterEvent(str(sock_dir), listen=True) as me:
start = time.time()
finish = start + 5
evt1 = me.get_event(wait=0, tag="evt1", no_block=True)
@ -122,8 +103,8 @@ def test_event_single_no_block(sock_dir):
@pytest.mark.slow_test
def test_event_single_wait_0_no_block_False(sock_dir):
"""Test a single event is received with wait=0 and no_block=False and doesn't spin the while loop"""
with eventpublisher_process(sock_dir):
with salt.utils.event.MasterEvent(sock_dir, listen=True) as me:
with eventpublisher_process(str(sock_dir)):
with salt.utils.event.MasterEvent(str(sock_dir), listen=True) as me:
me.fire_event({"data": "foo1"}, "evt1")
# This is too fast and will be None but assures we're not blocking
evt1 = me.get_event(wait=0, tag="evt1", no_block=False)

8
tests/pytests/unit/utils/event/test_event_return.py
View file

@ -1,11 +1,3 @@
"""
:codeauthor: Pedro Algarvio (pedro@algarvio.me)
tests.unit.utils.event_test
~~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
import pytest
import salt.config
import salt.ext.tornado.ioloop