|
32 | 32 | _PIPE_CMD = f"{sys.executable} {_DESCRIBE_WORKER}" |
33 | 33 |
|
34 | 34 | _DESCRIBE_HTTP_WORKER = str(Path(__file__).parent / "serve_fixture_describe_http.py") |
35 | | -_DESCRIBE_UNIX_WORKER = str(Path(__file__).parent / "serve_fixture_describe_unix.py") |
36 | | -_DESCRIBE_UNIX_THREADED_WORKER = str(Path(__file__).parent / "serve_fixture_describe_unix_threaded.py") |
37 | 35 |
|
38 | 36 |
|
39 | 37 | # --------------------------------------------------------------------------- |
@@ -63,74 +61,23 @@ def describe_http_port() -> Iterator[int]: |
63 | 61 | proc.wait(timeout=5) |
64 | 62 |
|
65 | 63 |
|
66 | | -@pytest.fixture(scope="module") |
67 | | -def describe_unix_path() -> Iterator[str]: |
68 | | - """Spawn a Unix socket server with describe enabled for the test module.""" |
69 | | - from tests.conftest import _short_unix_path, _wait_for_unix |
70 | | - |
71 | | - path = _short_unix_path("cli-fix") |
72 | | - proc = subprocess.Popen( |
73 | | - [sys.executable, _DESCRIBE_UNIX_WORKER, path], |
74 | | - stdout=subprocess.PIPE, |
75 | | - ) |
76 | | - try: |
77 | | - assert proc.stdout is not None |
78 | | - line = proc.stdout.readline().decode().strip() |
79 | | - assert line == f"UNIX:{path}", f"Expected UNIX:{path}, got: {line!r}" |
80 | | - _wait_for_unix(path) |
81 | | - yield path |
82 | | - finally: |
83 | | - proc.terminate() |
84 | | - proc.wait(timeout=5) |
85 | | - |
86 | | - |
87 | | -@pytest.fixture(scope="module") |
88 | | -def describe_unix_threaded_path() -> Iterator[str]: |
89 | | - """Spawn a threaded Unix socket server with describe enabled for the test module.""" |
90 | | - from tests.conftest import _short_unix_path, _wait_for_unix |
91 | | - |
92 | | - path = _short_unix_path("cli-ft") |
93 | | - proc = subprocess.Popen( |
94 | | - [sys.executable, _DESCRIBE_UNIX_THREADED_WORKER, path], |
95 | | - stdout=subprocess.PIPE, |
96 | | - ) |
97 | | - try: |
98 | | - assert proc.stdout is not None |
99 | | - line = proc.stdout.readline().decode().strip() |
100 | | - assert line == f"UNIX:{path}", f"Expected UNIX:{path}, got: {line!r}" |
101 | | - _wait_for_unix(path) |
102 | | - yield path |
103 | | - finally: |
104 | | - proc.terminate() |
105 | | - proc.wait(timeout=5) |
106 | | - |
107 | 64 |
|
108 | 65 | # --------------------------------------------------------------------------- |
109 | 66 | # Parametrized transport fixture |
110 | 67 | # --------------------------------------------------------------------------- |
111 | 68 |
|
112 | 69 |
|
113 | | -_SKIP_UNIX = pytest.mark.skipif(sys.platform == "win32", reason="Unix sockets not available on Windows") |
114 | | - |
115 | 70 |
|
116 | 71 | @pytest.fixture( |
117 | 72 | params=[ |
118 | 73 | "pipe", |
119 | 74 | "http", |
120 | | - pytest.param("unix", marks=_SKIP_UNIX), |
121 | | - pytest.param("unix_threaded", marks=_SKIP_UNIX), |
122 | 75 | ] |
123 | 76 | ) |
124 | 77 | def transport_args(request: pytest.FixtureRequest) -> list[str]: |
125 | | - """Return CLI args that select a transport (``--cmd …``, ``--url …``, or ``--unix …``).""" |
| 78 | + """Return CLI args that select a transport (``--cmd …`` or ``--url …``).""" |
126 | 79 | if request.param == "pipe": |
127 | 80 | return ["--cmd", _PIPE_CMD] |
128 | | - elif request.param == "unix": |
129 | | - path: str = request.getfixturevalue("describe_unix_path") |
130 | | - return ["--unix", path] |
131 | | - elif request.param == "unix_threaded": |
132 | | - path = request.getfixturevalue("describe_unix_threaded_path") |
133 | | - return ["--unix", path] |
134 | 81 | port: int = request.getfixturevalue("describe_http_port") |
135 | 82 | return ["--url", f"http://127.0.0.1:{port}"] |
136 | 83 |
|
@@ -402,6 +349,78 @@ def test_format_table(self) -> None: |
402 | 349 | assert "---" in result.output |
403 | 350 |
|
404 | 351 |
|
| 352 | +# --------------------------------------------------------------------------- |
| 353 | +# Coercion edge-case tests (pipe transport, covers _coerce_value bool branch |
| 354 | +# and _coerce_map_value dict-from-tuples branch) |
| 355 | +# --------------------------------------------------------------------------- |
| 356 | + |
| 357 | + |
| 358 | +class TestCoercionEdgeCases: |
| 359 | + """Tests for CLI value coercion helpers.""" |
| 360 | + |
| 361 | + def test_coerce_boolean(self) -> None: |
| 362 | + """Boolean key=value coercion via _coerce_value.""" |
| 363 | + from vgi_rpc.cli import _coerce_value |
| 364 | + |
| 365 | + assert _coerce_value("true", pa.bool_()) is True |
| 366 | + assert _coerce_value("false", pa.bool_()) is False |
| 367 | + assert _coerce_value("1", pa.bool_()) is True |
| 368 | + assert _coerce_value("yes", pa.bool_()) is True |
| 369 | + assert _coerce_value("0", pa.bool_()) is False |
| 370 | + |
| 371 | + def test_map_column_dict_output(self) -> None: |
| 372 | + """Dict-returning method renders map columns as dicts in JSON output.""" |
| 373 | + result = _invoke( |
| 374 | + ["--cmd", _PIPE_CMD, "--format", "json", "call", "echo_mapping", "--json", '{"mapping": {"x": 1, "y": 2}}'], |
| 375 | + ) |
| 376 | + assert result.exit_code == 0, f"Failed: {result.output}" |
| 377 | + data = json.loads(result.output) |
| 378 | + assert data["result"] == {"x": 1, "y": 2} |
| 379 | + |
| 380 | + |
| 381 | +# --------------------------------------------------------------------------- |
| 382 | +# Unix socket smoke test (covers _introspect_unix code path) |
| 383 | +# --------------------------------------------------------------------------- |
| 384 | + |
| 385 | + |
| 386 | +_DESCRIBE_UNIX_WORKER = str(Path(__file__).parent / "serve_fixture_describe_unix.py") |
| 387 | + |
| 388 | +_SKIP_UNIX = pytest.mark.skipif(sys.platform == "win32", reason="Unix sockets not available on Windows") |
| 389 | + |
| 390 | + |
| 391 | +@pytest.fixture(scope="module") |
| 392 | +def describe_unix_path() -> Iterator[str]: |
| 393 | + """Spawn a Unix socket server with describe enabled for the test module.""" |
| 394 | + from tests.conftest import _short_unix_path, _wait_for_unix |
| 395 | + |
| 396 | + path = _short_unix_path("cli-fix") |
| 397 | + proc = subprocess.Popen( |
| 398 | + [sys.executable, _DESCRIBE_UNIX_WORKER, path], |
| 399 | + stdout=subprocess.PIPE, |
| 400 | + ) |
| 401 | + try: |
| 402 | + assert proc.stdout is not None |
| 403 | + line = proc.stdout.readline().decode().strip() |
| 404 | + assert line == f"UNIX:{path}", f"Expected UNIX:{path}, got: {line!r}" |
| 405 | + _wait_for_unix(path) |
| 406 | + yield path |
| 407 | + finally: |
| 408 | + proc.terminate() |
| 409 | + proc.wait(timeout=5) |
| 410 | + |
| 411 | + |
| 412 | +class TestUnixTransport: |
| 413 | + """Smoke test for --unix CLI transport.""" |
| 414 | + |
| 415 | + @_SKIP_UNIX |
| 416 | + def test_describe_unix(self, describe_unix_path: str) -> None: |
| 417 | + """Describe via --unix exercises _introspect_unix.""" |
| 418 | + result = _invoke(["--unix", describe_unix_path, "--format", "json", "describe"]) |
| 419 | + assert result.exit_code == 0, f"Failed: {result.output}" |
| 420 | + data = json.loads(result.output) |
| 421 | + assert data["protocol_name"] == "RpcFixtureService" |
| 422 | + |
| 423 | + |
405 | 424 | # --------------------------------------------------------------------------- |
406 | 425 | # Logging tests |
407 | 426 | # --------------------------------------------------------------------------- |
|
0 commit comments