30184746367

Committed 26 Jul 2026 02:33AM UTC coverage: 43.905% (+0.03%) from 43.876%

Build # 30184746367

Build Type

push

github

Committed by

ChristianTremblay

Commit Message

failing test check up

Coverage Stats

4 of 5 new or added lines in 2 files covered. (80.0%)

24 existing lines in 1 file now uncovered.

2489 of 5669 relevant lines covered (43.91%)

0.44 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.7

/BAC0/scripts/Base.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 by Christian Tremblay, P.Eng <christian.tremblay@servisys.com>
# Licensed under LGPLv3, see file LICENSE in this source tree.
#
"""
Doc here
"""
import asyncio
import random
import sys
import typing as t
from collections import defaultdict

# --- standard Python modules ---
from bacpypes3.basetypes import DeviceStatus, HostNPort, ObjectTypesSupported
from bacpypes3.json.util import sequence_to_json
from bacpypes3.local.device import DeviceObject
from bacpypes3.local.networkport import NetworkPortObject
from bacpypes3.pdu import Address
from bacpypes3.primitivedata import CharacterString, ObjectIdentifier
from bacpypes3.vendor import VendorInfo, get_vendor_info

# --- this application's modules ---
from ..core.app.asyncApp import (
    BAC0Application,
)  # BAC0BBMDDeviceApplication,; BAC0ForeignDeviceApplication,
from ..core.functions.GetIPAddr import validate_ip_address
from ..core.functions.TimeSync import TimeHandler
from ..core.io.IOExceptions import InitializationError, UnknownObjectError
from ..core.utils.notes import note_and_log
from ..tasks.TaskManager import stopAllTasks

# --- 3rd party modules ---

# ------------------------------------------------------------------------------


@note_and_log
class LocalObjects(object):
    def __init__(self, device):
        self.device = device

    def __getitem__(self, obj):
        item = None
        if isinstance(obj, tuple):
            obj_type, instance = obj
            item = self.device.this_application.app.get_object_id((obj_type, instance))
        elif isinstance(obj, str):
            name = obj
            item = self.device.this_application.app.get_object_name(name)
        if item is None:
            raise UnknownObjectError(f"Can't find {obj} in local device")
        else:
            return item


def charstring(val):
    return CharacterString(val) if isinstance(val, str) else val

class DiscoveredDevice(t.TypedDict):
    object_instance: ObjectIdentifier
    address: Address
    network_number: t.Set[int]
    vendor_id: int
    vendor_name: str

@note_and_log
class Base:
    """
    Build a running BACnet/IP device that accepts WhoIs and IAm requests
    Initialization requires some minimial information about the local device.

    :param localIPAddr='127.0.0.1':
    :param localObjName='BAC0':
    :param deviceId=None:
    :param maxAPDULengthAccepted='1024':
    :param maxSegmentsAccepted='1024':
    :param segmentationSupported='segmentedBoth':
    """

    _used_ips: t.Set[Address] = set()
    _last_cov_identifier = 0
    _running_cov_tasks = {}

    def __init__(
        self,
        localIPAddr: Address = Address("127.0.0.1/24"),
        networkNumber: t.Optional[int] = None,
        localObjName: str = "BAC0",
        deviceId: t.Optional[int] = None,
        firmwareRevision: str = "".join(sys.version.split("|")[:2]),
        maxAPDULengthAccepted: str = "1024",
        maxSegmentsAccepted: str = "1024",
        segmentationSupported: str = "segmentedBoth",
        bbmdAddress: t.Optional[str] = None,
        bbmdTTL: t.Optional[int] = 0,
        bdtable: t.Optional[list] = None,
        modelName: str = "BAC0 Scripting Tool",
        vendorId: int = 842,
        vendorName: str = "SERVISYS inc.",
        description: str = "http://christiantremblay.github.io/BAC0/",
        location: str = "Bromont, Québec",
        timezone: str = "America/Montreal",
        json_file: t.Optional[str] = None,
    ):
        self.log("Configurating app", level="debug")

        # Register Servisys
        try:
            _BAC0_vendor = VendorInfo(vendorId)
            _BAC0_vendor.register_object_class(
                ObjectTypesSupported.networkPort, NetworkPortObject
            )
            _BAC0_vendor.register_object_class(ObjectTypesSupported.device, DeviceObject)
        except RuntimeError:
            pass  # we are re-running the script... forgive us
            _BAC0_vendor = get_vendor_info(vendorId)


        self.timehandler = TimeHandler(tz=timezone)

        self.response = None
        self._initialized = False
        self._started = False
        self._stopped = False

        if localIPAddr in Base._used_ips:
            raise InitializationError(
                "IP Address provided ({}) already used by BAC0. Check if another software is using port 47808 on this network interface. If so, you can define multiple IP per interface. Or specify another IP using BAC0.lite(ip='IP/mask')".format(
                    localIPAddr
                )
            )

        if validate_ip_address(localIPAddr):
            self.localIPAddr = localIPAddr
        else:
            raise InitializationError(
                "IP Address provided ({}) invalid. Check if another software is using port 47808 on this network interface. If so, you can define multiple IP per interface. Or specify another IP using BAC0.lite(ip='IP/mask')".format(
                    localIPAddr
                )
            )
        self.networkNumber = networkNumber

        self.Boid = (
            int(deviceId) if deviceId else (3056177 + int(random.uniform(0, 1000)))
        )

        self.segmentationSupported = segmentationSupported
        self.maxSegmentsAccepted = maxSegmentsAccepted
        self.localObjName = localObjName
        self.local_objects = LocalObjects(device=self)

        self.maxAPDULengthAccepted = maxAPDULengthAccepted
        self.vendorId = vendorId
        self.vendorName = charstring(vendorName)
        self.modelName = charstring(modelName)
        self.description = charstring(description)
        self.location = charstring(location)

        self.discoveredDevices: t.Optional[t.Dict[str, DiscoveredDevice]] = None
        self.systemStatus = DeviceStatus(1)

        self.bbmdAddress = bbmdAddress
        self.bbmdTTL = bbmdTTL
        self.bdtable = bdtable

        self.firmwareRevision = firmwareRevision
        self._ric = {}
        self.subscription_contexts = {}
        # Cannot reference db.InfluxDB directly since it's an optional import
        self.database: t.Optional[t.Any] = None
        self.json_file = json_file

        try:
            self.startApp()
        except InitializationError as error:
            raise InitializationError(
                f"Gros probleme : {error}. Address requested : {localIPAddr}"
            )

    def startApp(self):
        """
        Define the local device, including services supported.
        Once defined, start the BACnet stack in its own thread.
        """
        self.log("Create Local Device", level="debug")
        try:
            app_type = "BACnet/IP App"

            class config(defaultdict):
                "Simple class to mimic args dot retrieval"

                def __init__(self, cfg):
                    for k, v in cfg.items():
                        self[k] = v

                def __getattr__(self, key):
                    return self[key]

            if self.bbmdAddress is not None:
                mode = "foreign"
            elif self.bdtable:
                mode = "bbmd"
            else:
                mode = "normal"
            cfg = {
                "BAC0": {
                    "bbmdAddress": self.bbmdAddress,
                    "bdt": self.bdtable,
                    "ttl": self.bbmdTTL,
                },
                "device": {
                    "object-name": self.localObjName,
                    # "firmware-revision": self.firmwareRevision,
                    "vendor-identifier": self.vendorId,
                    "vendor-name": "Servisys inc.",
                    "object-identifier": f"device,{self.Boid}",
                    "object-list": [f"device,{self.Boid}", "network-port,1"],
                    "model-name": self.modelName,
                    # "max-apdu-length-accepted": self.maxAPDULengthAccepted,
                    # "max-segments-accepted": self.maxSegmentsAccepted,
                    # "location": self.location,
                    # "description": self.description
                },
                "network-port": {
                    "ip-address": str(self.localIPAddr),
                    "ip-subnet-mask": str(self.localIPAddr.netmask),
                    "bacnet-ip-udp-port": self.localIPAddr.addrPort,
                    "network-number": None,
                    "fd-bbmd-address": sequence_to_json(HostNPort(self.bbmdAddress)),
                    "fd-subscription-lifetime": self.bbmdTTL,
                    "bacnet-ip-mode": mode,
                },
            }
            if mode == "bbmd":
                # bdt_json_seq = [f"BDTEntry({addr})" for addr in self.bdtable]
                cfg["network-port"]["bbmdBroadcastDistributionTable"] = self.bdtable

            _cfg = config(cfg)

            self.this_application = BAC0Application(
                _cfg, self.localIPAddr, json_file=self.json_file
            )
            if mode == "bbmd":
                self._log.info(f"Populating BDT with {self.bdtable}")
                self.this_application.populate_bdt()

            if mode == "foreign":
                self._log.info(
                    f"Registering as a foreign device to host {self.bbmdAddress} for {self.bbmdTTL} seconds"
                )
                if self.bbmdAddress is None or self.bbmdTTL is None:
                    raise ValueError("Missing bbmdAddress and/or bbmdTTL")
                self.this_application.register_as_foreign_device_to(
                    host=self.bbmdAddress, lifetime=self.bbmdTTL
                )

            self.log("Starting", level="debug")
            self._initialized = True

            try:
                Base._used_ips.add(self.localIPAddr)
                self.log(f"Registered as {app_type} | mode {mode}", level="info")
                self._started = True
            except OSError as error:
                self.log(f"Error opening socket: {error}", level="warning")
                raise InitializationError(f"Error opening socket: {error}")
            self.log("Running", level="debug")
        except OSError as error:
            self.log(f"an error has occurred: {error}", level="error")
            raise InitializationError(f"Error starting app: {error}")
            self.log("finally", level="debug")

    def register_foreign_device(self, addr=None, ttl=0):
        # self.this_application.register_to_bbmd(addr, ttl)
        raise NotImplementedError()

    def unregister_foreign_device(self):
        self.this_application.unregister_from_bbmd()

    def disconnect(self) -> asyncio.Task:
        task = asyncio.create_task(self._disconnect())
        return task

    async def _disconnect(self):
        """
        Stop the BACnet stack.  Free the IP socket.
        """
        self.log("Stopping All running tasks", level="debug")
        await stopAllTasks()
        self.log("Stopping BACnet stack", level="debug")
        # Freeing socket
        self.this_application.app.close()

        self._stopped = True  # Stop stack thread
        # self.t.join()
        self._started = False
        Base._used_ips.discard(self.localIPAddr)
        self.log("BACnet stopped", level="info")

    @property
    def routing_table(self):
        """
        Routing Table will give all the details about routers and how they
        connect BACnet networks together.

        It's a decoded presentation of what bacpypes.router_info_cache contains.

        Returns a dict with the address of routers as key.
        """

        class Router:
            def __init__(self, snet, address, dnets, path=None):
                self.source_network: int = snet
                self.address: Address = address
                self.destination_networks: set = dnets
                self.path: list = path

            def __repr__(self):
                return "Source Network: {} | Address: {} | Destination Networks: {} | Path: {}".format(
                    self.source_network,
                    self.address,
                    self.destination_networks,
                    self.path,
                )

        self._routers = {}

        self._ric = self.this_application.app.nsap.router_info_cache

        for router, dnets in self._ric.router_dnets.items():
            snet, address = router
            self._routers[str(address)] = Router(snet, address, dnets, path=[])
        for path, router_info in self._ric.path_info.items():
            router_address, router_status = router_info
            snet, dnet = path
            self._routers[str(router_address)].path.append((path, router_status))

        return self._routers

    @classmethod
    def extract_value_from_primitive_data(cls, value):
        if isinstance(value, float):
            return float(value)
        # elif isinstance(value, Boolean):
        #    if value == int(1):
        #        return True
        #    else:
        #        return False
        elif isinstance(value, int):
            return int(value)
        elif isinstance(value, str):
            return str(value)
        else:
            return value

1	#!/usr/bin/python
2	# -- coding: utf-8 --
3	#
4	# Copyright (C) 2015 by Christian Tremblay, P.Eng <christian.tremblay@servisys.com>
5	# Licensed under LGPLv3, see file LICENSE in this source tree.
6	#
7	"""
8	Doc here
9	"""
10	import asyncio	1✔
11	import random	1✔
12	import sys	1✔
13	import typing as t	1✔
14	from collections import defaultdict	1✔
15
16	# --- standard Python modules ---
17	from bacpypes3.basetypes import DeviceStatus, HostNPort, ObjectTypesSupported	1✔
18	from bacpypes3.json.util import sequence_to_json	1✔
19	from bacpypes3.local.device import DeviceObject	1✔
20	from bacpypes3.local.networkport import NetworkPortObject	1✔
21	from bacpypes3.pdu import Address	1✔
22	from bacpypes3.primitivedata import CharacterString, ObjectIdentifier	1✔
23	from bacpypes3.vendor import VendorInfo, get_vendor_info	1✔
24
25	# --- this application's modules ---
26	from ..core.app.asyncApp import (	1✔
27	BAC0Application,
28	) # BAC0BBMDDeviceApplication,; BAC0ForeignDeviceApplication,
29	from ..core.functions.GetIPAddr import validate_ip_address	1✔
30	from ..core.functions.TimeSync import TimeHandler	1✔
31	from ..core.io.IOExceptions import InitializationError, UnknownObjectError	1✔
32	from ..core.utils.notes import note_and_log	1✔
33	from ..tasks.TaskManager import stopAllTasks	1✔
34
35	# --- 3rd party modules ---
36
37	# ------------------------------------------------------------------------------
38
39
40	@note_and_log	1✔
41	class LocalObjects(object):	1✔
42	def __init__(self, device):	1✔
43	self.device = device	1✔
44
45	def __getitem__(self, obj):	1✔
46	item = None	×
47	if isinstance(obj, tuple):	×
48	obj_type, instance = obj	×
49	item = self.device.this_application.app.get_object_id((obj_type, instance))	×
50	elif isinstance(obj, str):	×
51	name = obj	×
52	item = self.device.this_application.app.get_object_name(name)	×
53	if item is None:	×
54	raise UnknownObjectError(f"Can't find {obj} in local device")	×
55	else:
56	return item	×
57
58
59	def charstring(val):	1✔
60	return CharacterString(val) if isinstance(val, str) else val	1✔
61
62	class DiscoveredDevice(t.TypedDict):	1✔
63	object_instance: ObjectIdentifier
64	address: Address
65	network_number: t.Set[int]
66	vendor_id: int
67	vendor_name: str
68
69	@note_and_log	1✔
70	class Base:	1✔
71	"""
72	Build a running BACnet/IP device that accepts WhoIs and IAm requests
73	Initialization requires some minimial information about the local device.
74
75	:param localIPAddr='127.0.0.1':
76	:param localObjName='BAC0':
77	:param deviceId=None:
78	:param maxAPDULengthAccepted='1024':
79	:param maxSegmentsAccepted='1024':
80	:param segmentationSupported='segmentedBoth':
81	"""
82
83	_used_ips: t.Set[Address] = set()	1✔
84	_last_cov_identifier = 0	1✔
85	_running_cov_tasks = {}	1✔
86
87	def __init__(	1✔
88	self,
89	localIPAddr: Address = Address("127.0.0.1/24"),
90	networkNumber: t.Optional[int] = None,
91	localObjName: str = "BAC0",
92	deviceId: t.Optional[int] = None,
93	firmwareRevision: str = "".join(sys.version.split("\|")[:2]),
94	maxAPDULengthAccepted: str = "1024",
95	maxSegmentsAccepted: str = "1024",
96	segmentationSupported: str = "segmentedBoth",
97	bbmdAddress: t.Optional[str] = None,
98	bbmdTTL: t.Optional[int] = 0,
99	bdtable: t.Optional[list] = None,
100	modelName: str = "BAC0 Scripting Tool",
101	vendorId: int = 842,
102	vendorName: str = "SERVISYS inc.",
103	description: str = "http://christiantremblay.github.io/BAC0/",
104	location: str = "Bromont, Québec",
105	timezone: str = "America/Montreal",
106	json_file: t.Optional[str] = None,
107	):
108	self.log("Configurating app", level="debug")	1✔
109
110	# Register Servisys
111	try:	1✔
112	_BAC0_vendor = VendorInfo(vendorId)	1✔
113	_BAC0_vendor.register_object_class(	1✔
114	ObjectTypesSupported.networkPort, NetworkPortObject
115	)
116	_BAC0_vendor.register_object_class(ObjectTypesSupported.device, DeviceObject)	1✔
117	except RuntimeError:	1✔
118	pass # we are re-running the script... forgive us	1✔
119	_BAC0_vendor = get_vendor_info(vendorId)	1✔
120
121
122	self.timehandler = TimeHandler(tz=timezone)	1✔
123
124	self.response = None	1✔
125	self._initialized = False	1✔
126	self._started = False	1✔
127	self._stopped = False	1✔
128
129	if localIPAddr in Base._used_ips:	1✔
UNCOV 130	raise InitializationError(	×
131	"IP Address provided ({}) already used by BAC0. Check if another software is using port 47808 on this network interface. If so, you can define multiple IP per interface. Or specify another IP using BAC0.lite(ip='IP/mask')".format(
132	localIPAddr
133	)
134	)
135
136	if validate_ip_address(localIPAddr):	1✔
137	self.localIPAddr = localIPAddr	1✔
138	else:
UNCOV 139	raise InitializationError(	×
140	"IP Address provided ({}) invalid. Check if another software is using port 47808 on this network interface. If so, you can define multiple IP per interface. Or specify another IP using BAC0.lite(ip='IP/mask')".format(
141	localIPAddr
142	)
143	)
144	self.networkNumber = networkNumber	1✔
145
146	self.Boid = (	1✔
147	int(deviceId) if deviceId else (3056177 + int(random.uniform(0, 1000)))
148	)
149
150	self.segmentationSupported = segmentationSupported	1✔
151	self.maxSegmentsAccepted = maxSegmentsAccepted	1✔
152	self.localObjName = localObjName	1✔
153	self.local_objects = LocalObjects(device=self)	1✔
154
155	self.maxAPDULengthAccepted = maxAPDULengthAccepted	1✔
156	self.vendorId = vendorId	1✔
157	self.vendorName = charstring(vendorName)	1✔
158	self.modelName = charstring(modelName)	1✔
159	self.description = charstring(description)	1✔
160	self.location = charstring(location)	1✔
161
162	self.discoveredDevices: t.Optional[t.Dict[str, DiscoveredDevice]] = None	1✔
163	self.systemStatus = DeviceStatus(1)	1✔
164
165	self.bbmdAddress = bbmdAddress	1✔
166	self.bbmdTTL = bbmdTTL	1✔
167	self.bdtable = bdtable	1✔
168
169	self.firmwareRevision = firmwareRevision	1✔
170	self._ric = {}	1✔
171	self.subscription_contexts = {}	1✔
172	# Cannot reference db.InfluxDB directly since it's an optional import
173	self.database: t.Optional[t.Any] = None	1✔
174	self.json_file = json_file	1✔
175
176	try:	1✔
177	self.startApp()	1✔
178	except InitializationError as error:	×
UNCOV 179	raise InitializationError(	×
180	f"Gros probleme : {error}. Address requested : {localIPAddr}"
181	)
182
183	def startApp(self):	1✔
184	"""
185	Define the local device, including services supported.
186	Once defined, start the BACnet stack in its own thread.
187	"""
188	self.log("Create Local Device", level="debug")	1✔
189	try:	1✔
190	app_type = "BACnet/IP App"	1✔
191
192	class config(defaultdict):	1✔
193	"Simple class to mimic args dot retrieval"
194
195	def __init__(self, cfg):	1✔
196	for k, v in cfg.items():	1✔
197	self[k] = v	1✔
198
199	def __getattr__(self, key):	1✔
UNCOV 200	return self[key]	×
201
202	if self.bbmdAddress is not None:	1✔
UNCOV 203	mode = "foreign"	×
204	elif self.bdtable:	1✔
UNCOV 205	mode = "bbmd"	×
206	else:
207	mode = "normal"	1✔
208	cfg = {	1✔
209	"BAC0": {
210	"bbmdAddress": self.bbmdAddress,
211	"bdt": self.bdtable,
212	"ttl": self.bbmdTTL,
213	},
214	"device": {
215	"object-name": self.localObjName,
216	# "firmware-revision": self.firmwareRevision,
217	"vendor-identifier": self.vendorId,
218	"vendor-name": "Servisys inc.",
219	"object-identifier": f"device,{self.Boid}",
220	"object-list": [f"device,{self.Boid}", "network-port,1"],
221	"model-name": self.modelName,
222	# "max-apdu-length-accepted": self.maxAPDULengthAccepted,
223	# "max-segments-accepted": self.maxSegmentsAccepted,
224	# "location": self.location,
225	# "description": self.description
226	},
227	"network-port": {
228	"ip-address": str(self.localIPAddr),
229	"ip-subnet-mask": str(self.localIPAddr.netmask),
230	"bacnet-ip-udp-port": self.localIPAddr.addrPort,
231	"network-number": None,
232	"fd-bbmd-address": sequence_to_json(HostNPort(self.bbmdAddress)),
233	"fd-subscription-lifetime": self.bbmdTTL,
234	"bacnet-ip-mode": mode,
235	},
236	}
237	if mode == "bbmd":	1✔
238	# bdt_json_seq = [f"BDTEntry({addr})" for addr in self.bdtable]
UNCOV 239	cfg["network-port"]["bbmdBroadcastDistributionTable"] = self.bdtable	×
240
241	_cfg = config(cfg)	1✔
242
243	self.this_application = BAC0Application(	1✔
244	_cfg, self.localIPAddr, json_file=self.json_file
245	)
246	if mode == "bbmd":	1✔
247	self._log.info(f"Populating BDT with {self.bdtable}")	×
UNCOV 248	self.this_application.populate_bdt()	×
249
250	if mode == "foreign":	1✔
UNCOV 251	self._log.info(	×
252	f"Registering as a foreign device to host {self.bbmdAddress} for {self.bbmdTTL} seconds"
253	)
254	if self.bbmdAddress is None or self.bbmdTTL is None:	×
255	raise ValueError("Missing bbmdAddress and/or bbmdTTL")	×
UNCOV 256	self.this_application.register_as_foreign_device_to(	×
257	host=self.bbmdAddress, lifetime=self.bbmdTTL
258	)
259
260	self.log("Starting", level="debug")	1✔
261	self._initialized = True	1✔
262
263	try:	1✔
264	Base._used_ips.add(self.localIPAddr)	1✔
265	self.log(f"Registered as {app_type} \| mode {mode}", level="info")	1✔
266	self._started = True	1✔
267	except OSError as error:	×
268	self.log(f"Error opening socket: {error}", level="warning")	×
UNCOV 269	raise InitializationError(f"Error opening socket: {error}")	×
270	self.log("Running", level="debug")	1✔
271	except OSError as error:	×
272	self.log(f"an error has occurred: {error}", level="error")	×
UNCOV 273	raise InitializationError(f"Error starting app: {error}")	×
274	self.log("finally", level="debug")
275
276	def register_foreign_device(self, addr=None, ttl=0):	1✔
277	# self.this_application.register_to_bbmd(addr, ttl)
UNCOV 278	raise NotImplementedError()	×
279
280	def unregister_foreign_device(self):	1✔
UNCOV 281	self.this_application.unregister_from_bbmd()	×
282
283	def disconnect(self) -> asyncio.Task:	1✔
284	task = asyncio.create_task(self._disconnect())	×
UNCOV 285	return task	×
286
287	async def _disconnect(self):	1✔
288	"""
289	Stop the BACnet stack. Free the IP socket.
290	"""
291	self.log("Stopping All running tasks", level="debug")	1✔
292	await stopAllTasks()	1✔
293	self.log("Stopping BACnet stack", level="debug")	1✔
294	# Freeing socket
295	self.this_application.app.close()	1✔
296
297	self._stopped = True # Stop stack thread	1✔
298	# self.t.join()
299	self._started = False	1✔
300	Base._used_ips.discard(self.localIPAddr)	1✔
301	self.log("BACnet stopped", level="info")	1✔
302
303	@property	1✔
304	def routing_table(self):	1✔
305	"""
306	Routing Table will give all the details about routers and how they
307	connect BACnet networks together.
308
309	It's a decoded presentation of what bacpypes.router_info_cache contains.
310
311	Returns a dict with the address of routers as key.
312	"""
313
314	class Router:	×
315	def __init__(self, snet, address, dnets, path=None):	×
316	self.source_network: int = snet	×
317	self.address: Address = address	×
318	self.destination_networks: set = dnets	×
UNCOV 319	self.path: list = path	×
320
321	def __repr__(self):	×
UNCOV 322	return "Source Network: {} \| Address: {} \| Destination Networks: {} \| Path: {}".format(	×
323	self.source_network,
324	self.address,
325	self.destination_networks,
326	self.path,
327	)
328
UNCOV 329	self._routers = {}	×
330
UNCOV 331	self._ric = self.this_application.app.nsap.router_info_cache	×
332
333	for router, dnets in self._ric.router_dnets.items():	×
334	snet, address = router	×
335	self._routers[str(address)] = Router(snet, address, dnets, path=[])	×
336	for path, router_info in self._ric.path_info.items():	×
337	router_address, router_status = router_info	×
338	snet, dnet = path	×
UNCOV 339	self._routers[str(router_address)].path.append((path, router_status))	×
340
UNCOV 341	return self._routers	×
342
343	@classmethod	1✔
344	def extract_value_from_primitive_data(cls, value):	1✔
345	if isinstance(value, float):	×
UNCOV 346	return float(value)	×
347	# elif isinstance(value, Boolean):
348	# if value == int(1):
349	# return True
350	# else:
351	# return False
352	elif isinstance(value, int):	×
353	return int(value)	×
354	elif isinstance(value, str):	×
UNCOV 355	return str(value)	×
356	else:
UNCOV 357	return value	×

ChristianTremblay / BAC0 / 30184746367

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous