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

Improve tests

Browse source
This commit is contained in:
Daniel A. Wozniak 2024-10-28 17:16:06 -07:00 • committed by Daniel Wozniak
parent 486462b992
commit 5eb5a4035c
1 changed files with 15 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
tests/pytests/unit/transport/test_tcp.py
View file

@ -687,7 +687,7 @@ async def test_pub_server_publish_payload_closed_stream(master_opts, io_loop):
assert server.clients == set()
async def test_pub_server_pull_path_no_perms(master_opts, io_loop):
async def test_pub_server_paths_no_perms(master_opts, io_loop):
def publish_payload(payload):
return payload
@ -698,6 +698,8 @@ async def test_pub_server_pull_path_no_perms(master_opts, io_loop):
pull_host="127.0.0.1",
pull_port=5152,
)
assert pubserv.pull_path is None
assert pubserv.pub_path is None
with patch("os.chmod") as p:
await pubserv.publisher(publish_payload)
assert p.call_count == 0
@ -707,7 +709,7 @@ async def test_pub_server_publisher_pull_path_perms(master_opts, io_loop, tmp_pa
def publish_payload(payload):
return payload
pull_path = tmp_path / "pull.ipc"
pull_path = str(tmp_path / "pull.ipc")
pull_path_perms = 0o664
pubserv = salt.transport.tcp.PublishServer(
master_opts,
@ -715,9 +717,13 @@ async def test_pub_server_publisher_pull_path_perms(master_opts, io_loop, tmp_pa
pub_port=5151,
pull_host=None,
pull_port=None,
pull_path=str(pull_path),
pull_path=pull_path,
pull_path_perms=pull_path_perms,
)
assert pubserv.pull_path == pull_path
assert pubserv.pull_path_perms == pull_path_perms
assert pubserv.pull_host is None
assert pubserv.pull_port is None
with patch("os.chmod") as p:
await pubserv.publisher(publish_payload)
assert p.call_count == 1
@ -728,18 +734,22 @@ async def test_pub_server_publisher_pub_path_perms(master_opts, io_loop, tmp_pat
def publish_payload(payload):
return payload
pub_path = tmp_path / "pub.ipc"
pub_path = str(tmp_path / "pub.ipc")
pub_path_perms = 0o664
pubserv = salt.transport.tcp.PublishServer(
master_opts,
pub_host=None,
pub_port=None,
pub_path=str(pub_path),
pub_path=pub_path,
pub_path_perms=pub_path_perms,
pull_host="127.0.0.1",
pull_port=5151,
pull_path=None,
)
assert pubserv.pub_path == pub_path
assert pubserv.pub_path_perms == pub_path_perms
assert pubserv.pub_host is None
assert pubserv.pub_port is None
with patch("os.chmod") as p:
await pubserv.publisher(publish_payload)
assert p.call_count == 1