Something went wrong. Try again.
Flat White Vitriol - An incremental solver protocol for combinatorial solving using shared objects
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454"""End-to-end tests for the FZnSO Python bindings.
A model is built in Python and run against the example solver. The examplesolver assigns each decision its domain's lower bound, shifts the first by thesolution index, and warns that it ignores the constraints."""
import ctypes.utilimport pathlibimport shutilimport sys
import pytest
import fznso
REPO = pathlib.Path(__file__).resolve().parents[2]
# The type every decision in these fixtures has.VAR_INT = fznso.Type("int", decision=True)
def solver_path() -> str: prefix, suffix = ("lib", ".dylib") if sys.platform == "darwin" else ("lib", ".so") path = REPO / "target" / "debug" / "examples" / f"{prefix}example_solver{suffix}" if not path.exists(): pytest.skip(f"{path} not built; run `cargo build --example example_solver`") return str(path)
def threaded_solver_path() -> str: prefix, suffix = ("lib", ".dylib") if sys.platform == "darwin" else ("lib", ".so") path = REPO / "target" / "debug" / "examples" / f"{prefix}threaded_solver{suffix}" if not path.exists(): pytest.skip(f"{path} not built; run `cargo build --example threaded_solver`") return str(path)
@pytest.fixturedef solver(): return fznso.Library(solver_path()).create_solver()
@pytest.fixturedef threaded_solver(): """A solver that reads the model and reports from a worker thread.""" return fznso.Library(threaded_solver_path()).create_solver()
def build_model(): m = fznso.LayeredModel() a = m.add_decision(VAR_INT, range(3, 21), name="a") # inclusive [3, 20] b = m.add_decision(VAR_INT, fznso.int_set(10, 20), name="b") m.add_constraint("int_le", [0, a]) m.set_objective("minimize", a) return m, a, b
def test_options_round_trip(solver): assert solver.option_get("solution_limit") == 1 solver.option_set("solution_limit", 3) assert solver.option_get("solution_limit") == 3
def test_solver_statistic(solver): # `runs` is declared solver-scoped, so it is read from the solver instance. assert solver.statistic("runs") == 0 # Unknown, and solution-scoped, statistics are absent here. assert solver.statistic("nope") is None assert solver.statistic("solutions") is None
def test_unknown_option_raises_with_message(solver): with pytest.raises(ValueError, match="unknown option `nope`"): solver.option_set("nope", 1)
def test_run_reports_solutions_and_messages(solver): solver.option_set("solution_limit", 3) m, a, b = build_model()
solutions, messages = [], [] status = solver.run( m, on_solution=lambda s: solutions.append((s[a], s[b], s.statistic("solutions"))), on_message=lambda scope, value: messages.append((scope, value)), )
assert status == fznso.Status.Complete assert solutions == [(3, 10, 0), (4, 10, 1), (5, 10, 2)] assert messages == [("warn", "ignoring 1 constraint(s)")]
def test_solution_is_invalid_outside_callback(solver): m, a, _ = build_model() escaped = [] solver.run(m, on_solution=lambda s: escaped.append(s)) # The solution borrowed a value that no longer exists; using it must raise # rather than read freed memory. with pytest.raises(RuntimeError, match="outside the on_solution callback"): escaped[0][a]
def test_callback_exception_propagates(solver): m, _, _ = build_model()
def boom(_sol): raise KeyError("from the callback")
with pytest.raises(KeyError, match="from the callback"): solver.run(m, on_solution=boom)
def test_value_kinds_round_trip(): # Reading every value kind back through the solver's model view is covered # elsewhere; here we check the Python value conversions directly via the # domain a solution reports. m = fznso.LayeredModel() x = m.add_decision(VAR_INT, fznso.int_set(-5, 5), name="x") assert isinstance(x, fznso.Decision) assert int(x) == 0
class DomainModel(fznso.Model): """A model implemented in Python rather than built — the solver reads it on demand by calling these methods.
The values it returns are borrowed, not copied, so it owns them: every object it hands back is built once in ``__init__`` and stored on the instance."""
def __init__(self, domains, constraints=0): super().__init__() self._domains = [fznso.int_set(lo, hi) for lo, hi in domains] self._names = [f"x{i}" for i in range(len(domains))] self._ident = "int_le" self._arguments = [[0, fznso.Decision(i)] for i in range(constraints)]
def decision_count(self): return len(self._domains)
def decision_type(self, i): return VAR_INT
def decision_domain(self, i): return self._domains[i]
def decision_name(self, i): return self._names[i]
def constraint_count(self): return len(self._arguments)
def constraint_ident(self, i): return self._ident
def constraint_arguments(self, i): return self._arguments[i]
def test_custom_python_model(solver): solver.option_set("solution_limit", 2) model = DomainModel([(7, 20), (30, 40)], constraints=2)
solutions, messages = [], [] solver.run( model, on_solution=lambda s: solutions.append((s.value(0), s.value(1))), on_message=lambda scope, value: messages.append((scope, value)), ) # Lower bounds 7 and 30; the solver shifts the first by the solution index. assert solutions == [(7, 30), (8, 30)] assert messages == [("warn", "ignoring 2 constraint(s)")]
def test_run_can_be_stopped_early(solver): """`should_stop` ends a run the caller has already started.""" solver.option_set("solution_limit", 5) m, a, _ = build_model()
# "One solution is enough", decided from inside the callback. solutions = [] status = solver.run( m, on_solution=lambda s: solutions.append(s[a]), should_stop=lambda: len(solutions) >= 1, ) assert len(solutions) == 1, "the solver kept going after being asked to stop" assert status == fznso.Status.Incomplete
# Already asked to stop before the run begins: no search at all. solutions = [] status = solver.run( m, on_solution=lambda s: solutions.append(s[a]), should_stop=lambda: True ) assert solutions == [] assert status == fznso.Status.Incomplete
def test_stop_predicate_error_propagates(solver): """An exception from should_stop surfaces once the run unwinds.""" m, _, _ = build_model()
def boom(): raise KeyError("from should_stop")
with pytest.raises(KeyError, match="from should_stop"): solver.run(m, should_stop=boom)
def test_watchdog_thread_can_stop_a_run(solver): """Another Python thread can cancel a long run.
This works only because ``run`` releases the GIL: the watchdog needs it to execute at all while the solver is searching. """ import threading
solver.option_set("solution_limit", 100_000) m, a, _ = build_model()
seen = [] stop = threading.Event()
def watchdog(): while len(seen) < 3: pass stop.set()
thread = threading.Thread(target=watchdog) thread.start() status = solver.run( m, on_solution=lambda s: seen.append(s[a]), should_stop=stop.is_set ) thread.join()
assert status == fznso.Status.Incomplete assert 3 <= len(seen) < 100_000, f"expected an early stop, saw {len(seen)}"
def test_threaded_solver_does_not_deadlock(threaded_solver): """A solver reporting from its own thread must not hang.
``run`` releases the GIL for the duration of the search, so a worker thread can take it to invoke the callbacks. Were it held across the run, this would *hang* rather than fail — which is exactly why the case needs a test. """ m, a, _ = build_model() # a native LayeredModel: two decisions
solutions, messages = [], [] status = threaded_solver.run( m, on_solution=lambda s: solutions.append(s[a]), on_message=lambda scope, value: messages.append((scope, value)), )
assert status == fznso.Status.Complete # The fixture reports the decision count it read from the worker thread. assert solutions == [2] assert messages == [("log", "read 2 decision(s) from a worker thread")]
def test_threaded_solver_reads_a_python_model(threaded_solver): """The strongest case: the worker thread re-enters Python for *model reads*.
With a model implemented in Python, every query the worker makes has to reacquire the GIL from a thread that never held it, on top of the callbacks. """ model = DomainModel([(7, 20), (30, 40), (1, 2)])
solutions = [] status = threaded_solver.run(model, on_solution=lambda s: solutions.append(s.value(0)))
assert status == fznso.Status.Complete # Three decisions, counted by the worker through the Python model. assert solutions == [3]
class TwoLayerModel(fznso.Model): """A custom model split across two layers, to check the layer protocol."""
def __init__(self, domains, split): super().__init__() self._domains = [fznso.int_set(lo, hi) for lo, hi in domains] self._names = [f"x{i}" for i in range(len(domains))] self._split = split # number of decisions in the first layer self._ident = "int_le" self._arguments = [7, fznso.Decision(0)] self._objective = "minimize" self._objective_arg = fznso.Decision(0)
def decision_count(self): return len(self._domains)
def decision_type(self, i): return VAR_INT
def decision_domain(self, i): return self._domains[i]
def decision_name(self, i): return self._names[i]
def layer_count(self): return 2
def decision_layer_end(self, layer): return self._split if layer == 0 else len(self._domains)
def constraint_count(self): return 1
def constraint_ident(self, i): return self._ident
def constraint_arguments(self, i): return self._arguments
def objective_ident(self): return self._objective
def objective_argument(self): return self._objective_arg
def test_custom_layered_model(solver): model = TwoLayerModel([(7, 20), (30, 40), (100, 110)], split=2)
solutions, messages = [], [] solver.run( model, on_solution=lambda s: solutions.append((s.value(0), s.value(1), s.value(2))), on_message=lambda scope, value: messages.append((scope, value)), ) assert solutions == [(7, 30, 100)] # The solver reports the per-layer decision counts: 2 in layer 0, 1 in layer 1. assert ("layers", "2,1") in messages # It also reads back the model's strings (objective, decision name, constraint # identifier) and a constraint argument, all borrowing native Python objects. assert ("model", "obj=minimize;name0=x0;con0=int_le(7)") in messages
def test_types(): # A decision's type is what it is; its domain is what it may take. Only the # type tells `var bool` from an unbounded `var int`, since both have no # domain to report. assert VAR_INT.base == "int" assert VAR_INT.decision assert not VAR_INT.set_of assert fznso.Type("int") != VAR_INT, "a parameter is not a decision variable" with pytest.raises(ValueError, match="unknown base type"): fznso.Type("complex") with pytest.raises(TypeError): fznso.Type("int", True) # the qualifiers are keyword-only
# Every shape the interface can carry is expressible, including the ones no # decision variable uses, and every qualifier reads back under the name it # was given — so a type can be taken apart and rebuilt. every = fznso.Type("string", list_of=True, decision=True, set_of=True, opt=True) assert (every.base, every.list_of, every.decision, every.set_of, every.opt) == ( "string", True, True, True, True, ) assert ( fznso.Type( every.base, list_of=every.list_of, decision=every.decision, set_of=every.set_of, opt=every.opt, ) == every )
# `str` spells the type the way the protocol does, in every binding. assert str(VAR_INT) == "var int" assert str(fznso.Type("bool", decision=True)) == "var bool" assert str(fznso.Type("int", decision=True, set_of=True)) == "var set of int" assert str(fznso.Type("float", decision=True, opt=True)) == "var opt float" assert str(fznso.Type("int", list_of=True)) == "list of int" assert str(every) == "list of var opt set of string"
# `repr` stays the expression that rebuilds it. assert repr(fznso.Type("int")) == 'Type("int")' assert repr(VAR_INT) == 'Type("int", decision=True)' assert ( repr(every) == 'Type("string", list_of=True, decision=True, set_of=True, opt=True)' ) assert eval(repr(every), {"Type": fznso.Type}) == every
m = fznso.LayeredModel() b = m.add_decision(fznso.Type("bool", decision=True), name="b") assert int(b) == 0
def test_decision_type_has_no_default(solver): """A model that declares decisions has to say what they are.
Every other accessor has a sensible default — no domain, no name, not defined — but a type cannot be guessed from anything else the model reports, so the base class refuses rather than inventing one. """
class NoType(fznso.Model): def decision_count(self): return 1
with pytest.raises(NotImplementedError, match="decision_type"): NoType().decision_type(0)
def test_custom_model_method_error_propagates(solver): class Broken(fznso.Model): def decision_count(self): return 1
def decision_domain(self, i): raise ValueError("bad domain")
with pytest.raises(ValueError, match="bad domain"): solver.run(Broken(), on_solution=lambda s: None)
def test_find_loads_a_solver_from_the_search_path(tmp_path, monkeypatch): # Stage the built example in a directory on the search path, under the file # name it will be found by: `libexample_solver.*` names `example_solver`. built = pathlib.Path(solver_path()) shutil.copy(built, tmp_path / built.name) monkeypatch.setenv("FZNSO_SOLVER_PATH", str(tmp_path))
# The staged directory takes precedence over the user and system ones. assert fznso.Library.search_paths()[0] == str(tmp_path)
# It is listed without being loaded, carrying no version in its file name. found = [d for d in fznso.Library.discover() if d.name == "example_solver"] assert found, f"discover() did not list it: {fznso.Library.discover()}" assert found[0].version == "" assert found[0].path == str(tmp_path / built.name)
# Loading by name works just like loading by path. solver = fznso.Library.find("example_solver").create_solver() assert solver.option_get("solution_limit") == 1
# The staged example carries no version, so a specific request misses. with pytest.raises(RuntimeError, match="no solver named"): fznso.Library.find("example_solver", "9")
# An unknown name reports where it looked. with pytest.raises(RuntimeError, match=str(tmp_path)): fznso.Library.find("definitely_absent")