Coverage for microservice_websocket/app/utils/node.py: 81%

1from datetime import datetime, timedelta 

2 

3from beanie import PydanticObjectId 

4 

5from ..config import config as Config 

6from ..services.database import Application, Node, NodeSettings 

7from .enums import EventType, NodeState 

8from .exceptions import NotFoundException 

9from .node_settings import send_update_settings 

10 

11 

12def on_keep_alive(current_state: NodeState) -> NodeState: 

13 if current_state == NodeState.ERROR: 

14 return NodeState.READY 

15 

16 return current_state 

17 

18 

19def update_state( 

20 current_state: NodeState, 

21 lastSeenAt: datetime, 

22 event: EventType | None = None, 

23) -> NodeState: 

24 

25 if (datetime.now() - lastSeenAt) > timedelta( 

26 seconds=Config.app.NODE_TIMEOUT_INTERVAL 

27 ): 

28 return NodeState.ERROR 

29 

30 if event is not None: 30 ↛ 46line 30 didn't jump to line 46, because the condition on line 30 was never false

31 if event == EventType.START_REC: 

32 current_state = NodeState.RUNNING 

33 elif event == EventType.STOP_REC: 

34 current_state = NodeState.READY 

35 elif event == EventType.RAISE_ALERT: 

36 current_state = NodeState.ALERT_READY 

37 elif event == EventType.HANDLE_ALERT: 

38 current_state = NodeState.READY 

39 elif event == EventType.KEEP_ALIVE: 

40 current_state = on_keep_alive(current_state) 

41 elif event == EventType.ON_LAUNCH: 41 ↛ 44line 41 didn't jump to line 44, because the condition on line 41 was never false

42 current_state = NodeState.READY 

43 else: 

44 raise NotImplementedError(f"EventType '{event}' not implemented yet") 

45 

46 return current_state 

47 

48 

49async def get_nodes(applicationID: str): 

50 application: Application | None = await Application.get( 

51 PydanticObjectId(applicationID) 

52 ) 

53 if application is None: 53 ↛ 54line 53 didn't jump to line 54, because the condition on line 53 was never true

54 raise NotFoundException("Application") 

55 

56 nodes: list[Node] = await Node.find(Node.application == application.id).to_list() 

57 

58 return nodes 

59 

60 

61async def on_launch(node: Node): 

62 settings = await NodeSettings.find_one(NodeSettings.node == node.id) 

63 if settings is None: 

64 print(f"No settings found for node: {node}") 

65 return 

66 

67 await send_update_settings(str(node.application), node.nodeID, settings)