Coverage for microservice_websocket/app/utils/session.py: 100%

1from beanie.operators import And, Eq 

2 

3from ..services.database import Node, Reading 

4from .exceptions import NotFoundException 

5 

6 

7async def get_session(nodeID: int, sessionID: int | None) -> list[Reading]: 

8 node: Node | None = await Node.find_one(Node.nodeID == nodeID) 

9 if node is None: 

10 raise NotFoundException("Node") 

11 

12 if sessionID is None: 

13 latest_reading: Reading | None = await ( 

14 Reading.find(Reading.node == node.id).sort("-sessionID").first_or_none() 

15 ) 

16 if latest_reading is None: 

17 return [] 

18 

19 sessionID = latest_reading.sessionID 

20 

21 readings: list[Reading] = await Reading.find( 

22 And( 

23 Eq(Reading.node, node.id), 

24 Eq(Reading.sessionID, sessionID), 

25 ) 

26 ).to_list() 

27 

28 return [x for x in readings] 

29 

30 

31async def get_sessions_id(nodeID: int) -> list[int]: 

32 node: Node | None = await Node.find_one(Node.nodeID == nodeID) 

33 if node is None: 

34 raise NotFoundException("Node") 

35 

36 sessions: list[Reading] = await Reading.find(Reading.node == node.id).to_list() 

37 

38 sessions_id = [x.sessionID for x in sessions] 

39 

40 # Remove duplicates 

41 sessions_id = list(set(sessions_id)) 

42 

43 return sessions_id