19256588545

Committed 11 Nov 2025 06:01AM UTC coverage: 93.425% (-0.04%) from 93.462%

Build # 19256588545

Build Type

Pull #743

github

Committed by

dbrattli

Commit Message

Test fixes

Pull Request Pull Request #743: RxPY v5

Run Details

3390 of 3626 new or added lines in 119 files covered. (93.49%)

13 existing lines in 3 files now uncovered.

25007 of 26767 relevant lines covered (93.42%)

0.93 hits per line

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

94.44

/tests/test_core/test_priorityqueue.py

import unittest

from reactivex.internal import PriorityQueue


class TestItem:
    __test__ = False

    def __init__(self, value: int, label: str | None = None) -> None:
        self.value = value
        self.label = label

    def __str__(self) -> str:
        if self.label:
            return f"{self.value} ({self.label})"
        else:
            return f"{self.value}"

    def __repr__(self) -> str:
        return str(self)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, TestItem):
            return NotImplemented
        return self.value == other.value  # and self.label == other.label

    def __lt__(self, other: "TestItem") -> bool:
        return self.value < other.value

    def __gt__(self, other: "TestItem") -> bool:
        return self.value > other.value


class TestPriorityQueue(unittest.TestCase):
    def test_priorityqueue_count(self) -> None:
        assert PriorityQueue.MIN_COUNT < 0

    def test_priorityqueue_empty(self) -> None:
        """Must be empty on construction"""

        p: PriorityQueue[int] = PriorityQueue()
        assert len(p) == 0
        assert p.items == []

        # Still empty after enqueue/dequeue
        p.enqueue(42)
        p.dequeue()
        assert len(p) == 0

    def test_priorityqueue_length(self) -> None:
        """Test that length is n after n invocations"""

        p: PriorityQueue[int] = PriorityQueue()

        assert len(p) == 0
        for n in range(42):
            p.enqueue(n)
        assert len(p) == 42
        p.dequeue()
        assert len(p) == 41
        p.remove(10)
        assert len(p) == 40
        for n in range(len(p)):
            p.dequeue()
        assert len(p) == 0

    def test_priorityqueue_enqueue_dequeue(self) -> None:
        """Enqueue followed by dequeue should give the same result"""

        p: PriorityQueue[int] = PriorityQueue()
        self.assertRaises(IndexError, p.dequeue)

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)

        assert [p.dequeue(), p.dequeue(), p.dequeue()] == [41, 42, 43]

    def test_priorityqueue_sort_stability(self) -> None:
        """Items with same value should be returned in the order they were
        added"""

        p: PriorityQueue[TestItem] = PriorityQueue()

        p.enqueue(TestItem(43, "high"))
        p.enqueue(TestItem(42, "first"))
        p.enqueue(TestItem(42, "second"))
        p.enqueue(TestItem(42, "last"))
        p.enqueue(TestItem(41, "low"))

        assert len(p) == 5

        assert p.dequeue() == TestItem(41, "low")
        assert p.dequeue() == TestItem(42, "first")
        assert p.dequeue() == TestItem(42, "second")
        assert p.dequeue() == TestItem(42, "last")
        assert p.dequeue() == TestItem(43, "high")

    def test_priorityqueue_remove(self) -> None:
        """Remove item from queue"""

        p: PriorityQueue[int] = PriorityQueue()
        assert not p.remove(42)

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(42)
        assert [p.dequeue(), p.dequeue()] == [41, 43]

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(41)
        assert [p.dequeue(), p.dequeue()] == [42, 43]

        p.enqueue(42)
        p.enqueue(41)
        p.enqueue(43)
        assert p.remove(43)
        assert [p.dequeue(), p.dequeue()] == [41, 42]

    def test_priorityqueue_peek(self) -> None:
        """Peek at first element in queue"""

        p: PriorityQueue[int] = PriorityQueue()

        self.assertRaises(IndexError, p.peek)
        p.enqueue(42)
        assert p.peek() == 42
        p.enqueue(41)
        assert p.peek() == 41
        p.enqueue(43)
        assert p.peek() == 41

1	import unittest	1✔
2
3	from reactivex.internal import PriorityQueue	1✔
4
5
6	class TestItem:	1✔
7	__test__ = False	1✔
8
9	def __init__(self, value: int, label: str \| None = None) -> None:	1✔
10	self.value = value	1✔
11	self.label = label	1✔
12
13	def __str__(self) -> str:	1✔
14	if self.label:	×
NEW 15	return f"{self.value} ({self.label})"	×
16	else:
NEW 17	return f"{self.value}"	×
18
19	def __repr__(self) -> str:	1✔
20	return str(self)	×
21
22	def __eq__(self, other: object) -> bool:	1✔
23	if not isinstance(other, TestItem):	1✔
24	return NotImplemented
25	return self.value == other.value # and self.label == other.label	1✔
26
27	def __lt__(self, other: "TestItem") -> bool:	1✔
28	return self.value < other.value	1✔
29
30	def __gt__(self, other: "TestItem") -> bool:	1✔
31	return self.value > other.value	×
32
33
34	class TestPriorityQueue(unittest.TestCase):	1✔
35	def test_priorityqueue_count(self) -> None:	1✔
36	assert PriorityQueue.MIN_COUNT < 0	1✔
37
38	def test_priorityqueue_empty(self) -> None:	1✔
39	"""Must be empty on construction"""
40
41	p: PriorityQueue[int] = PriorityQueue()	1✔
42	assert len(p) == 0	1✔
43	assert p.items == []	1✔
44
45	# Still empty after enqueue/dequeue
46	p.enqueue(42)	1✔
47	p.dequeue()	1✔
48	assert len(p) == 0	1✔
49
50	def test_priorityqueue_length(self) -> None:	1✔
51	"""Test that length is n after n invocations"""
52
53	p: PriorityQueue[int] = PriorityQueue()	1✔
54
55	assert len(p) == 0	1✔
56	for n in range(42):	1✔
57	p.enqueue(n)	1✔
58	assert len(p) == 42	1✔
59	p.dequeue()	1✔
60	assert len(p) == 41	1✔
61	p.remove(10)	1✔
62	assert len(p) == 40	1✔
63	for n in range(len(p)):	1✔
64	p.dequeue()	1✔
65	assert len(p) == 0	1✔
66
67	def test_priorityqueue_enqueue_dequeue(self) -> None:	1✔
68	"""Enqueue followed by dequeue should give the same result"""
69
70	p: PriorityQueue[int] = PriorityQueue()	1✔
71	self.assertRaises(IndexError, p.dequeue)	1✔
72
73	p.enqueue(42)	1✔
74	p.enqueue(41)	1✔
75	p.enqueue(43)	1✔
76
77	assert [p.dequeue(), p.dequeue(), p.dequeue()] == [41, 42, 43]	1✔
78
79	def test_priorityqueue_sort_stability(self) -> None:	1✔
80	"""Items with same value should be returned in the order they were
81	added"""
82
83	p: PriorityQueue[TestItem] = PriorityQueue()	1✔
84
85	p.enqueue(TestItem(43, "high"))	1✔
86	p.enqueue(TestItem(42, "first"))	1✔
87	p.enqueue(TestItem(42, "second"))	1✔
88	p.enqueue(TestItem(42, "last"))	1✔
89	p.enqueue(TestItem(41, "low"))	1✔
90
91	assert len(p) == 5	1✔
92
93	assert p.dequeue() == TestItem(41, "low")	1✔
94	assert p.dequeue() == TestItem(42, "first")	1✔
95	assert p.dequeue() == TestItem(42, "second")	1✔
96	assert p.dequeue() == TestItem(42, "last")	1✔
97	assert p.dequeue() == TestItem(43, "high")	1✔
98
99	def test_priorityqueue_remove(self) -> None:	1✔
100	"""Remove item from queue"""
101
102	p: PriorityQueue[int] = PriorityQueue()	1✔
103	assert not p.remove(42)	1✔
104
105	p.enqueue(42)	1✔
106	p.enqueue(41)	1✔
107	p.enqueue(43)	1✔
108	assert p.remove(42)	1✔
109	assert [p.dequeue(), p.dequeue()] == [41, 43]	1✔
110
111	p.enqueue(42)	1✔
112	p.enqueue(41)	1✔
113	p.enqueue(43)	1✔
114	assert p.remove(41)	1✔
115	assert [p.dequeue(), p.dequeue()] == [42, 43]	1✔
116
117	p.enqueue(42)	1✔
118	p.enqueue(41)	1✔
119	p.enqueue(43)	1✔
120	assert p.remove(43)	1✔
121	assert [p.dequeue(), p.dequeue()] == [41, 42]	1✔
122
123	def test_priorityqueue_peek(self) -> None:	1✔
124	"""Peek at first element in queue"""
125
126	p: PriorityQueue[int] = PriorityQueue()	1✔
127
128	self.assertRaises(IndexError, p.peek)	1✔
129	p.enqueue(42)	1✔
130	assert p.peek() == 42	1✔
131	p.enqueue(41)	1✔
132	assert p.peek() == 41	1✔
133	p.enqueue(43)	1✔
134	assert p.peek() == 41	1✔

ReactiveX / RxPY / 19256588545

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