Coverage for src/unit_cooler/webui/worker.py: 80%

1#!/usr/bin/env python3 

2import logging 

3import threading 

4import traceback 

5 

6import my_lib.footprint 

7 

8import unit_cooler.const 

9import unit_cooler.pubsub.subscribe 

10 

11# グローバル終了フラグ 

12should_terminate = threading.Event() 

13 

14 

15def term(): 

16 """終了フラグを設定する関数""" 

17 should_terminate.set() 

18 logging.info("Termination flag set for webui worker") 

19 

20 

21def queue_put(message_queue, message, liveness_file): 

22 message["state"] = unit_cooler.const.COOLING_STATE(message["state"]) 

23 

24 if message_queue.full(): 

25 message_queue.get() 

26 

27 logging.info("Receive message: %s", message) 

28 

29 message_queue.put(message) 

30 my_lib.footprint.update(liveness_file) 

31 

32 

33# NOTE: 制御メッセージを Subscribe して、キューに積み、cooler_stat.py で WebUI に渡すワーカ 

34def subscribe_worker(config, control_host, pub_port, message_queue, liveness_file, msg_count=0): # noqa: PLR0913 

35 logging.info("Start webui subscribe worker (%s:%d)", control_host, pub_port) 

36 

37 ret = 0 

38 try: 

39 # 終了フラグを渡してstart_clientを呼び出し 

40 unit_cooler.pubsub.subscribe.start_client( 

41 control_host, 

42 pub_port, 

43 lambda message: queue_put(message_queue, message, liveness_file), 

44 msg_count, 

45 should_terminate, 

46 ) 

47 except Exception: 

48 logging.exception("Failed to receive control message") 

49 unit_cooler.util.notify_error(config, traceback.format_exc()) 

50 ret = -1 

51 

52 logging.warning("Stop subscribe worker") 

53 

54 return ret