Coverage for netbox_data_import/_dns_worker.py: 88.89%

1# SPDX-License-Identifier: Apache-2.0 

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""Resolve one host with libc in a deadline-bounded child process.""" 

4 

5import json 

6import signal 

7import socket 

8import sys 

9 

10from typing import TextIO 

11 

12 

13def main(stdin: TextIO, stdout: TextIO) -> int: 

14 """Read one request from stdin and write its ordered addresses to stdout.""" 

15 try: 

16 host, port, timeout = json.load(stdin) 

17 except Exception: # noqa: BLE001 - the parent exposes one typed resolution failure 

18 return 1 

19 # The alarm is the deadline. An unarmed worker must crash, not read as a failed lookup. 

20 signal.setitimer(signal.ITIMER_REAL, timeout) 

21 try: 

22 answers = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP) 

23 addresses = list(dict.fromkeys(str(answer[4][0]) for answer in answers)) 

24 except Exception: # noqa: BLE001 - the parent exposes one typed resolution failure 

25 return 1 

26 json.dump(addresses, stdout) 

27 return 0 

28 

29 

30if __name__ == "__main__": # pragma: no cover - child entry point; main() is tested in process 

31 raise SystemExit(main(sys.stdin, sys.stdout))