9357576700

Committed 03 Jun 2024 09:35PM UTC coverage: 80.992% (+0.4%) from 80.594%

Build # 9357576700

Build Type

Pull #211

github

Committed by

web-flow

Commit Message

Merge f8f7a1ff9 into 9d80cb79d

Pull Request Pull Request #211: DNS server basic implementation WIP

Run Details

1009 of 1354 branches covered (74.52%)

Branch coverage included in aggregate %.

307 of 359 new or added lines in 7 files covered. (85.52%)

12 existing lines in 1 file now uncovered.

3823 of 4612 relevant lines covered (82.89%)

4.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.49

/aiomisc/service/dns/store.py

from typing import Optional, Sequence, Tuple

from .records import DNSRecord, RecordType
from .tree import RadixTree
from .zone import DNSZone


class DNSStore:
    zones: RadixTree[DNSZone]

    __slots__ = ("zones",)

    def __init__(self) -> None:
        self.zones = RadixTree()

    def add_zone(self, zone: DNSZone) -> None:
        zone_tuple = self.get_reverse_tuple(zone.name)
        if self.zones.search(zone_tuple):
            raise ValueError(f"Zone {zone.name} already exists.")
        self.zones.insert(zone_tuple, zone)

    def remove_zone(self, zone_name: str) -> None:
        zone_tuple = self.get_reverse_tuple(zone_name)
        if not self.zones.search(zone_tuple):
            raise ValueError(
                f"Zone {zone_name} does not exist.",
            )
        # Clear zone from RadixTree
        self.zones.insert(zone_tuple, None)

    def get_zone(self, zone_name: str) -> Optional[DNSZone]:
        zone_tuple = self.get_reverse_tuple(zone_name)
        return self.zones.search(zone_tuple)

    def query(self, name: str, record_type: RecordType) -> Sequence[DNSRecord]:
        if not name.endswith("."):
            name += "."
        zone_tuple = self.get_zone_for_name(name)
        if not zone_tuple:
            return ()
        zone = self.zones.search(zone_tuple)
        return zone.get_records(name, record_type) if zone is not None else ()

    def get_zone_for_name(self, name: str) -> Optional[Tuple[str, ...]]:
        labels = self.get_reverse_tuple(name)
        result = self.zones.find_prefix(labels)
        return result[0] if result else None

    @staticmethod
    def get_reverse_tuple(zone_name: str) -> Tuple[str, ...]:
        return tuple(zone_name.strip(".").split("."))[::-1]

1	from typing import Optional, Sequence, Tuple	5✔
2
3	from .records import DNSRecord, RecordType	5✔
4	from .tree import RadixTree	5✔
5	from .zone import DNSZone	5✔
6
7
8	class DNSStore:	5✔
9	zones: RadixTree[DNSZone]	5✔
10
11	__slots__ = ("zones",)	5✔
12
13	def __init__(self) -> None:	5✔
14	self.zones = RadixTree()	5✔
15
16	def add_zone(self, zone: DNSZone) -> None:	5✔
17	zone_tuple = self.get_reverse_tuple(zone.name)	5✔
18	if self.zones.search(zone_tuple):	5✔
19	raise ValueError(f"Zone {zone.name} already exists.")	5✔
20	self.zones.insert(zone_tuple, zone)	5✔
21
22	def remove_zone(self, zone_name: str) -> None:	5✔
23	zone_tuple = self.get_reverse_tuple(zone_name)	5✔
24	if not self.zones.search(zone_tuple):	5✔
25	raise ValueError(	5✔
26	f"Zone {zone_name} does not exist.",
27	)
28	# Clear zone from RadixTree
29	self.zones.insert(zone_tuple, None)	5✔
30
31	def get_zone(self, zone_name: str) -> Optional[DNSZone]:	5✔
32	zone_tuple = self.get_reverse_tuple(zone_name)	5✔
33	return self.zones.search(zone_tuple)	5✔
34
35	def query(self, name: str, record_type: RecordType) -> Sequence[DNSRecord]:	5✔
36	if not name.endswith("."):	5!
NEW 37	name += "."	×
38	zone_tuple = self.get_zone_for_name(name)	5✔
39	if not zone_tuple:	5!
NEW 40	return ()	×
41	zone = self.zones.search(zone_tuple)	5✔
42	return zone.get_records(name, record_type) if zone is not None else ()	5✔
43
44	def get_zone_for_name(self, name: str) -> Optional[Tuple[str, ...]]:	5✔
45	labels = self.get_reverse_tuple(name)	5✔
46	result = self.zones.find_prefix(labels)	5✔
47	return result[0] if result else None	5✔
48
49	@staticmethod	5✔
50	def get_reverse_tuple(zone_name: str) -> Tuple[str, ...]:	5✔
51	return tuple(zone_name.strip(".").split("."))[::-1]	5✔

aiokitchen / aiomisc / 9357576700

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous