diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7afa4372..25722537 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -61,18 +61,18 @@ jobs: # Run tests and collect coverage data. python -m pytest # Generate LCOV format coverage data for coveralls. - python -m coverage lcov -o coverage.lcov + #coverage lcov -o coverage.lcov - name: Build packages run: | python -m flit build - - name: Send coverage data to coveralls.io - uses: coverallsapp/github-action@main - with: - flag-name: run-${{ join(matrix.*, '-') }} - file: coverage.lcov - parallel: true + #- name: Send coverage data to coveralls.io + # uses: coverallsapp/github-action@main + # with: + # flag-name: run-${{ join(matrix.*, '-') }} + # file: coverage.lcov + # parallel: true finalize: name: finalize diff --git a/pyproject.toml b/pyproject.toml index 7660c190..c267940d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,9 +80,22 @@ src_paths = ["src", "tests", "scripts"] [tool.coverage.run] branch = true # source_pkgs = sdx.pce -omit = [ "tests/*" ] +omit = ["tests/*", "src/*"] relative_files = true +[tool.coverage.report] +exclude_also = [ + "def __repr__", + "if self.debug:", + "self._logger", + "raise", + "if 0:", + "if __name__ == .__main__.:", + "if TYPE_CHECKING:", + "class .*\\bProtocol\\):", + "@(abc\\.)?abstractmethod", + ] + # The section below will let us have relative paths in test coverage # report. See https://hynek.me/articles/testing-packaging/ [tool.coverage.paths] diff --git a/src/sdx_pce/topology/temanager.py b/src/sdx_pce/topology/temanager.py index c4cee8dc..0fd4a4b1 100644 --- a/src/sdx_pce/topology/temanager.py +++ b/src/sdx_pce/topology/temanager.py @@ -20,7 +20,7 @@ ) from sdx_pce.topology.manager import TopologyManager from sdx_pce.utils.constants import Constants -from sdx_pce.utils.exceptions import UnknownRequestError, ValidationError +from sdx_pce.utils.exceptions import TEError, UnknownRequestError, ValidationError UNUSED_VLAN = None @@ -417,8 +417,8 @@ def generate_connection_breakdown( assigned yet. We assign ports in this step. """ if solution is None or solution.connection_map is None: - self._logger.warning(f"Can't find a breakdown for {solution}") - return None + self._logger.warning(f"Can't find a TE solution for {connection_request}") + raise TEError(f"Can't find a TE solution for: {connection_request}", 410) breakdown = {} paths = solution.connection_map # p2p for now @@ -616,7 +616,10 @@ def generate_connection_breakdown( # Make tests pass, temporarily. # ToDo: need to throw an exception if tagged_breakdown is None if tagged_breakdown is None: - return None + raise TEError( + f"Can't find a valid vlan breakdown solution for: {connection_request}", + 409, + ) assert isinstance(tagged_breakdown, VlanTaggedBreakdowns) diff --git a/src/sdx_pce/utils/exceptions.py b/src/sdx_pce/utils/exceptions.py index 4dfde92e..f7bc9758 100644 --- a/src/sdx_pce/utils/exceptions.py +++ b/src/sdx_pce/utils/exceptions.py @@ -19,3 +19,17 @@ def __init__(self, message: str, request_id: str): """ super().__init__(f"{message} (ID: {request_id})") self.request_id = request_id + + +class TEError(Exception): + """ + A custom exception to represent TE (Traffic Engineering) errors. + """ + + def __init__(self, message: str, te_code: int): + """ + :param message: a string containing the error message. + :param te_code: an integer representing the TE error code. + """ + super().__init__(f"{message} (TE Code: {te_code})") + self.te_code = te_code diff --git a/tests/test_te_manager.py b/tests/test_te_manager.py index 20747d4d..9822b0f9 100644 --- a/tests/test_te_manager.py +++ b/tests/test_te_manager.py @@ -7,7 +7,7 @@ from sdx_pce.load_balancing.te_solver import TESolver from sdx_pce.models import ConnectionRequest, ConnectionSolution, TrafficMatrix from sdx_pce.topology.temanager import TEManager -from sdx_pce.utils.exceptions import UnknownRequestError +from sdx_pce.utils.exceptions import TEError, UnknownRequestError from . import TestData @@ -65,7 +65,17 @@ def test_generate_solver_input(self): def test_connection_breakdown_none_input(self): # Expect no breakdown when input is None. - self.assertIsNone(self.temanager.generate_connection_breakdown(None, None)) + """ + Test that generate_connection_breakdown() raises an exception + when given invalid input. + """ + invalid_solution = None + invalid_request = None + + with self.assertRaises(TEError): + self.temanager.generate_connection_breakdown( + invalid_solution, invalid_request + ) def test_connection_breakdown_tm(self): # Breaking down a traffic matrix. @@ -227,8 +237,8 @@ def test_connection_breakdown_some_input(self): self.assertEqual(solution.cost, 0) # If there's no solution, there should be no breakdown either. - breakdown = self.temanager.generate_connection_breakdown(solution, request) - self.assertIsNone(breakdown) + with self.assertRaises(TEError): + self.temanager.generate_connection_breakdown(solution, request) def test_generate_graph_and_connection_with_sax_2_invalid(self): """ @@ -1456,17 +1466,12 @@ def test_disallowed_vlan(self): self.assertIsNotNone(solution) - breakdown = temanager.generate_connection_breakdown( - solution, connection_request - ) - - print(f"breakdown: {json.dumps(breakdown)}") - # Although we have a solution (in terms of connectivity # between ports), we should not have a breakdown at this # point, because we asked for a port that is not present on # the port. - self.assertIsNone(breakdown) + with self.assertRaises(TEError): + self.temanager.generate_connection_breakdown(solution, connection_request) def _vlan_meets_request(self, requested_vlan: str, assigned_vlan: int) -> bool: """