Skip to content

Fix code quality issues in test_cluster.py #628

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Copilot wants to merge 3 commits into
base: master
Choose a base branch
from
Draft

Fix code quality issues in test_cluster.py #628

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
183f20a
Initial plan
Copilot Dec 23, 2025
38e919a
Fix code quality issues in test_cluster.py
Copilot Dec 23, 2025
79fe231
Fix additional '== None' comparison for consistency
Copilot Dec 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions tests/unit/test_cluster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,10 @@ class ClusterTest(unittest.TestCase):

def test_tuple_for_contact_points(self):
cluster = Cluster(contact_points=[('localhost', 9045), ('127.0.0.2', 9046), '127.0.0.3'], port=9999)
localhost_addr = set([addr[0] for addr in [t for (_,_,_,_,t) in socket.getaddrinfo("localhost",80)]])
# Refactored for clarity
Copy link

@mykaul mykaul Dec 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't recall asking for this change?

addr_info = socket.getaddrinfo("localhost", 80)
sockaddr_tuples = [info[4] for info in addr_info] # info[4] is sockaddr
localhost_addr = set([sockaddr[0] for sockaddr in sockaddr_tuples])
for cp in cluster.endpoints_resolved:
if cp.address in localhost_addr:
assert cp.port == 9045
Expand All @@ -107,7 +110,7 @@ def test_invalid_contact_point_types(self):
Cluster(contact_points="not a sequence", protocol_version=4, connect_timeout=1)

def test_port_str(self):
"""Check port passed as tring is converted and checked properly"""
"""Check port passed as string is converted and checked properly"""
cluster = Cluster(contact_points=['127.0.0.1'], port='1111')
for cp in cluster.endpoints_resolved:
if cp.address in ('::1', '127.0.0.1'):
Expand Down Expand Up @@ -181,7 +184,7 @@ def test_event_delay_timing(self, *_):
"""
sched = _Scheduler(None)
sched.schedule(0, lambda: None)
sched.schedule(0, lambda: None) # pre-473: "TypeError: unorderable types: function() < function()"t
sched.schedule(0, lambda: None) # pre-473: "TypeError: unorderable types: function() < function()"


class SessionTest(unittest.TestCase):
Expand Down Expand Up @@ -291,8 +294,8 @@ def test_default_exec_parameters(self):
assert cluster.profile_manager.default.request_timeout == 10.0
assert session.default_consistency_level == ConsistencyLevel.LOCAL_ONE
assert cluster.profile_manager.default.consistency_level == ConsistencyLevel.LOCAL_ONE
assert session.default_serial_consistency_level == None
assert cluster.profile_manager.default.serial_consistency_level == None
assert session.default_serial_consistency_level is None
assert cluster.profile_manager.default.serial_consistency_level is None
assert session.row_factory == named_tuple_factory
assert cluster.profile_manager.default.row_factory == named_tuple_factory

Expand Down