17030550552

Committed 18 Aug 2025 03:56AM UTC coverage: 40.475% (-0.2%) from 40.7%

Build # 17030550552

Build Type

Pull #564

github

Committed by

web-flow

Commit Message

Merge 830537094 into c4d8486a9

Pull Request Pull Request #564: Mypy

Run Details

70 of 171 new or added lines in 22 files covered. (40.94%)

13 existing lines in 8 files now uncovered.

2248 of 5554 relevant lines covered (40.48%)

0.4 hits per line

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

0.0

/BAC0/scripts/script_runner.py

import asyncio
from signal import SIGINT, SIGTERM, signal


def run(main_task, bacnet):
    #global loop
    try:
        loop = (
            asyncio.get_running_loop()
        )  # we also could have chosen get_event_loop(), and if we did we would not have needed to set_event_loop() and create a new event loop
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    def handler(sig, sig2):
        if bacnet is not None:
            bacnet._log.info(f"Got signal: {sig!s}, shutting down.")
            bacnet.disconnect()
        loop.stop()


    # for sig in (SIGINT, SIGTERM):
    #    loop.add_signal_handler(sig, handler)
    signal(SIGINT, handler)
    signal(SIGTERM, handler)

    loop.create_task(main_task())
    loop.run_forever()
    tasks = asyncio.all_tasks(loop=loop)
    for t in tasks:
        t.cancel()
    group = asyncio.gather(*tasks, return_exceptions=True)
    loop.run_until_complete(group)
    loop.close()

UNCOV 1	import asyncio	×
UNCOV 2	from signal import SIGINT, SIGTERM, signal	×
3
4
UNCOV 5	def run(main_task, bacnet):	×
6	#global loop
UNCOV 7	try:	×
8	loop = (	×
9	asyncio.get_running_loop()
10	) # we also could have chosen get_event_loop(), and if we did we would not have needed to set_event_loop() and create a new event loop
11	except RuntimeError:	×
12	loop = asyncio.new_event_loop()	×
13	asyncio.set_event_loop(loop)	×
14
NEW 15	def handler(sig, sig2):	×
NEW 16	if bacnet is not None:	×
NEW 17	bacnet._log.info(f"Got signal: {sig!s}, shutting down.")	×
NEW 18	bacnet.disconnect()	×
NEW 19	loop.stop()	×
20
21
22	# for sig in (SIGINT, SIGTERM):
23	# loop.add_signal_handler(sig, handler)
24	signal(SIGINT, handler)	×
25	signal(SIGTERM, handler)	×
26
27	loop.create_task(main_task())	×
28	loop.run_forever()	×
29	tasks = asyncio.all_tasks(loop=loop)	×
30	for t in tasks:	×
31	t.cancel()	×
32	group = asyncio.gather(*tasks, return_exceptions=True)	×
33	loop.run_until_complete(group)	×
34	loop.close()	×

ChristianTremblay / BAC0 / 17030550552

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