Skip to content

Commit

Permalink
Use caplog in test_mmsf_validator
Browse files Browse the repository at this point in the history
  • Loading branch information
maarten-ic committed Aug 27, 2024
1 parent 73b9f35 commit 9386709
Showing 1 changed file with 75 additions and 46 deletions.
121 changes: 75 additions & 46 deletions libmuscle/python/libmuscle/test/test_mmsf_validator.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from logging import WARNING
from typing import Any
from unittest.mock import Mock

Expand All @@ -8,18 +9,16 @@
from libmuscle.mmsf_validator import MMSFValidator


# For testing purposes we monkeypatch _logger.warning so it raises the following
# exception: ot is easier to verify that an exception is raised than checking that a
# warning message is logged.
class TestMMSFValidatorException(Exception):
pass
class Contains:
"""Helper class to simplify tests using caplog.record_tuples"""
def __init__(self, value: Any) -> None:
self.value = value

def __eq__(self, other: Any) -> bool:
return self.value in other

@pytest.fixture(autouse=True)
def patch_logger_to_raise_error(monkeypatch):
def raise_on_log(msg: str, *args: Any) -> None:
raise TestMMSFValidatorException(msg % args)
monkeypatch.setattr("libmuscle.mmsf_validator._logger.warning", raise_on_log)
def __repr__(self) -> str:
return f"Contains({self.value!r})"


@pytest.fixture
Expand All @@ -45,7 +44,7 @@ def validator_simple(mock_peer_info) -> MMSFValidator:

@pytest.mark.parametrize("num_iterations", [0, 1, 2])
@pytest.mark.parametrize("num_reuse", [1, 5])
def test_simple_correct(num_iterations, num_reuse, validator_simple):
def test_simple_correct(num_iterations, num_reuse, validator_simple, caplog):
for _ in range(num_reuse):
validator_simple.reuse_instance()
validator_simple.check_receive("f_i", None)
Expand All @@ -55,77 +54,100 @@ def test_simple_correct(num_iterations, num_reuse, validator_simple):
validator_simple.check_send("o_f", None)
# Final reuse_instance()
validator_simple.reuse_instance()
assert caplog.record_tuples == []


def test_simple_skip_f_init(validator_simple):
def test_simple_skip_f_init(validator_simple, caplog):
validator_simple.reuse_instance()
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_send("o_i", None)
validator_simple.check_send("o_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Send on port 'o_i'"))]


def test_simple_skip_o_i(validator_simple):
def test_simple_skip_o_i(validator_simple, caplog):
validator_simple.reuse_instance()
validator_simple.check_receive("f_i", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_receive("f_i", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_receive("s", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.reuse_instance()

validator_simple.check_receive("f_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Receive on port 'f_i'"))]

caplog.clear()
validator_simple.check_receive("s", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Receive on port 's'"))]

caplog.clear()
validator_simple.reuse_instance()
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("reuse_instance()"))]


def test_simple_skip_s(validator_simple):
def test_simple_skip_s(validator_simple, caplog):
validator_simple.reuse_instance()
validator_simple.check_receive("f_i", None)
validator_simple.check_send("o_i", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_send("o_i", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_send("o_f", None)

validator_simple.check_send("o_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Send on port 'o_i'"))]

caplog.clear()
validator_simple.check_send("o_f", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Send on port 'o_f'"))]

def test_simple_skip_o_f(validator_simple):

def test_simple_skip_o_f(validator_simple, caplog):
validator_simple.reuse_instance()
validator_simple.check_receive("f_i", None)
validator_simple.check_send("o_i", None)
validator_simple.check_receive("s", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.reuse_instance()

validator_simple.reuse_instance()
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("reuse_instance()"))]


def test_simple_skip_reuse_instance(validator_simple):
def test_simple_skip_reuse_instance(validator_simple, caplog):
validator_simple.reuse_instance()
validator_simple.check_receive("f_i", None)
validator_simple.check_receive("o_f", None)
with pytest.raises(TestMMSFValidatorException):
validator_simple.check_receive("f_i", None)

validator_simple.check_receive("f_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Receive on port 'f_i'"))]


def test_only_o_f(mock_peer_info):
def test_only_o_f(mock_peer_info, caplog):
port_manager = PortManager([], {Operator.O_F: ["o_f"]})
port_manager.connect_ports(mock_peer_info)
validator = MMSFValidator(port_manager)

for _ in range(5):
validator.reuse_instance()
validator.check_send("o_f", None)
with pytest.raises(TestMMSFValidatorException):
validator.check_send("o_f", None)

validator.check_send("o_f", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Send on port 'o_f'"))]

def test_only_f_i(mock_peer_info):

def test_only_f_i(mock_peer_info, caplog):
port_manager = PortManager([], {Operator.F_INIT: ["f_i"]})
port_manager.connect_ports(mock_peer_info)
validator = MMSFValidator(port_manager)

for _ in range(5):
validator.reuse_instance()
validator.check_receive("f_i", None)
with pytest.raises(TestMMSFValidatorException):
validator.check_receive("f_i", None)

validator.check_receive("f_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Receive on port 'f_i'"))]


def test_micro(mock_peer_info):
def test_micro(mock_peer_info, caplog):
port_manager = PortManager([], {Operator.F_INIT: ["f_i"], Operator.O_F: ["o_f"]})
port_manager.connect_ports(mock_peer_info)
validator = MMSFValidator(port_manager)
Expand All @@ -136,19 +158,26 @@ def test_micro(mock_peer_info):
validator.check_receive("o_f", None)
validator.reuse_instance()
validator.check_receive("f_i", None)
with pytest.raises(TestMMSFValidatorException):
validator.reuse_instance()
with pytest.raises(TestMMSFValidatorException):
validator.check_receive("f_i", None)

validator.check_receive("f_i", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Receive on port 'f_i'"))]

caplog.clear()
validator.reuse_instance()
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("reuse_instance()"))]


def test_not_all_ports_used(mock_peer_info):
def test_not_all_ports_used(mock_peer_info, caplog):
port_manager = PortManager([], {
Operator.F_INIT: ["f_i1", "f_i2"], Operator.O_F: ["o_f"]})
port_manager.connect_ports(mock_peer_info)
validator = MMSFValidator(port_manager)

validator.reuse_instance()
validator.check_receive("f_i1", None)
with pytest.raises(TestMMSFValidatorException):
validator.check_send("o_f", None)

validator.check_send("o_f", None)
assert caplog.record_tuples == [
("libmuscle.mmsf_validator", WARNING, Contains("Send on port 'o_f'"))]

0 comments on commit 9386709

Please sign in to comment.