Loading src/service/service/tools/OpticalTools.py +5 −0 Changes for src/service/service/tools/OpticalTools.py: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -280,17 +280,22 @@ def DelFlexLightpath( src, dst, bitrate, ob_id, flow_id=None) -> str: headers = {"Content-Type": "application/json"} r = requests.delete(urlx, headers=headers) if r.status_code >= 500: r.raise_for_status() reply = r.text code = r.status_code return (reply, code) def delete_lightpath ( src, dst, bitrate, flow_id): reply = "200" code = 200 base_url = get_optical_controller_base_url() if not TESTING: urlx = "{:s}/DelLightpath/{}/{}/{}/{}".format(base_url, src, dst, bitrate, flow_id) headers = {"Content-Type": "application/json"} r = requests.delete(urlx, headers=headers) if r.status_code >= 500: r.raise_for_status() reply = r.text code = r.status_code return (reply, code) Loading src/service/tests/test_optical_spectrum_reservation.py +25 −0 Changes for src/service/tests/test_optical_spectrum_reservation.py: 25 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -20,6 +20,31 @@ from service.service.tools.OpticalSpectrumReservation import parse_optical_spect from service.service.tools import OpticalTools @pytest.mark.parametrize("flex", [False, True]) @pytest.mark.parametrize("status", [200, 400, 404, 500, 503]) def test_optical_delete_propagates_server_failure(monkeypatch, flex, status): response = OpticalTools.requests.Response() response.status_code = status response._content = b"optical reply" response.url = "http://optical/OpticalTFS/delete" monkeypatch.setattr(OpticalTools, "TESTING", False) monkeypatch.setattr(OpticalTools, "get_optical_controller_base_url", lambda: "http://optical/OpticalTFS") monkeypatch.setattr(OpticalTools.requests, "delete", lambda *a, **k: response) def delete(): if flex: return OpticalTools.DelFlexLightpath("A", "B", 800, 7, 1) return OpticalTools.delete_lightpath("A", "B", 800, 1) if status >= 500: with pytest.raises(OpticalTools.requests.HTTPError): delete() else: assert delete() == ("optical reply", status) def test_parse_compact_service_reservation_constraint(): constraints = [ Constraint(custom=Constraint_Custom( Loading Loading
src/service/service/tools/OpticalTools.py +5 −0 Changes for src/service/service/tools/OpticalTools.py: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -280,17 +280,22 @@ def DelFlexLightpath( src, dst, bitrate, ob_id, flow_id=None) -> str: headers = {"Content-Type": "application/json"} r = requests.delete(urlx, headers=headers) if r.status_code >= 500: r.raise_for_status() reply = r.text code = r.status_code return (reply, code) def delete_lightpath ( src, dst, bitrate, flow_id): reply = "200" code = 200 base_url = get_optical_controller_base_url() if not TESTING: urlx = "{:s}/DelLightpath/{}/{}/{}/{}".format(base_url, src, dst, bitrate, flow_id) headers = {"Content-Type": "application/json"} r = requests.delete(urlx, headers=headers) if r.status_code >= 500: r.raise_for_status() reply = r.text code = r.status_code return (reply, code) Loading
src/service/tests/test_optical_spectrum_reservation.py +25 −0 Changes for src/service/tests/test_optical_spectrum_reservation.py: 25 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -20,6 +20,31 @@ from service.service.tools.OpticalSpectrumReservation import parse_optical_spect from service.service.tools import OpticalTools @pytest.mark.parametrize("flex", [False, True]) @pytest.mark.parametrize("status", [200, 400, 404, 500, 503]) def test_optical_delete_propagates_server_failure(monkeypatch, flex, status): response = OpticalTools.requests.Response() response.status_code = status response._content = b"optical reply" response.url = "http://optical/OpticalTFS/delete" monkeypatch.setattr(OpticalTools, "TESTING", False) monkeypatch.setattr(OpticalTools, "get_optical_controller_base_url", lambda: "http://optical/OpticalTFS") monkeypatch.setattr(OpticalTools.requests, "delete", lambda *a, **k: response) def delete(): if flex: return OpticalTools.DelFlexLightpath("A", "B", 800, 7, 1) return OpticalTools.delete_lightpath("A", "B", 800, 1) if status >= 500: with pytest.raises(OpticalTools.requests.HTTPError): delete() else: assert delete() == ("optical reply", status) def test_parse_compact_service_reservation_constraint(): constraints = [ Constraint(custom=Constraint_Custom( Loading