dohertj2/mxaccessgw
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add bulk read/write CLI subcommands and e2e matrix coverage

Browse Source 
The previous commit added the bulk read/write library surface in every
client; this commit makes that surface reachable from each client's CLI
and exercises it through scripts/run-client-e2e-tests.ps1.

Five new subcommands in every client CLI (.NET / Go / Rust / Python /
Java): read-bulk, write-bulk, write2-bulk, write-secured-bulk, and
write-secured2-bulk. Each follows the existing subscribe-bulk shape:

  - read-bulk takes --server-handle, --items <csv tag list>, and
    --timeout-ms (0 = worker default). JSON output carries the
    BulkReadResult fields, including was_cached so the e2e matrix can
    verify the cached-path semantics.
  - The four bulk-write families take --server-handle, --item-handles
    <csv>, --type, --values <csv>. write2-bulk and write-secured2-bulk
    add a single --timestamp applied to every entry; the secured
    variants take --current-user-id and --verifier-user-id. All four
    output BulkWriteResult JSON.

A new -SkipReadWriteBulk switch on the matrix script (default OFF)
controls two new e2e phases:

  - After the existing subscribe-bulk phase leaves tags advised, the
    script runs read-bulk against the same tag list and asserts most
    results return was_cached = true. This is the only e2e coverage of
    the cache-then-snapshot fork — the unit + gateway tests verify the
    semantics with a fake worker, but only the live cross-language
    matrix proves the cache populates from real OnDataChange events and
    survives the round-trip through every client''s JSON parser.
  - When -VerifyWrite is set, the write phase now also runs a single-
    entry write-bulk against the same writable item handle (using a
    distinct sentinel value) and asserts a per-entry success. Confirms
    the BulkWriteResult wire format end-to-end without complicating
    the OnWriteComplete echo assertion the single-item phase already
    verifies.

Dry-run validation passes for all five clients: each emits the correct
read-bulk and write-bulk CLI invocations with the right flags.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-05-20 04:06:14 -04:00
parent 5e375f6d3d
commit f220908f3f
6 changed files with 1411 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/python/src/mxgateway_cli/commands.py
+198
View File
@@ -19,6 +19,7 @@ from mxgateway.errors import MxGatewayError
from mxgateway.generated import mxaccess_gateway_pb2 as pb
from mxgateway.options import ClientOptions
from mxgateway.session import Session
from mxgateway.values import to_mx_value
from mxgateway.values import MxValueInput
MAX_AGGREGATE_EVENTS = 10_000
@@ -186,6 +187,89 @@ def unsubscribe_bulk(**kwargs: Any) -> None:
)
@main.command("read-bulk")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
@click.option("--items", required=True, help="Comma-separated MXAccess tag addresses.")
@click.option("--timeout-ms", default=0, type=int, show_default=True,
help="Per-tag snapshot timeout in milliseconds. 0 = worker default.")
@click.option("--correlation-id", default="", help="Client correlation id.")
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def read_bulk(**kwargs: Any) -> None:
"""Invoke MXAccess ReadBulk — cached value when advised, snapshot otherwise."""
_run(_read_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
@main.command("write-bulk")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
@click.option("--type", "value_type", default="string", show_default=True)
@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
@click.option("--user-id", default=0, type=int, show_default=True)
@click.option("--correlation-id", default="", help="Client correlation id.")
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def write_bulk(**kwargs: Any) -> None:
"""Invoke MXAccess WriteBulk — sequential Write per entry."""
_run(_write_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
@main.command("write2-bulk")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
@click.option("--type", "value_type", default="string", show_default=True)
@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
@click.option("--timestamp", required=True, help="ISO-8601 timestamp shared across all entries.")
@click.option("--user-id", default=0, type=int, show_default=True)
@click.option("--correlation-id", default="", help="Client correlation id.")
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def write2_bulk(**kwargs: Any) -> None:
"""Invoke MXAccess Write2Bulk — timestamped sequential Write2 per entry."""
_run(_write2_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
@main.command("write-secured-bulk")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
@click.option("--type", "value_type", default="string", show_default=True)
@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
@click.option("--current-user-id", default=0, type=int, show_default=True)
@click.option("--verifier-user-id", default=0, type=int, show_default=True)
@click.option("--correlation-id", default="", help="Client correlation id.")
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def write_secured_bulk(**kwargs: Any) -> None:
"""Invoke MXAccess WriteSecuredBulk — credential-sensitive."""
_run(_write_secured_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
@main.command("write-secured2-bulk")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
@click.option("--type", "value_type", default="string", show_default=True)
@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
@click.option("--timestamp", required=True, help="ISO-8601 timestamp shared across all entries.")
@click.option("--current-user-id", default=0, type=int, show_default=True)
@click.option("--verifier-user-id", default=0, type=int, show_default=True)
@click.option("--correlation-id", default="", help="Client correlation id.")
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
def write_secured2_bulk(**kwargs: Any) -> None:
"""Invoke MXAccess WriteSecured2Bulk — timestamped + credential-sensitive."""
_run(_write_secured2_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
@main.command("stream-events")
@gateway_options
@click.option("--session-id", required=True, help="Gateway session id.")
@@ -340,6 +424,120 @@ async def _unsubscribe_bulk(**kwargs: Any) -> dict[str, Any]:
return {"results": [_message_dict(result) for result in results]}
async def _read_bulk(**kwargs: Any) -> dict[str, Any]:
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])
results = await session.read_bulk(
kwargs["server_handle"],
_parse_string_list(kwargs["items"]),
timeout_ms=kwargs["timeout_ms"],
correlation_id=kwargs["correlation_id"],
)
return {"results": [_message_dict(result) for result in results]}
def _build_write_bulk_entries(kwargs: dict[str, Any]):
"""Build (item_handle, MxValue) pairs for the bulk-write families.
The CLI accepts a single ``--type`` plus ``--values`` (comma-separated
string-encoded values, one per ``--item-handles`` entry). Returns the
parsed item-handle list and the per-entry MxValue protobuf instances —
callers wrap these into the appropriate per-entry message type.
"""
handles = _parse_int_list(kwargs["item_handles"])
value_texts = _parse_string_list(kwargs["values"])
if len(handles) != len(value_texts):
raise click.UsageError(
f"item-handles count ({len(handles)}) does not match values count ({len(value_texts)})",
)
parsed = [_parse_value(text, kwargs["value_type"]) for text in value_texts]
values = [to_mx_value(v) for v in parsed]
return handles, values
async def _write_bulk(**kwargs: Any) -> dict[str, Any]:
handles, values = _build_write_bulk_entries(kwargs)
entries = [
pb.WriteBulkEntry(item_handle=handle, user_id=kwargs["user_id"], value=value)
for handle, value in zip(handles, values)
]
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])
results = await session.write_bulk(
kwargs["server_handle"],
entries,
correlation_id=kwargs["correlation_id"],
)
return {"results": [_message_dict(result) for result in results]}
async def _write2_bulk(**kwargs: Any) -> dict[str, Any]:
handles, values = _build_write_bulk_entries(kwargs)
timestamp_value = to_mx_value(_parse_datetime(kwargs["timestamp"]))
entries = [
pb.Write2BulkEntry(
item_handle=handle,
user_id=kwargs["user_id"],
value=value,
timestamp_value=timestamp_value,
)
for handle, value in zip(handles, values)
]
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])
results = await session.write2_bulk(
kwargs["server_handle"],
entries,
correlation_id=kwargs["correlation_id"],
)
return {"results": [_message_dict(result) for result in results]}
async def _write_secured_bulk(**kwargs: Any) -> dict[str, Any]:
handles, values = _build_write_bulk_entries(kwargs)
entries = [
pb.WriteSecuredBulkEntry(
item_handle=handle,
current_user_id=kwargs["current_user_id"],
verifier_user_id=kwargs["verifier_user_id"],
value=value,
)
for handle, value in zip(handles, values)
]
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])
results = await session.write_secured_bulk(
kwargs["server_handle"],
entries,
correlation_id=kwargs["correlation_id"],
)
return {"results": [_message_dict(result) for result in results]}
async def _write_secured2_bulk(**kwargs: Any) -> dict[str, Any]:
handles, values = _build_write_bulk_entries(kwargs)
timestamp_value = to_mx_value(_parse_datetime(kwargs["timestamp"]))
entries = [
pb.WriteSecured2BulkEntry(
item_handle=handle,
current_user_id=kwargs["current_user_id"],
verifier_user_id=kwargs["verifier_user_id"],
value=value,
timestamp_value=timestamp_value,
)
for handle, value in zip(handles, values)
]
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])
results = await session.write_secured2_bulk(
kwargs["server_handle"],
entries,
correlation_id=kwargs["correlation_id"],
)
return {"results": [_message_dict(result) for result in results]}
async def _stream_events(**kwargs: Any) -> dict[str, Any]:
async with await _connect(kwargs) as client:
session = _session(client, kwargs["session_id"])