kovagoadiandClaude
25953bea06
Resolve 51 findings from the I/RUF/BLE/TRY002/S110/PLR0402 rule set: - Sort imports and __all__ (I001, RUF022, PLR0402). The sys.path.insert calls in check_public_calls.py and tests/test_e2e.py still precede the openttd imports that depend on them. - Replace unused unpacked values with _ (RUF059) and annotate the two timetable lookup tables as ClassVar (RUF012). - Narrow the best-effort excepts in OpenTTDClient.quit and OpenTTDAdminClient.quit to (OSError, SocketClosed) and log at debug rather than swallowing silently (BLE001, S110). The test doubles now raise an OSError subclass so they still exercise that branch. - Narrow the gamescript JSON fallback to json.JSONDecodeError. The broad catch in receive_packet keeps a noqa: it guards untrusted wire data and must degrade to a no-op packet instead of killing the connection. - Use contextlib.suppress instead of try/except/pass in tests. ruff check . is clean, 102 tests pass, coverage stays at 100%. Co-Authored-By: Claude <[email protected]>
113 lines
4.2 KiB
Python
Executable File
113 lines
4.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import ast
|
|
import inspect
|
|
import os
|
|
import sys
|
|
|
|
# Add lib and tests to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'lib'))
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from openttd import OpenTTDAdminClient, OpenTTDClient
|
|
from openttd.protocol import OpenTTDAdminProtocol, OpenTTDProtocol
|
|
|
|
from tests import test_e2e
|
|
|
|
# 1. Gather public functions dynamically at runtime using reflection
|
|
classes = [OpenTTDClient, OpenTTDAdminClient, OpenTTDProtocol, OpenTTDAdminProtocol]
|
|
public_funcs = {} # name -> has_params
|
|
|
|
# Standard asyncio/lifecycle protocol methods that are not user-facing APIs
|
|
auto_exclude = {
|
|
"connection_made", "connection_lost", "data_received",
|
|
"eof_received", "pause_writing", "resume_writing", "connected"
|
|
}
|
|
|
|
for cls in classes:
|
|
# Class constructor call (e.g. OpenTTDClient(host, ...))
|
|
sig_init = inspect.signature(cls.__init__)
|
|
has_params_init = len([p for name, p in sig_init.parameters.items() if name != 'self']) > 0
|
|
public_funcs[cls.__name__] = has_params_init
|
|
|
|
# Get all functions on the class (including inherited)
|
|
for name, val in inspect.getmembers(cls):
|
|
if name == '__init__':
|
|
continue
|
|
if not (inspect.isfunction(val) or inspect.ismethod(val) or inspect.iscoroutinefunction(val)):
|
|
continue
|
|
# Exclude private methods
|
|
if name.startswith('_'):
|
|
continue
|
|
# Exclude protocol dispatch targets (receive_* except receive_packet)
|
|
if name.startswith('receive_') and name != 'receive_packet':
|
|
continue
|
|
# Exclude lifecycle callbacks and manually decorated ones
|
|
if name in auto_exclude or getattr(val, '__exclude_call_check__', False):
|
|
continue
|
|
|
|
sig = inspect.signature(val)
|
|
has_params = len([p for p_name, p in sig.parameters.items() if p_name != 'self']) > 0
|
|
public_funcs[name] = has_params
|
|
|
|
# 2. Inspect E2E test module dynamically at runtime using reflection
|
|
|
|
calls = {name: [] for name in public_funcs}
|
|
|
|
class E2ECallVisitor(ast.NodeVisitor):
|
|
def visit_Call(self, node):
|
|
method_name = None
|
|
if isinstance(node.func, ast.Attribute):
|
|
method_name = node.func.attr
|
|
elif isinstance(node.func, ast.Name):
|
|
method_name = node.func.id
|
|
|
|
if method_name in public_funcs:
|
|
# Serialize arguments to string to compare data
|
|
args_str = [ast.unparse(a) for a in node.args]
|
|
kwargs_str = [f"{kw.arg}={ast.unparse(kw.value)}" for kw in node.keywords]
|
|
call_data = (tuple(args_str), tuple(sorted(kwargs_str)))
|
|
calls[method_name].append(call_data)
|
|
|
|
self.generic_visit(node)
|
|
|
|
visitor = E2ECallVisitor()
|
|
|
|
# Reflectively iterate through all test functions defined inside tests.test_e2e
|
|
for name, val in inspect.getmembers(test_e2e, predicate=inspect.isfunction):
|
|
# Only analyze functions defined directly in the module (skips imported ones)
|
|
if inspect.getmodule(val) == test_e2e:
|
|
# Retrieve function source dynamically via reflection
|
|
source = inspect.getsource(val)
|
|
func_tree = ast.parse(source)
|
|
visitor.visit(func_tree)
|
|
|
|
# 3. Perform assertion checks
|
|
failed = False
|
|
print("=== Public Function E2E Calls Verification (Reflection API Inspection) ===")
|
|
for name, has_params in sorted(public_funcs.items()):
|
|
func_calls = calls[name]
|
|
num_calls = len(func_calls)
|
|
|
|
if num_calls < 2:
|
|
print(f"❌ {name}: Called {num_calls} time(s) (expected at least 2).")
|
|
failed = True
|
|
continue
|
|
|
|
if has_params:
|
|
unique_calls = set(func_calls)
|
|
num_unique = len(unique_calls)
|
|
if num_unique < 2:
|
|
print(f"❌ {name}: Called {num_calls} times but with identical data: {unique_calls}")
|
|
failed = True
|
|
else:
|
|
print(f"✅ {name}: Called {num_calls} times with {num_unique} different inputs.")
|
|
else:
|
|
print(f"✅ {name}: Called {num_calls} times (no parameters).")
|
|
|
|
if failed:
|
|
print("❌ Verification FAILED: Some public functions do not meet E2E call requirement.")
|
|
sys.exit(1)
|
|
else:
|
|
print("✅ Verification PASSED: All public functions called at least 2 times with different data.")
|
|
sys.exit(0)
|