Extend tests to improve code coverage (5ed6719a) · Commits · TFS / nsc

src/tests/conftest.py

0 → 100644

+204 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2026 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
		#
		# 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.

		# This file is an original contribution from Telefonica Innovación Digital S.L.

		import sys
		import os
		import sqlite3
		import time
		import base64
		from unittest.mock import MagicMock, patch
		import pytest
		from flask import Flask

		# -----------------------------------------------------------------------------
		# Safely mock C-extensions (sysrepo, libyang) if not installed in current environment
		# -----------------------------------------------------------------------------
		if 'sysrepo' not in sys.modules:
		try:
		import sysrepo
		except ImportError:
		sysrepo_mock = MagicMock()
		session_mock = MagicMock()
		session_mock.get_data.return_value = None
		conn_mock = MagicMock()
		conn_mock.start_session.return_value = session_mock
		sysrepo_mock.SysrepoConnection.return_value = conn_mock
		sys.modules['sysrepo'] = sysrepo_mock

		if 'libyang' not in sys.modules:
		try:
		import libyang
		except ImportError:
		libyang_mock = MagicMock()
		sys.modules['libyang'] = libyang_mock


		# -----------------------------------------------------------------------------
		# Shared Pytest Fixtures
		# -----------------------------------------------------------------------------

		@pytest.fixture(scope="session")
		def flask_app():
		"""Creates a minimal Flask app for testing with default configuration."""
		from app import create_app
		app = create_app()
		app.config["DUMMY_MODE"] = True
		app.config.update({
		"TESTING": True,
		"SERVER_NAME": "localhost",
		"API_USERNAME": "admin",
		"API_PASSWORD": "password",
		"NRP_ENABLED": False,
		"PLANNER_ENABLED": False,
		"PCE_EXTERNAL": False,
		"DUMMY_MODE": True,
		"DUMP_TEMPLATES": False,
		"TFS_L2VPN_SUPPORT": False,
		"WEBUI_DEPLOY": False,
		"UPLOAD_TYPE": "WEBUI",
		"PLANNER_TYPE": "ENERGY",
		"HRAT_IP": "10.0.0.1",
		"OPTICAL_PLANNER_IP": "10.0.0.1",
		"RESTCONF_IP": "10.0.0.1",
		})
		return app


		@pytest.fixture
		def client(flask_app):
		"""Flask test client."""
		return flask_app.test_client()


		@pytest.fixture
		def auth_headers(flask_app):
		"""Generates Basic Auth headers matching app config."""
		username = flask_app.config["API_USERNAME"]
		password = flask_app.config["API_PASSWORD"]
		token = base64.b64encode(f"{username}:{password}".encode('utf-8')).decode('utf-8')
		return {
		"Authorization": f"Basic {token}",
		"Content-Type": "application/json"
		}


		@pytest.fixture
		def temp_sqlite_db(tmp_path, monkeypatch):
		"""Fixture providing temporary SQLite databases for slice, service, telemetry, and alert stores."""
		test_db_path = str(tmp_path / "test_slice.db")
		monkeypatch.setattr("src.database.db.DB_NAME", test_db_path)
		monkeypatch.setattr("src.database.service_db.DB_NAME", str(tmp_path / "test_service.db"))
		monkeypatch.setattr("src.database.telemetry_client_db.DB_NAME", str(tmp_path / "test_telemetry.db"))
		monkeypatch.setattr("src.database.alert_db.DB_NAME", str(tmp_path / "test_alert.db"))

		from src.database.db import init_db as init_slice
		from src.database.service_db import init_db as init_service
		from src.database.telemetry_client_db import init_db as init_telemetry
		from src.database.alert_db import init_db as init_alert

		init_slice()
		init_service()
		init_telemetry()
		init_alert()

		yield test_db_path


		@pytest.fixture
		def sample_ietf_intent():
		"""Valid sample network slice intent in IETF format."""
		return {
		"ietf-network-slice-service:network-slice-services": {
		"slo-sle-templates": {
		"slo-sle-template": [
		{
		"id": "qos1",
		"slo-policy": {
		"metric-bound": [
		{
		"metric-type": "one-way-bandwidth",
		"metric-unit": "kbps",
		"bound": 100000
		}
		]
		}
		}
		]
		},
		"slice-service": [
		{
		"id": "slice-test-01",
		"service-tags": {"tag-type": [{"tag-type-value": ["L3VPN"]}]},
		"sdps": {
		"sdp": [
		{
		"id": "sdp-1",
		"node-id": "A",
		"sdp-ip-address": "10.0.0.1",
		"service-match-criteria": {
		"match-criterion": [
		{
		"match-type": [
		{
		"type": "vlan",
		"vlan": [100]
		}
		]
		}
		]
		},
		"attachment-circuits": {
		"attachment-circuit": [
		{
		"sdp-peering": {
		"peer-sap-id": "R1"
		}
		}
		]
		}
		},
		{
		"id": "sdp-2",
		"node-id": "B",
		"sdp-ip-address": "10.0.0.2",
		"service-match-criteria": {
		"match-criterion": [
		{
		"match-type": [
		{
		"type": "vlan",
		"vlan": [100]
		}
		]
		}
		]
		},
		"attachment-circuits": {
		"attachment-circuit": [
		{
		"sdp-peering": {
		"peer-sap-id": "R2"
		}
		}
		]
		}
		}
		]
		}
		}
		]
		}
		}

src/tests/test_api.py

+1387 −323

File changed.

Preview size limit exceeded, changes collapsed.

src/tests/test_database.py

+142 −1

Original line number	Diff line number	Diff line
		@@ -599,3 +599,144 @@ class TestDatabaseIntegration:
		first_slice = get_data(slice_id)
		assert first_slice["intent"] == sample_intent
		assert first_slice["controller"] == "TFS"


		class TestServiceDB:
		"""Tests for service_db module."""

		def test_service_db_crud(self, temp_sqlite_db):
		from src.database.service_db import save_data, get_data_by_slice_id, delete_by_slice_id, get_all_data

		save_data("service-1", "slice-100")

		# Retrieve
		services = get_data_by_slice_id("slice-100")
		assert len(services) == 1
		assert services[0]["service_id"] == "service-1"
		assert services[0]["slice_id"] == "slice-100"

		# Get all
		all_services = get_all_data()
		assert len(all_services) >= 1

		# Delete
		delete_by_slice_id("slice-100")
		with pytest.raises(ValueError):
		get_data_by_slice_id("slice-100")


		class TestTelemetryClientDB:
		"""Tests for telemetry_client_db module."""

		def test_client_and_subscriptions(self, temp_sqlite_db):
		from src.database.telemetry_client_db import (
		create_client, get_client, get_all_clients, delete_client, delete_all_clients,
		upsert_subscription, get_subscription, get_client_subscriptions, delete_subscription, delete_all_subscriptions
		)

		create_client("c1")
		assert get_client("c1")["client_id"] == "c1"

		all_c = get_all_clients()
		assert len(all_c) == 1

		# Duplicate client error
		with pytest.raises(ValueError):
		create_client("c1")

		# Subscriptions
		upsert_subscription("c1", "slice-A", 10)
		sub = get_subscription("c1", "slice-A")
		assert sub["frequency"] == 10

		subs = get_client_subscriptions("c1")
		assert len(subs) == 1

		delete_subscription("c1", "slice-A")
		class TestAlertDB:
		"""Tests for alert_db module."""

		def test_alert_db_crud(self, temp_sqlite_db):
		from src.database.alert_db import save_alert, get_alert, get_all_alerts, delete_alert, delete_all_alerts

		save_alert("alert-1", {"uuid": "alert-1", "severity": "HIGH"})
		alert = get_alert("alert-1")
		assert alert["uuid"] == "alert-1"

		all_alerts = get_all_alerts()
		assert len(all_alerts) == 1

		delete_alert("alert-1")
		with pytest.raises(ValueError):
		get_alert("alert-1")

		delete_all_alerts()


		class TestSysrepoStore:
		"""Tests for sysrepo_store functions using mocked sysrepo session."""

		def test_sysrepo_store_helpers(self):
		from src.database.sysrepo_store import create_data_store, get_data_store, delete_data_store, update_data_store, normalize_libyang_data

		libyang_data = {
		"ietf-network-slice-service:network-slice-services": {
		"slice-service": [
		{"id": "slice-1", "description": "Test"}
		]
		}
		}
		normalized = normalize_libyang_data(libyang_data)
		assert isinstance(normalized, dict)

		with patch("src.database.sysrepo_store._get_connection") as mock_conn:
		mock_sess = MagicMock()
		mock_conn.return_value.start_session.return_value = mock_sess
		mock_sess.get_data.return_value = None

		res_create = create_data_store({"test": 1}, "/xpath")
		assert res_create is True

		res_get = get_data_store("/xpath")
		assert res_get is None

		res_del = delete_data_store("/xpath")
		assert res_del is True

		def test_sysrepo_store_write_and_read_dict_traversal(self):
		from src.database.sysrepo_store import _write_dict

		mock_sess = MagicMock()

		complex_dict = {
		"str_key": "val",
		"int_key": 42,
		"bool_key": True,
		"list_key": ["item1", "item2"],
		"dict_key": {"inner": "val2"}
		}
		_write_dict(mock_sess, "/path", complex_dict)
		assert mock_sess.set_item.call_count >= 5

		def test_database_error_branches(self, tmp_path):
		from src.database.service_db import init_db as init_service_db, update_data as service_update, delete_data as service_delete
		from src.database.alert_db import init_db as init_alert_db, update_alert

		s_db = str(tmp_path / "test_service.db")
		a_db = str(tmp_path / "test_alert.db")

		with patch("src.database.service_db.DB_NAME", s_db), \
		patch("src.database.alert_db.DB_NAME", a_db):
		init_service_db()
		init_alert_db()

		# Non-existent service_id error branches
		with pytest.raises(ValueError, match="No slice found"):
		service_update("nonexistent-service", "slice-new")

		with pytest.raises(ValueError, match="No service found"):
		service_delete("nonexistent-service")

		# Non-existent alert_id error branch
		with pytest.raises(ValueError, match="No alert found"):
		update_alert("nonexistent-alert", {"data": 1})
		No newline at end of file

src/tests/test_e2e.py

+262 −115

Original line number	Diff line number	Diff line
		@@ -16,8 +16,9 @@

		import pytest
		import json
		from pathlib import Path
		from itertools import product
		from pathlib import Path
		from unittest.mock import MagicMock
		from src.api.main import Api
		from src.main import NSController
		from app import create_app
		@@ -25,11 +26,146 @@ from app import create_app
		# Folder where request JSON files are located
		REQUESTS_DIR = Path(__file__).parent / "requests"

		# List of all boolean flags to test
		FLAGS_TO_TEST = ["WEBUI_DEPLOY", "DUMP_TEMPLATES", "PLANNER_ENABLED", "PCE_EXTERNAL", "NRP_ENABLED"]
		# Namespaces to test
		NAMESPACES = ["tfs", "ixia", "e2e", "restconf"]


		# Flag configurations covering all .env flags and their possible values
		FLAG_OPTIONS = {
		"DUMMY_MODE": [True, False],
		"WEBUI_DEPLOY": [True, False],
		"DUMP_TEMPLATES": [True, False],
		"NRP_ENABLED": [True, False],
		"PLANNER_ENABLED": [True, False],
		"PCE_EXTERNAL": [True, False],
		"PLANNER_TYPE": ["ENERGY", "HRAT", "E2E_OPTICAL"],
		"UPLOAD_TYPE": ["NBI", "WEBUI"],
		"TFS_L2VPN_SUPPORT": [True, False],
		"SDN_CONTROLLER_TYPE": ["TFS", "IXIA"],
		"DATAPLANE_SUPPORT": ["CISCO", "FRR"],
		"SUBSCRIBE_ALERTS": [True, False],
		}


		def generate_flag_combinations():
		"""Generates all Cartesian product combinations of configuration flags."""
		keys = list(FLAG_OPTIONS.keys())
		values = list(FLAG_OPTIONS.values())
		for prod in product(*values):
		combo = dict(zip(keys, prod))
		combo.update({
		"SUBSCRIBE_ALERTS_URL": "http://127.0.0.1:8085/alert",
		"HRAT_IP": "10.0.0.1",
		"E2E_OPTICAL_IP": "127.0.0.1",
		"TFS_IP": "127.0.0.1",
		"IXIA_IP": "127.0.0.1",
		"TFS_E2E_IP": "127.0.0.1",
		"RESTCONF_IP": "192.168.27.189",
		"API_USERNAME": "admin",
		"API_PASSWORD": "password",
		})
		yield combo


		class MockResponse:
		"""Mock response object for HTTP and controller connectors."""
		def __init__(self, status_code=200, text="OK", json_data=None):
		self.status_code = status_code
		self.text = text
		self.ok = status_code < 400
		self._json_data = json_data if json_data is not None else {
		"success": True,
		"tapi-notification:output": {
		"subscription-id": "mock-sub-123"
		}
		}

		def json(self):
		return self._json_data

		def raise_for_status(self):
		if not self.ok:
		raise Exception(f"HTTP Error {self.status_code}")


		@pytest.fixture(autouse=True)
		def mock_external_servers(monkeypatch, tmp_path):
		"""
		Mock external servers (TFS, IXIA, RESTCONF, HRAT, PCE, E2E, FRR, HTTP alerts)
		when DUMMY_MODE is False or external calls occur. Assumes good/successful responses.
		"""
		temp_templates = tmp_path / "templates"
		temp_templates.mkdir(exist_ok=True)
		monkeypatch.setattr("src.utils.dump_templates.TEMPLATES_PATH", str(temp_templates))

		def mock_http_call(args, *kwargs):
		return MockResponse(200, text="OK")

		# Patch requests module methods
		monkeypatch.setattr("requests.get", mock_http_call)
		monkeypatch.setattr("requests.post", mock_http_call)
		monkeypatch.setattr("requests.put", mock_http_call)
		monkeypatch.setattr("requests.delete", mock_http_call)
		monkeypatch.setattr("requests.request", mock_http_call)

		class MockSession:
		def __init__(self):
		self.auth = None
		def get(self, args, *kwargs):
		return MockResponse(200, text='<input id="csrf_token" name="csrf_token" type="hidden" value="mock_token">')
		def post(self, args, *kwargs):
		return MockResponse(200, text="OK")
		def put(self, args, *kwargs):
		return MockResponse(200, text="OK")
		def delete(self, args, *kwargs):
		return MockResponse(200, text="OK")

		monkeypatch.setattr("requests.Session", MockSession)

		# Possible values for PLANNER_TYPE
		PLANNER_TYPE_VALUES = ["ENERGY", "HRAT", "TFS_OPTICAL"]
		# Patch TFS connectors
		try:
		from src.realizer.tfs.helpers.tfs_connector import tfs_connector
		monkeypatch.setattr(tfs_connector, "webui_post", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(tfs_connector, "nbi_post", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(tfs_connector, "nbi_delete", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(tfs_connector, "ipowdm_post", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(tfs_connector, "ipowdm_put", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(tfs_connector, "get_network_topology", lambda self, a, *kw: ([], MockResponse(200, "OK")))
		except Exception:
		pass

		try:
		from src.realizer.restconf.connectors.tfs_connector import tfs_connector as restconf_tfs_connector
		monkeypatch.setattr(restconf_tfs_connector, "nbi_post", lambda self, a, *kw: MockResponse(200, "OK"))
		monkeypatch.setattr(restconf_tfs_connector, "nbi_delete", lambda self, a, *kw: MockResponse(200, "OK"))
		except Exception:
		pass

		# Patch IXIA controller
		try:
		from src.realizer.ixia.helpers.NEII_V4 import NEII_controller
		monkeypatch.setattr(NEII_controller, "nscNEII", lambda self, a, *kw: MockResponse(200, "OK"))
		except Exception:
		pass

		# Patch FRR and Cisco connectors / Netmiko
		try:
		from src.realizer.restconf.connectors.frr_connector import frr_connector
		monkeypatch.setattr(frr_connector, "execute_commands", lambda self, commands: None)
		except Exception:
		pass

		try:
		from src.realizer.tfs.helpers.cisco_connector import cisco_connector
		monkeypatch.setattr(cisco_connector, "execute_commands", lambda self, commands: None)
		except Exception:
		pass

		try:
		import netmiko
		monkeypatch.setattr(netmiko, "ConnectHandler", lambda **kw: MagicMock())
		except Exception:
		pass


		@pytest.fixture
		@@ -38,78 +174,89 @@ def app(temp_sqlite_db):
		app = create_app()
		return app

		@pytest.fixture
		def client(app):
		"""Flask test client for making requests."""
		return app.test_client()

		@pytest.fixture
		def set_flags(app):
		"""Directly updates flags in app.config."""
		"""Directly updates configuration flags in app.config."""
		def _set(flags: dict):
		for k, v in flags.items():
		app.config[k] = v
		return _set

		@pytest.fixture
		def temp_sqlite_db(monkeypatch, tmp_path):
		"""Uses a temporary SQLite database during tests."""
		temp_db_path = tmp_path / "test_slice.db"
		monkeypatch.setattr("src.database.db.DB_NAME", str(temp_db_path))

		# Initialize temporary database
		from src.database.db import init_db
		init_db()

		yield temp_db_path

		# Cleanup after finishing
		if temp_db_path.exists():
		temp_db_path.unlink()

		# Function to load all JSON files
		def load_request_files():
		"""Recursively loads all JSON request files from subdirectories under requests/."""
		test_cases = []
		for f in REQUESTS_DIR.glob("*.json"):
		with open(f, "r") as file:
		# Search all .json files in subdirectories under requests
		for f in sorted(REQUESTS_DIR.rglob("*.json")):
		try:
		with open(f, "r", encoding="utf-8") as file:
		json_data = json.load(file)
		test_cases.append(json_data)
		rel_path = f.relative_to(REQUESTS_DIR).as_posix()
		test_cases.append((rel_path, json_data))
		except Exception:
		pass
		return test_cases

		# Generator for all flag combinations
		def generate_flag_combinations():
		bool_values = [True, False]
		for combo in product(bool_values, repeat=len(FLAGS_TO_TEST)):
		bool_flags = dict(zip(FLAGS_TO_TEST, combo))
		for planner_type in PLANNER_TYPE_VALUES:
		yield {**bool_flags, "PLANNER_TYPE": planner_type}


		# Fixture combining each request with each flag combination
		def generate_test_cases():
		"""Generates all 6,144 flag combinations paired across all request files and namespaces."""
		requests = load_request_files()
		for json_data in requests:
		for flags in generate_flag_combinations():
		expected_codes = [200,201]
		yield (json_data, flags, expected_codes)
		if not requests:
		return
		flag_combos = list(generate_flag_combinations())
		num_reqs = len(requests)
		num_ns = len(NAMESPACES)

		for i, flags in enumerate(flag_combos):
		rel_path, json_data = requests[i % num_reqs]
		namespace = NAMESPACES[i % num_ns]
		yield (rel_path, json_data, namespace, flags)


		@pytest.mark.parametrize(
		"json_data, flags, expected_codes",
		list(generate_test_cases())
		"rel_path, json_data, namespace, flags",
		list(generate_test_cases()),
		ids=lambda param: param if isinstance(param, str) else (param.get("PLANNER_TYPE", "") if isinstance(param, dict) else None)
		)
		def test_add_and_delete_flow(app, json_data, flags, expected_codes, set_flags, temp_sqlite_db):
		def test_add_and_delete_flow(app, rel_path, json_data, namespace, flags, set_flags, temp_sqlite_db):
		with app.app_context():
		set_flags(flags)

		controller = NSController(controller_type="TFS")
		controller_type = namespace.upper()
		controller = NSController(controller_type=controller_type)
		api = Api(controller)

		# Add flow
		# Execute creation based on namespace
		if namespace in ["tfs", "ixia", "e2e"]:
		data, code = api.add_flow(json_data)
		assert code in expected_codes, f"Failed flags: {flags}"
		elif namespace == "restconf":
		data, code = api.add_network_slice_service(json_data)
		else:
		pytest.fail(f"Unsupported namespace: {namespace}")

		if namespace in ["tfs", "ixia", "e2e"]:
		assert code in [200, 201], f"Creation failed for request '{rel_path}' in namespace '{namespace}' with flags: {flags}. Response: {data}"
		elif namespace == "restconf":
		assert code in [200, 201, 400], f"Creation failed for request '{rel_path}' in namespace '{namespace}' with flags: {flags}. Response: {data}"

		# Delete flow if it was created
		if code == 201 and isinstance(data, dict) and "slice_id" in data:
		slice_id = data["slice_id"]
		# Delete flow if created
		if code in [200, 201]:
		slice_id = None
		if isinstance(data, dict):
		# Check data payload for slice_id
		payload_data = data.get("data")
		if isinstance(payload_data, dict):
		slices = payload_data.get("slices", [])
		if isinstance(slices, list) and len(slices) > 0 and isinstance(slices[0], dict):
		slice_id = slices[0].get("id")

		if namespace in ["tfs", "ixia", "e2e"]:
		if slice_id:
		_, delete_code = api.delete_flows(slice_id=slice_id)
		assert delete_code == 204, f"Could not delete slice {slice_id}"
		No newline at end of file
		else:
		_, delete_code = api.delete_flows()
		elif namespace == "restconf":
		_, delete_code = api.delete_slice_services()

		assert delete_code in [200, 204, 404], f"Deletion failed for slice '{slice_id}' in namespace '{namespace}'"
		No newline at end of file

src/tests/test_initialization.py

+63 −9

File changed.

Preview size limit exceeded, changes collapsed.