Coverage for src/zapy/api/connection.py: 100%

1import json 

2import socket 

3import sys 

4from pathlib import Path 

5from typing import cast 

6 

7from .models import ServerConnection 

8 

9 

10def load_server_config() -> ServerConnection: 

11 virtual_env_path = Path(sys.prefix) 

12 zapy_etc = virtual_env_path / "etc" / "zapy" 

13 zapy_etc.mkdir(parents=True, exist_ok=True) 

14 

15 conn_file = zapy_etc / "connection.json" 

16 

17 if not conn_file.exists(): 

18 conn = create_connection(conn_file) 

19 else: 

20 conn = read_connection(conn_file) 

21 

22 return conn 

23 

24 

25def create_connection(conn_path: str | Path) -> ServerConnection: 

26 conn = ServerConnection( 

27 host="127.0.0.1", 

28 port=get_random_free_port(), 

29 ) 

30 with open(conn_path, "w") as outfile: 

31 outfile.write(conn.model_dump_json()) 

32 return conn 

33 

34 

35def read_connection(conn_path: str | Path) -> ServerConnection: 

36 with open(conn_path) as outfile: 

37 raw_data = json.load(outfile) 

38 return ServerConnection.model_validate(raw_data) 

39 

40 

41def get_random_free_port() -> int: 

42 sock = socket.socket() 

43 sock.bind(("", 0)) 

44 port = sock.getsockname()[1] 

45 sock.close() 

46 return cast(int, port)