140 lines
6.3 KiB
Python
140 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2010-2025 Google LLC
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from absl.testing import absltest
|
|
from ortools.util.python import pybind_solve_interrupter
|
|
from ortools.util.python import pybind_solve_interrupter_testing
|
|
|
|
|
|
class PybindPySolveInterrupterTest(absltest.TestCase):
|
|
"""Test pybind_solve_interrupter.PySolveInterrupter class.
|
|
|
|
This tests how pybind_solve_interrupter.PySolveInterrupter is accepted by
|
|
other
|
|
pybind-generated APIs in Python, using pybind_solve_interrupter_testing to do
|
|
so.
|
|
"""
|
|
|
|
def test_none_interrupter(self) -> None:
|
|
"""Test that a `const PySolveInterrupter*` can receive nullptr."""
|
|
# Here IsInterrupted() is a Clif wrapped C++ function that expects a `const
|
|
# PySolveInterrupter*`. It returns None (std::nullopt) for nullptr
|
|
# input.
|
|
self.assertIsNone(pybind_solve_interrupter_testing.IsInterrupted(None))
|
|
|
|
def test_untriggered_interrupter(self) -> None:
|
|
"""Test that an untriggered interrupter is properly passed."""
|
|
interrupter = pybind_solve_interrupter.PySolveInterrupter()
|
|
# Test the getter.
|
|
self.assertFalse(interrupter.interrupted)
|
|
# Test using a Clif wrapped C++ function. We test it is not None to make
|
|
# sure Clif passes a non-null pointer to the C++ code. We thus have to use
|
|
# assertIs() instead of assertFalse().
|
|
self.assertIs(
|
|
pybind_solve_interrupter_testing.IsInterrupted(interrupter), False
|
|
)
|
|
|
|
def test_triggered_interrupter(self) -> None:
|
|
"""Test that an untriggered interrupter is properly passed."""
|
|
interrupter = pybind_solve_interrupter.PySolveInterrupter()
|
|
interrupter.interrupt()
|
|
# Test the getter.
|
|
self.assertTrue(interrupter.interrupted)
|
|
# Test using a Clif wrapped C++ function.
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(interrupter))
|
|
|
|
def test_shared_reference(self) -> None:
|
|
"""Test that taking a std::shared_ptr<PySolveInterrupter> works."""
|
|
# Create an interrupter and a PySolveInterrupterReference class which
|
|
# constructor takes and std::shared_ptr<PySolveInterrupter>. We expect
|
|
# that only one instance of the C++ PySolveInterrupter will exist here.
|
|
interrupter = pybind_solve_interrupter.PySolveInterrupter()
|
|
interrupter_ref = pybind_solve_interrupter_testing.PySolveInterrupterReference(
|
|
interrupter
|
|
)
|
|
|
|
# Validate that we have the expected number of references of the shared_ptr
|
|
# hold by PySolveInterrupterReference.
|
|
self.assertEqual(interrupter_ref.use_count, 2)
|
|
self.assertFalse(interrupter_ref.is_interrupted)
|
|
|
|
# Triggering the interrupter should be visible the pointed
|
|
# PySolveInterrupter in C++.
|
|
interrupter.interrupt()
|
|
self.assertEqual(interrupter_ref.use_count, 2)
|
|
self.assertTrue(interrupter_ref.is_interrupted)
|
|
|
|
# Removing the Python `interrupter` reference should make `interrupter_ref`
|
|
# the only object that holds an std::shared_ptr on the interrupter.
|
|
del interrupter
|
|
self.assertEqual(interrupter_ref.use_count, 1)
|
|
self.assertTrue(interrupter_ref.is_interrupted)
|
|
|
|
def test_add_target(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.add_trigger_target(target)
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(target))
|
|
|
|
def test_remove_existing_target(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.add_trigger_target(target)
|
|
source.remove_trigger_target(target)
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
self.assertFalse(pybind_solve_interrupter_testing.IsInterrupted(target))
|
|
|
|
def test_remove_existing_target_added_twice(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.add_trigger_target(target)
|
|
source.add_trigger_target(target)
|
|
source.remove_trigger_target(target)
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
self.assertFalse(pybind_solve_interrupter_testing.IsInterrupted(target))
|
|
|
|
def test_dead_target(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.add_trigger_target(target)
|
|
del target
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
|
|
def test_remove_existing_target_twice(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.add_trigger_target(target)
|
|
source.remove_trigger_target(target)
|
|
source.remove_trigger_target(target)
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
self.assertFalse(pybind_solve_interrupter_testing.IsInterrupted(target))
|
|
|
|
def test_remove_non_target(self) -> None:
|
|
source = pybind_solve_interrupter.PySolveInterrupter()
|
|
target = pybind_solve_interrupter.PySolveInterrupter()
|
|
source.remove_trigger_target(target)
|
|
source.interrupt()
|
|
self.assertTrue(pybind_solve_interrupter_testing.IsInterrupted(source))
|
|
self.assertFalse(pybind_solve_interrupter_testing.IsInterrupted(target))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
absltest.main()
|